Bytes avro type not deserializing with kafka confluent GenericAvroDeserializer

Question

I'm using io.confluent.kafka.streams.serdes.avro.GenericAvroDeserializer to deserialize Kafka Message<Bytes> with kafka "value.deserializer" configured as org.apache.kafka.common.serialization.BytesDeserializer but the GenericAvroDeserializer not deserializing the avro payload with below type

{
      "name": "totalAmount",
      "type": [
        "null",
        {
          "logicalType": "decimal",
          "precision": 13,
          "scale": 2,
          "type": "bytes"
        }
      ]
    }

The resulted deserialized payload for attribute "amount" as below

"amount": {"bytes": "\u0001C"}

I see the internal implementation of AbstractKafkaAvroDeserializer as

if (schema.getType().equals(Type.BYTES)) {
    byte[] bytes = new byte[length];
    buffer.get(bytes, 0, length);
    result = bytes;
}

is there any way to properly deserialize the payload using io.confluent.kafka.streams.serdes.avro.GenericAvroDeserializer with avro type as bytes

Answer 1

I managed to deserialize this kind of bytes to decimal using this way:

LogicalTypes.Decimal decimalType = LogicalTypes.decimal(precision, scale);
BigDecimal bigDecimal = new Conversion.DecimalConversion().fromBytes((ByteBuffer) decimalInBytes, schema, decimalType);

This decimalInBytes is actually from GenericRecord when we do .get("totalAmount") from the instance of GenericRecord of your payload.