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)
