{"action":"created","issue":{"url":"https://api.github.com/repos/apache/flink/issues/6082","repository_url":"https://api.github.com/repos/apache/flink","labels_url":"https://api.github.com/repos/apache/flink/issues/6082/labels{/name}","comments_url":"https://api.github.com/repos/apache/flink/issues/6082/comments","events_url":"https://api.github.com/repos/apache/flink/issues/6082/events","html_url":"https://github.com/apache/flink/pull/6082","id":326735926,"number":6082,"title":"[FLINK-9444] KafkaAvroTableSource failed to work for map fields","user":{"login":"tragicjun","id":3949293,"avatar_url":"https://avatars2.githubusercontent.com/u/3949293?v=4","gravatar_id":"","url":"https://api.github.com/users/tragicjun","html_url":"https://github.com/tragicjun","followers_url":"https://api.github.com/users/tragicjun/followers","following_url":"https://api.github.com/users/tragicjun/following{/other_user}","gists_url":"https://api.github.com/users/tragicjun/gists{/gist_id}","starred_url":"https://api.github.com/users/tragicjun/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tragicjun/subscriptions","organizations_url":"https://api.github.com/users/tragicjun/orgs","repos_url":"https://api.github.com/users/tragicjun/repos","events_url":"https://api.github.com/users/tragicjun/events{/privacy}","received_events_url":"https://api.github.com/users/tragicjun/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":0,"created_at":"2018-05-26T12:50:37Z","updated_at":"2018-05-26T12:57:44Z","closed_at":null,"author_association":"CONTRIBUTOR","pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/6082","html_url":"https://github.com/apache/flink/pull/6082","diff_url":"https://github.com/apache/flink/pull/6082.diff","patch_url":"https://github.com/apache/flink/pull/6082.patch"},"body":"## What is the purpose of the change\r\n\r\nWhen some Avro schema has map fields and the corresponding TableSchema declares **MapTypeInfo** for these fields, an exception will be thrown when registering the **KafkaAvroTableSource**, complaining like:\r\n\r\n```\r\nException in thread \"main\" org.apache.flink.table.api.ValidationException: Type Map<String, String> of table field 'event' does not match with type GenericType<java.util.Map> of the field 'event' of the TableSource return type.\r\nat org.apache.flink.table.api.ValidationException$.apply(exceptions.scala:74)\r\nat org.apache.flink.table.sources.TableSourceUtil$$anonfun$validateTableSource$1.apply(TableSourceUtil.scala:92)\r\nat org.apache.flink.table.sources.TableSourceUtil$$anonfun$validateTableSource$1.apply(TableSourceUtil.scala:71)\r\nat scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)\r\nat scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)\r\nat org.apache.flink.table.sources.TableSourceUtil$.validateTableSource(TableSourceUtil.scala:71)\r\nat org.apache.flink.table.plan.schema.StreamTableSourceTable.<init>(StreamTableSourceTable.scala:33)\r\nat org.apache.flink.table.api.StreamTableEnvironment.registerTableSourceInternal(StreamTableEnvironment.scala:124)\r\nat org.apache.flink.table.api.TableEnvironment.registerTableSource(TableEnvironment.scala:438)\r\n```\r\n\r\nThis pull request adds a new unit test to expose the issue and then fixes it. \r\n\r\n**Note: In this implementation, following Avro primitive value types are supported: string, int, long, float, double and boolean, which should cater for most use cases.**\r\n\r\n## Brief change log\r\n\r\n  - Add a new unit test \"testHasMapFieldsAvroClass()\" in KafkaAvroTableSourceTestBase\r\n  - Add some logic in \"AvroTestUtils.createFlatAvroSchema()\" to create Avro MapSchema\r\n  - Add some logic in \"AvroRecordClassConverter.convertType()\" to convert \"GenericType<java.util.Map>\"  into \"MapTypeInfo\" with matching value types.\r\n\r\n## Verifying this change\r\n\r\nThis change can be verified as follows:\r\n- Run the unit test \"testHasMapFieldsAvroClass()\"  added in KafkaAvroTableSourceTestBase by this fix.\r\n- The unit test would fail with similar exceptions thrown described above.\r\n- Merge this fix and run the unit test again, it should pass \r\n\r\n## Does this pull request potentially affect one of the following parts:\r\n\r\n  - Dependencies (does it add or upgrade a dependency): (no)\r\n  - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)\r\n  - The serializers: ( no)\r\n  - The runtime per-record code paths (performance sensitive): (no)\r\n  - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)\r\n  - The S3 file system connector: (no)\r\n\r\n## Documentation\r\n\r\n  - Does this pull request introduce a new feature? (no)\r\n  - If yes, how is the feature documented? (not applicable)\r\n"},"comment":{"url":"https://api.github.com/repos/apache/flink/issues/comments/392259765","html_url":"https://github.com/apache/flink/pull/6082#issuecomment-392259765","issue_url":"https://api.github.com/repos/apache/flink/issues/6082","id":392259765,"user":{"login":"tragicjun","id":3949293,"avatar_url":"https://avatars2.githubusercontent.com/u/3949293?v=4","gravatar_id":"","url":"https://api.github.com/users/tragicjun","html_url":"https://github.com/tragicjun","followers_url":"https://api.github.com/users/tragicjun/followers","following_url":"https://api.github.com/users/tragicjun/following{/other_user}","gists_url":"https://api.github.com/users/tragicjun/gists{/gist_id}","starred_url":"https://api.github.com/users/tragicjun/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tragicjun/subscriptions","organizations_url":"https://api.github.com/users/tragicjun/orgs","repos_url":"https://api.github.com/users/tragicjun/repos","events_url":"https://api.github.com/users/tragicjun/events{/privacy}","received_events_url":"https://api.github.com/users/tragicjun/received_events","type":"User","site_admin":false},"created_at":"2018-05-26T12:57:44Z","updated_at":"2018-05-26T12:57:44Z","author_association":"CONTRIBUTOR","body":"Please ignore previous commits, only the latest commit is relevant to this issue."}}
{"actor":{"display_login":"tragicjun"}}