{"action":"created","issue":{"url":"https://api.github.com/repos/datamountaineer/stream-reactor/issues/80","repository_url":"https://api.github.com/repos/datamountaineer/stream-reactor","labels_url":"https://api.github.com/repos/datamountaineer/stream-reactor/issues/80/labels{/name}","comments_url":"https://api.github.com/repos/datamountaineer/stream-reactor/issues/80/comments","events_url":"https://api.github.com/repos/datamountaineer/stream-reactor/issues/80/events","html_url":"https://github.com/datamountaineer/stream-reactor/issues/80","id":186489242,"number":80,"title":"Cassandra Sink - Error when consuming Avro messages ","user":{"login":"ferhtaydn","id":5160614,"avatar_url":"https://avatars.githubusercontent.com/u/5160614?v=3","gravatar_id":"","url":"https://api.github.com/users/ferhtaydn","html_url":"https://github.com/ferhtaydn","followers_url":"https://api.github.com/users/ferhtaydn/followers","following_url":"https://api.github.com/users/ferhtaydn/following{/other_user}","gists_url":"https://api.github.com/users/ferhtaydn/gists{/gist_id}","starred_url":"https://api.github.com/users/ferhtaydn/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ferhtaydn/subscriptions","organizations_url":"https://api.github.com/users/ferhtaydn/orgs","repos_url":"https://api.github.com/users/ferhtaydn/repos","events_url":"https://api.github.com/users/ferhtaydn/events{/privacy}","received_events_url":"https://api.github.com/users/ferhtaydn/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":5,"created_at":"2016-11-01T09:13:52Z","updated_at":"2016-11-01T09:51:04Z","closed_at":null,"body":"Hi,\r\n\r\nI am trying to use cassandra-sink connector to gather avro messages in a topic. Each message is a GenericRecord. I am taking the \"unknown magic byte\" error as shown below;\r\n\r\n`[2016-11-01 10:58:51,944] INFO WorkerSinkTask{id=cassandra-sink-products-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSinkTask:261)\r\n[2016-11-01 10:58:51,959] ERROR Task cassandra-sink-products-0 threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:142)\r\norg.apache.kafka.connect.errors.DataException: Failed to deserialize data to Avro:\r\n\tat io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:109)\r\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:356)\r\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:226)\r\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:170)\r\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:142)\r\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:140)\r\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:175)\r\n\tat java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)\r\n\tat java.util.concurrent.FutureTask.run(FutureTask.java:266)\r\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)\r\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)\r\n\tat java.lang.Thread.run(Thread.java:745)\r\nCaused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1\r\nCaused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!\r\n[2016-11-01 10:58:51,959] ERROR Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:143)\r\n[2016-11-01 10:58:51,959] INFO Stopping Cassandra sink. (com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraSinkTask:95)\r\n[2016-11-01 10:58:51,959] INFO Shutting down Cassandra driver session and cluster. (com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraJsonWriter:166)\r\n[2016-11-01 10:58:54,188] INFO Publish thread interrupted! (io.confluent.monitoring.clients.interceptor.MonitoringInterceptor:161)\r\n[2016-11-01 10:58:54,296] INFO Publishing Monitoring Metrics stopped for clientID=consumer-4 (io.confluent.monitoring.clients.interceptor.MonitoringInterceptor:173)\r\n[2016-11-01 10:58:54,296] INFO Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms. (org.apache.kafka.clients.producer.KafkaProducer:658)\r\n[2016-11-01 10:58:54,300] INFO Closed monitoring interceptor for client ID=consumer-4 (io.confluent.monitoring.clients.interceptor.MonitoringInterceptor:195)`\r\n\r\nI try both the official apache-kafka and confluent fork. The error is the same. \r\n\r\n```\r\n\"io.confluent\" % \"kafka-avro-serializer\" % \"3.0.1\",\r\n\"org.apache.kafka\" % \"kafka_2.11\" % \"0.10.0.1-cp1\"\r\n//\"org.apache.kafka\" % \"kafka_2.11\" % \"0.10.0.1\"\r\n```\r\n\r\nI can successfully feed the orders table (producer records are given from the console-producer), as explained in the example (http://docs.datamountaineer.com/en/latest/cassandra-sink.html). However, when I try using records in a topic, it does not work.\r\n\r\nDoes anyone have a solution for this problem? I investigated the problem in the web and try many things but could not find a working solution.\r\n\r\nThanks,\r\nFerhat\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/datamountaineer/stream-reactor/issues/comments/257526885","html_url":"https://github.com/datamountaineer/stream-reactor/issues/80#issuecomment-257526885","issue_url":"https://api.github.com/repos/datamountaineer/stream-reactor/issues/80","id":257526885,"user":{"login":"andrewstevenson","id":10235749,"avatar_url":"https://avatars.githubusercontent.com/u/10235749?v=3","gravatar_id":"","url":"https://api.github.com/users/andrewstevenson","html_url":"https://github.com/andrewstevenson","followers_url":"https://api.github.com/users/andrewstevenson/followers","following_url":"https://api.github.com/users/andrewstevenson/following{/other_user}","gists_url":"https://api.github.com/users/andrewstevenson/gists{/gist_id}","starred_url":"https://api.github.com/users/andrewstevenson/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/andrewstevenson/subscriptions","organizations_url":"https://api.github.com/users/andrewstevenson/orgs","repos_url":"https://api.github.com/users/andrewstevenson/repos","events_url":"https://api.github.com/users/andrewstevenson/events{/privacy}","received_events_url":"https://api.github.com/users/andrewstevenson/received_events","type":"User","site_admin":false},"created_at":"2016-11-01T09:51:04Z","updated_at":"2016-11-01T09:51:04Z","body":"@ferhtaydn click on the slack icon on the README.md"}}
{"actor":{"display_login":"andrewstevenson"}}