HDFS connector issue about kafka-connect-hdfs

[jcustenborder]

Hi @amitmalagi I'm not sure we want to skip records on error. Do you have a good idea of how the invalid record entered the topic? What are you using to produce to the topic?

[amitmalagi]

Hello @jcustenborder, The producer is a node.js application. I am using 'kafka-node' module to post messages to kafka and 'avsc' module to encode messages into arvo format. In some cases, if the message size is bigger than the allocated buffer size used for avro encoding, the resulting record would be invalid.

[tony-lijinwen]

@amitmalagi, I also encountered the similar issue, did you resolve it?

[amitmalagi]

@tony-lijinwen, I addressed this issue in our producer application.

[renukaradhya]

Even I am facing same issue. Please share the fix for the same.

[2016-09-26 13:51:42,569] INFO WorkerSinkTask{id=elasticsearch-schema-sink-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSinkTask)
[2016-09-26 13:52:34,437] INFO WorkerSinkTask{id=elasticsearch-schema-sink-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSinkTask)
[2016-09-26 13:52:34,446] ERROR Task elasticsearch-schema-sink-0 threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask)
org.apache.kafka.connect.errors.DataException: Failed to deserialize data to Avro:
at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:109)
at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:357)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:226)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:170)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:142)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:140)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:175)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

That error occurrs because your producer application sends messages that are not compatible with consumer. The confluent producer/consumer use some specific format of messages: first byte is 0, then 4 bytes - Id of schema in schema registry (The schema must be registered for the same topic as messages are sent to) and after that - message itself.

[cotedm]

@Tseretyan is correct and this format is documented here. I'm not seeing anything outstanding here so closing this out.