Kafka Avro反序列化报错时，如何将springDeserializerExceptionValue解码为可读字符串？

Hey there, I've run into this exact issue before with Spring Kafka's Avro deserialization errors! That byte array in the springDeserializerExceptionValue header isn't a regular string—it's a Java-serialized DeserializationException object from Spring Kafka. Directly converting it to a string will just give you garbage characters, so you need to deserialize it back to the original exception object to get meaningful details.

Option 1: Write a Simple Java Tool to Deserialize the Byte Array #

You can use Java's ObjectInputStream to convert the byte array back into a DeserializationException instance, then extract the error message, cause, and even the failed payload. Here's a quick example:

import org.springframework.kafka.support.serializer.DeserializationException;
import java.io.ByteArrayInputStream;
import java.io.ObjectInputStream;

public class KafkaErrorDecoder {
    public static void main(String[] args) throws Exception {
        // Replace this with your actual byte array from the error header
        byte[] errorBytes = {-84, -19, 0, 5, 115, 114, 0, 69, 111, 114, 103, 46, 115, 112, 114, 105, 110, 103, 102, 114, 97, 109, 101, 119, 111, 114, 107, 46, 107, 97, 102, 107, 97, 46, 115, 117, 112, 112, 111, 114, 116, 46, 115, 101, 114, 105, 97, 108, 105, 122, 101, 114, 46, 68, 101, 115, 101, 114, 105, 97, 108, 105, 122, 97, 116, 105, 111, 110, 69, 120, 99, 101, 112, 116, 105, 111, 110, -26, -50, 105, 87, -16, 47, -111, -25, 2, 0, 2, 90, 0, 5, 105, 115, 75, 101, 121, 91, 0, 4, 100, 97, 116, 97, 116, 0, 2, 91, 66, 120, 114, 0, 40, 111, 114, 103, 46, 115, 112, 114, 105, 110, 103, 102, 114, 97, 109, 101, 119, 111, 114, 107, 46, 107, 97, 102, 107, 97, 46, 75, 97, 102, 107, 97, 69, 120, 99, 101, 112, 116, 105, 111, 110, 67, 55, -37, -114};

        try (ByteArrayInputStream bais = new ByteArrayInputStream(errorBytes);
             ObjectInputStream ois = new ObjectInputStream(bais)) {
            DeserializationException deserializationEx = (DeserializationException) ois.readObject();
            
            // Print key details from the exception
            System.out.println("Deserialization Error Message: " + deserializationEx.getMessage());
            System.out.println("Root Cause: " + deserializationEx.getCause());
            System.out.println("Failed Message Key Flag: " + deserializationEx.isKey());
            // Uncomment below to inspect raw payload bytes
            // System.out.println("Raw Payload Bytes: " + Arrays.toString(deserializationEx.getData()));
        }
    }
}

Note: You'll need the Spring Kafka dependency in your project to access the DeserializationException class.

Option 2: Configure Spring Kafka to Print Errors Directly (Preventative Measure) #

Instead of parsing the serialized header later, you can configure Spring Kafka's error handling to log readable errors upfront. Use ErrorHandlingDeserializer with a custom error handler to print details when deserialization fails:

import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.DeserializationException;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    
    // Configure error handler to log detailed errors
    factory.setCommonErrorHandler(new DefaultErrorHandler((record, exception) -> {
        System.err.println("Failed to process record - Topic: " + record.topic() + ", Offset: " + record.offset());
        if (exception instanceof DeserializationException) {
            DeserializationException de = (DeserializationException) exception;
            System.err.println("Deserialization Failed: " + de.getMessage());
            System.err.println("Root Cause: " + de.getCause());
        }
    }, new org.springframework.util.backoff.FixedBackOff(0, 0)));
    
    return factory;
}

@Bean
public DefaultKafkaConsumerFactory<String, Object> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    // Add your Kafka consumer configs here
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-bootstrap-servers");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "your-group-id");
    
    // Use ErrorHandlingDeserializer to wrap Avro deserializer
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
    props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, KafkaAvroDeserializer.class);
    
    // Add your Avro configs
    props.put("schema.registry.url", "your-schema-registry-url");
    props.put("specific.avro.reader", "true");
    
    return new DefaultKafkaConsumerFactory<>(props);
}

This setup will log clear, readable error messages as soon as a deserialization failure happens, so you don't have to manually parse the serialized header later.

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