{"action":"opened","issue":{"url":"https://api.github.com/repos/confluentinc/kafka-connect-hdfs/issues/97","repository_url":"https://api.github.com/repos/confluentinc/kafka-connect-hdfs","labels_url":"https://api.github.com/repos/confluentinc/kafka-connect-hdfs/issues/97/labels{/name}","comments_url":"https://api.github.com/repos/confluentinc/kafka-connect-hdfs/issues/97/comments","events_url":"https://api.github.com/repos/confluentinc/kafka-connect-hdfs/issues/97/events","html_url":"https://github.com/confluentinc/kafka-connect-hdfs/issues/97","id":168581381,"number":97,"title":"Did not find matching union field for data","user":{"login":"sanchitgrover","id":5622422,"avatar_url":"https://avatars.githubusercontent.com/u/5622422?v=3","gravatar_id":"","url":"https://api.github.com/users/sanchitgrover","html_url":"https://github.com/sanchitgrover","followers_url":"https://api.github.com/users/sanchitgrover/followers","following_url":"https://api.github.com/users/sanchitgrover/following{/other_user}","gists_url":"https://api.github.com/users/sanchitgrover/gists{/gist_id}","starred_url":"https://api.github.com/users/sanchitgrover/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sanchitgrover/subscriptions","organizations_url":"https://api.github.com/users/sanchitgrover/orgs","repos_url":"https://api.github.com/users/sanchitgrover/repos","events_url":"https://api.github.com/users/sanchitgrover/events{/privacy}","received_events_url":"https://api.github.com/users/sanchitgrover/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":0,"created_at":"2016-08-01T07:13:01Z","updated_at":"2016-08-01T07:13:01Z","closed_at":null,"body":"I am trying to write data in avro format from my Java code to Kafka to HDFS using kafka HDFS connector and I am getting some issues. When I use the simple schema and data provided on the confluent platform website, I am able to write data to HDFS, but when I try to use complex avro schema, I get this error in the HDFS connector logs:\r\n\r\n`ERROR Task hdfs-sink-0 threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:142)\r\norg.apache.kafka.connect.errors.DataException: Did not find matching union field for data: PROD\r\n    at io.confluent.connect.avro.AvroData.toConnectData(AvroData.java:973)\r\n    at io.confluent.connect.avro.AvroData.toConnectData(AvroData.java:981)\r\n    at io.confluent.connect.avro.AvroData.toConnectData(AvroData.java:981)\r\n    at io.confluent.connect.avro.AvroData.toConnectData(AvroData.java:981)\r\n    at io.confluent.connect.avro.AvroData.toConnectData(AvroData.java:981)\r\n    at io.confluent.connect.avro.AvroData.toConnectData(AvroData.java:782)\r\n    at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:103)\r\n    at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:346)\r\n    at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:226)\r\n    at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:170)\r\n    at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:142)\r\n    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:140)\r\n    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:175)\r\n    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)\r\n    at java.util.concurrent.FutureTask.run(FutureTask.java:266)\r\n    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)\r\n    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)\r\n    at java.lang.Thread.run(Thread.java:745)`\r\n\r\nI am using confluent platform 3.0.0\r\n\r\nMy Java code:\r\n\r\n`Properties props = new Properties();`\r\n`props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);`\r\n`props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,io.confluent.kafka.serializers.KafkaAvroSerializer.class);`\r\n`props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,io.confluent.kafka.serializers.KafkaAvroSerializer.class);`\r\n`props.put(\"schema.registry.url\", <url>);`\r\n\r\n`KafkaProducer producer = new KafkaProducer(props);`\r\n\r\n`Schema schema = new Schema.Parser().parse(new FileInputStream(\"avsc/schema.avsc\"));`\r\n`DatumReader<Object> reader = new GenericDatumReader<Object>(schema);`\r\n\r\n`InputStream input = new FileInputStream(\"json/data.json\");`\r\n`DataInputStream din = new DataInputStream(input);`\r\n`Decoder decoder = DecoderFactory.get().jsonDecoder(schema, din);`\r\n\r\n`Object datum = null;`\r\n`while (true) {`\r\n`    try {`\r\n`        datum = reader.read(null, decoder);`\r\n`    } catch (EOFException e) {`\r\n`       break;`\r\n`    }`\r\n`}`\r\n\r\n`ProducerRecord<Object, Object> message = new ProducerRecord<Object, Object>(topic, datum);`\r\n`producer.send(message);`\r\n`producer.close();`\r\n\r\nThe schema (this is created from avdl file):\r\n\r\n`{\r\n  \"type\" : \"record\",\r\n  \"name\" : \"RiskMeasureEvent\",\r\n  \"namespace\" : \"risk\",\r\n  \"fields\" : [ {\r\n    \"name\" : \"info\",\r\n    \"type\" : {\r\n      \"type\" : \"record\",\r\n      \"name\" : \"RiskMeasureInfo\",\r\n      \"fields\" : [ {\r\n        \"name\" : \"source\",\r\n        \"type\" : {\r\n          \"type\" : \"record\",\r\n          \"name\" : \"Source\",\r\n          \"fields\" : [ {\r\n            \"name\" : \"app\",\r\n            \"type\" : {\r\n              \"type\" : \"record\",\r\n              \"name\" : \"Application\",\r\n              \"fields\" : [ {\r\n                \"name\" : \"csi_id\",\r\n                \"type\" : \"string\"\r\n              }, {\r\n                \"name\" : \"name\",\r\n                \"type\" : \"string\"\r\n              } ]\r\n            }\r\n          }, {\r\n            \"name\" : \"env\",\r\n            \"type\" : {\r\n              \"type\" : \"record\",\r\n              \"name\" : \"Environment\",\r\n              \"fields\" : [ {\r\n                \"name\" : \"value\",\r\n                \"type\" : [ {\r\n                  \"type\" : \"enum\",\r\n                  \"name\" : \"EnvironmentConstants\",\r\n                  \"symbols\" : [ \"DEV\", \"UAT\", \"PROD\" ]\r\n                }, \"string\" ]\r\n              } ]\r\n            }\r\n          }, ...`\r\n\r\nThe json file:\r\n\r\n`{\r\n  \"info\": {\r\n    \"source\": {\r\n      \"app\": {\r\n        \"csi_id\": \"123\",\r\n        \"name\": \"ABC\"\r\n      },\r\n      \"env\": {\r\n        \"value\": {\r\n          \"risk.EnvironmentConstants\": \"PROD\"\r\n        }\r\n      }, ...`\r\n\r\nAny suggestions?"}}
{"actor":{"display_login":"sanchitgrover"}}