使用GenericAvroDeserializer反序列化含decimal逻辑类型的Avro字节失败如何解决

解决方案 #

可以正常反序列化，按以下步骤操作即可：

1. 修正反序列化器配置 #

你当前将value.deserializer配置为org.apache.kafka.common.serialization.BytesDeserializer属于冗余配置，不需要手动中转Bytes类型，直接配置对应Avro反序列化器即可：

• 普通Kafka Consumer场景：value.deserializer配置为io.confluent.kafka.serializers.KafkaAvroDeserializer
• Kafka Streams场景：直接使用io.confluent.kafka.streams.serdes.avro.GenericAvroSerde作为流的Value Serde

2. 开启逻辑类型自动转换 #

在配置中添加以下参数，开启Avro逻辑类型自动转换能力，反序列化器会自动将带decimal逻辑类型的bytes字段转换为java.math.BigDecimal类型，而非返回原始字节数组：

deserializer.avro.use.logical.type.converters=true

3. 手动转换兜底方案 #

如果你的业务流程必须保留先反序列化为Bytes、再手动调用GenericAvroDeserializer的逻辑，可以拿到原始字节数组后，用Avro自带的转换工具手动解码：

import org.apache.avro.Conversions;
import org.apache.avro.LogicalTypes.Decimal;

// 已拿到的原始字节数组
byte[] amountBytes = (byte[]) genericRecord.get("totalAmount");
// 对应字段的Avro Schema
Decimal decimalType = (Decimal) fieldSchema.getLogicalType();

BigDecimal totalAmount = new Conversions.DecimalConversion()
        .fromBytes(ByteBuffer.wrap(amountBytes), fieldSchema, decimalType);

内容的提问来源于stack exchange，提问作者Digital