我正在尝试使用汇合的kafka连接功能来收听我发布的主题。但是,我无法对其进行反序列化。我相信它的avro序列化但无法找到正确的解串器。
消息如下所示在控制台主题中
null {"c1":{"int":10},"c2":{"string":"foo"},"create_ts":1552598863000,"update_ts":1552598863000}
下面是解串器
public class AvroDeserializer<T extends SpecificRecordBase> implements Deserializer<T> {
private static final Logger LOGGER = LoggerFactory.getLogger(AvroDeserializer.class);
protected final Class<T> targetType;
public AvroDeserializer(Class<T> targetType) {
this.targetType = targetType;
}
@Override
public void close() {
// No-op
}
@Override
public void configure(Map<String, ?> arg0, boolean arg1) {
// No-op
}
@SuppressWarnings("unchecked")
@Override
public T deserialize(String topic, byte[] data) {
try {
T result = null;
if (data != null) {
LOGGER.debug("data='{}'", DatatypeConverter.printHexBinary(data));
DatumReader<GenericRecord> datumReader =
new SpecificDatumReader<>(targetType.newInstance().getSchema());
Decoder decoder = DecoderFactory.get().binaryDecoder(data, null);
result = (T) datumReader.read(null, decoder);
LOGGER.debug("deserialized data='{}'", result);
}
return result;
} catch (Exception ex) {
throw new SerializationException(
"Can't deserialize data '" + Arrays.toString(data) + "' from topic '" + topic + "'", ex);
}
}
}
例外
org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition mysql-foobar-0 at offset 10. If needed, please seek past the record to continue consumption.
Caused by: org.apache.kafka.common.errors.SerializationException: Can't deserialize data '[0, 0, 0, 0, 21, 2, 20, 2, 6, 102, 111, 111, -80, -78, -44, -31, -81, 90, -80, -78, -44, -31, -81, 90]' from topic 'mysql-foobar'
Caused by: java.lang.InstantiationException: null
at sun.reflect.InstantiationExceptionConstructorAccessorImpl.newInstance(InstantiationExceptionConstructorAccessorImpl.java:48) ~[na:1.8.0_131]
at java.lang.reflect.Constructor.newInstance(Constructor.java:423) ~[na:1.8.0_131]
at java.lang.Class.newInstance(Class.java:442) ~[na:1.8.0_131]
at com.spring.kafkaexample.springbootkafkaconsumer.config.AvroDeserializer.deserialize(AvroDeserializer.java:48) ~[classes/:na]
at com.spring.kafkaexample.springbootkafkaconsumer.config.AvroDeserializer.deserialize(AvroDeserializer.java:18) ~[classes/:na]
如下所示在控制台主题中
目前尚不清楚你是使用kafka-avro-console-consumer
还是仅仅使用kafka-console-consumer
。了解您的数据是否为Avro的方法是查看Producer / connector配置。
但是,没有必要编写自己的反序列化器。另外,Confluent不会使用您的代码所具有的Avro架构+消息约定(因此您会收到该错误)。您需要首先从架构注册表中查找架构。
添加Confluent Maven仓库
<repositories>
<repository>
<id>confluent</id>
<url>https://packages.confluent.io/maven/</url>
</repository>
</repositories>
然后添加Confluent序列化程序依赖项
<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-avro-serializer</artifactId>
<version>${confluent.version}</version>
</dependency>
然后import io.confluent.kafka.serializers.KafkaAvroDeserializer
,或在您的消费者配置中使用该类
https://docs.confluent.io/current/clients/install.html#java
或者,您可以将MySQL连接器切换为不使用Avro Converter