Hello,
I'm attempting to upgrade an application from 0.8 to 0.10 broker / client
libs, and integrate kafka streams. I am currently using the following
producer / consumer configs:
Producer:
Properties props = new Properties();
props.put("bootstrap.servers", brokerList);
props = ProducerConfig.addSerializerToConfig(props,
Serdes.String().serializer(), Serdes.String().serializer());
return props;
Consumer (kafka streams Processor)
Properties streamsSettings = new Properties();
streamsSettings.put("bootstrap.servers", brokerList);
streamsSettings.put("application.id", consumerGroupId);
streamsSettings.put("key.serde",
Serdes.StringSerde.class.getName());
streamsSettings.put("value.serde",
Serdes.StringSerde.class.getName());
StreamsConfig config = new StreamsConfig(streamsSettings);
I'm running a 0.10 broker. However, when I publish a message, I see the
following error on the consumer side:
org.apache.kafka.common.KafkaException: Error deserializing key/value for
partition <topic name>-0 at offset 0
at
org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:665)
at
org.apache.kafka.clients.consumer.internals.Fetcher.handleFetchResponse(Fetcher.java:593)
at
org.apache.kafka.clients.consumer.internals.Fetcher.access$000(Fetcher.java:71)
at
org.apache.kafka.clients.consumer.internals.Fetcher$1.onSuccess(Fetcher.java:142)
at
org.apache.kafka.clients.consumer.internals.Fetcher$1.onSuccess(Fetcher.java:139)
at
org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:133)
at
org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:107)
at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.onComplete(ConsumerNetworkClient.java:426)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:278)
at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:360)
at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:224)
at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:201)
at
org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:998)
at
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:937)
at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:295)
at
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:218)
Caused by: java.lang.IllegalArgumentException: null
at java.nio.Buffer.limit(Buffer.java:275)
at org.apache.kafka.common.record.Record.sliceDelimited(Record.java:392)
at org.apache.kafka.common.record.Record.key(Record.java:376)
at
org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:650)
... 15 common frames omitted
What could be causing this error? I see the same error if I attempt to use
the kafka bin tools to consume from the topic.
Thanks,
Ryan
