Support Questions
Find answers, ask questions, and share your expertise
Announcements
Alert: Welcome to the Unified Cloudera Community. Former HCC members be sure to read and learn how to activate your account here.

Kafka Spout exception

Kafka Spout exception

I have a java app which produces ProducerRecords like this:

producer.send(new ProducerRecord<String, byte[]>(message.getMetadata().getTopic().getValue(), bytes), new ProducerCallback<T>(message, failedKafkaMessageRepository));

Serializer setting for producer:

props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
Now, I am trying to consumer from a Storm Kafka Spout this produced message:
17:13:50.225 [Thread-18-callEndKafkaSpout-executor[2 2]] ERROR com.plantronics.platform.data.storm.scheme.kafka.EventScheme - Error parsing Message from Kafka
java.io.CharConversionException: Invalid UTF-32 character 0x656e7473(above 10ffff)  at char #5, byte #19)
	at com.fasterxml.jackson.core.io.UTF32Reader.reportInvalid(UTF32Reader.java:153)
	at com.fasterxml.jackson.core.io.UTF32Reader.read(UTF32Reader.java:106)
	at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.loadMore(ReaderBasedJsonParser.java:127)
	at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._skipWSOrEnd(ReaderBasedJsonParser.java:1659)
	at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:561)
	at com.fasterxml.jackson.databind.ObjectMapper._initForReading(ObjectMapper.java:2834)
	at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:2781)
	at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:2028)
	at com.plantronics.platform.data.storm.scheme.kafka.EventScheme.deserialize(EventScheme.java:105)
	at com.plantronics.platform.data.storm.scheme.kafka.EventScheme.deserialize(EventScheme.java:66)
	at org.apache.storm.spout.SchemeAsMultiScheme.deserialize(SchemeAsMultiScheme.java:34)
	at org.apache.storm.kafka.KafkaUtils.generateTuples(KafkaUtils.java:235)
	at org.apache.storm.kafka.PartitionManager.next(PartitionManager.java:161)
	at org.apache.storm.kafka.KafkaSpout.nextTuple(KafkaSpout.java:135)




Don't have an account?
Coming from Hortonworks? Activate your account here