Avro 序列化异常 - java.time.Instant 无法转换为 java.lang.Long

Question

我想发送一个 Kafka 消息，其负载为 class，扩展了 SpecificRecordBase；它是在 Maven 插件的帮助下生成的 class。

我的模式的一个字段有一个时间戳类型，它对应于生成的 class 中的 java.time.Instant。

字段定义如下：

{"name": "processingTime", "type": {
   "type": "long",
   "logicalType": "timestamp-millis"
   }
},

当我创建这个 class 的实例并设置处理时间时，

setProcessingTime(RandomDate.randomInstant())

一切正常，但是当我运行程序并尝试将其发送到 Kafka 时，出现以下错误：

org.apache.kafka.common.errors.SerializationException: Can't convert value of class poc.avroGenerated.AvroMeasurement to class poc.avroSerde.AvroSerializer specified in value.serializer
Caused by: java.lang.ClassCastException: class java.time.Instant cannot be cast to class java.lang.Long (java.time.Instant and java.lang.Long are in module java.base of loader 'bootstrap')

这是我的自定义序列化器 class:

@Override
public byte[] serialize(String topic, T data) {

    byte[] result = null;
    try {
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        BinaryEncoder binaryEncoder = EncoderFactory.get().binaryEncoder(byteArrayOutputStream, null);
        DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(data.getSchema());
        datumWriter.write(data, binaryEncoder);
        binaryEncoder.flush();
        byteArrayOutputStream.close();
        result = byteArrayOutputStream.toByteArray();
    } catch (IOException e) {
        LOGGER.error(e);
    }
    return result;
}

Answer 1

使用 SpecificDatumWriter 代替 GenericDatumWriter。

加入那一项更改，您的自定义序列化程序看起来不错！

这经常是一个混淆点。在 Java 实现中，"generic" 数据不考虑内置到特定记录中的任何自定义，包括逻辑类型转换。

Avro 序列化异常 - java.time.Instant 无法转换为 java.lang.Long

Avro serialization exception - java.time.Instant cannot be cast to java.lang.Long

java

serialization

avro

apache-kafka