{"action":"created","issue":{"url":"https://api.github.com/repos/confluentinc/kafka-connect-hdfs/issues/51","repository_url":"https://api.github.com/repos/confluentinc/kafka-connect-hdfs","labels_url":"https://api.github.com/repos/confluentinc/kafka-connect-hdfs/issues/51/labels{/name}","comments_url":"https://api.github.com/repos/confluentinc/kafka-connect-hdfs/issues/51/comments","events_url":"https://api.github.com/repos/confluentinc/kafka-connect-hdfs/issues/51/events","html_url":"https://github.com/confluentinc/kafka-connect-hdfs/issues/51","id":149659878,"number":51,"title":"HDFS connector issue","user":{"login":"amitmalagi","id":17897319,"avatar_url":"https://avatars.githubusercontent.com/u/17897319?v=3","gravatar_id":"","url":"https://api.github.com/users/amitmalagi","html_url":"https://github.com/amitmalagi","followers_url":"https://api.github.com/users/amitmalagi/followers","following_url":"https://api.github.com/users/amitmalagi/following{/other_user}","gists_url":"https://api.github.com/users/amitmalagi/gists{/gist_id}","starred_url":"https://api.github.com/users/amitmalagi/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/amitmalagi/subscriptions","organizations_url":"https://api.github.com/users/amitmalagi/orgs","repos_url":"https://api.github.com/users/amitmalagi/repos","events_url":"https://api.github.com/users/amitmalagi/events{/privacy}","received_events_url":"https://api.github.com/users/amitmalagi/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":2,"created_at":"2016-04-20T05:07:03Z","updated_at":"2016-04-21T05:04:26Z","closed_at":null,"body":"HDFS connector throws this exception and halts execution even if one message in a kafka topic is erroneous. I found that kafka-avro-console-consumer continues with processing of subsequent messages when --skip-message-on-error option is used. Can the same behaviour be provided in HDFS connector?\r\n\r\nThanks.\r\n\r\nException in thread \"WorkerSinkTask-hdfs-sink-0\" org.apache.kafka.connect.errors.DataException: Failed to deserialize data to Avro:\r\n        at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:109)\r\n        at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:266)\r\n        at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:175)\r\n        at org.apache.kafka.connect.runtime.WorkerSinkTaskThread.iteration(WorkerSinkTaskThread.java:90)\r\n        at org.apache.kafka.connect.runtime.WorkerSinkTaskThread.execute(WorkerSinkTaskThread.java:58)\r\n        at org.apache.kafka.connect.util.ShutdownableThread.run(ShutdownableThread.java:82)\r\nCaused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 61\r\nCaused by: java.io.EOFException\r\n        at org.apache.avro.io.BinaryDecoder$ByteArrayByteSource.readRaw(BinaryDecoder.java:944)\r\n        at org.apache.avro.io.BinaryDecoder.doReadBytes(BinaryDecoder.java:349)\r\n        at org.apache.avro.io.BinaryDecoder.readString(BinaryDecoder.java:263)\r\n        at org.apache.avro.io.ResolvingDecoder.readString(ResolvingDecoder.java:201)\r\n        at org.apache.avro.generic.GenericDatumReader.readString(GenericDatumReader.java:363)\r\n        at org.apache.avro.generic.GenericDatumReader.readString(GenericDatumReader.java:355)\r\n        at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:157)\r\n        at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:193)\r\n        at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:183)\r\n        at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:151)\r\n        at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:142)\r\n        at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:130)\r\n        at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserializeWithSchemaAndVersion(AbstractKafkaAvroDeserializer.java:191)\r\n        at io.confluent.connect.avro.AvroConverter$Deserializer.deserialize(AvroConverter.java:130)\r\n        at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:99)\r\n        at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:266)\r\n        at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:175)"},"comment":{"url":"https://api.github.com/repos/confluentinc/kafka-connect-hdfs/issues/comments/212746795","html_url":"https://github.com/confluentinc/kafka-connect-hdfs/issues/51#issuecomment-212746795","issue_url":"https://api.github.com/repos/confluentinc/kafka-connect-hdfs/issues/51","id":212746795,"user":{"login":"amitmalagi","id":17897319,"avatar_url":"https://avatars.githubusercontent.com/u/17897319?v=3","gravatar_id":"","url":"https://api.github.com/users/amitmalagi","html_url":"https://github.com/amitmalagi","followers_url":"https://api.github.com/users/amitmalagi/followers","following_url":"https://api.github.com/users/amitmalagi/following{/other_user}","gists_url":"https://api.github.com/users/amitmalagi/gists{/gist_id}","starred_url":"https://api.github.com/users/amitmalagi/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/amitmalagi/subscriptions","organizations_url":"https://api.github.com/users/amitmalagi/orgs","repos_url":"https://api.github.com/users/amitmalagi/repos","events_url":"https://api.github.com/users/amitmalagi/events{/privacy}","received_events_url":"https://api.github.com/users/amitmalagi/received_events","type":"User","site_admin":false},"created_at":"2016-04-21T05:04:26Z","updated_at":"2016-04-21T05:04:26Z","body":"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.  "}}