处理 Kafka 批量监听器反序列化错误的重试机制

@org.springframework.context.annotation.Configuration
@EnableKafka
public class Configuration {
    @Bean("myContainerFactory")
     public ConcurrentKafkaListenerContainerFactory<String, String> createFactory(
             KafkaProperties properties
    ) {
        var factory = new ConcurrentKafkaListenerContainerFactory<String, String>();
        factory.setConsumerFactory(
                new DefaultKafkaConsumerFactory(
                        properties.buildConsumerProperties(),
                        new StringDeserializer(),
                        new ErrorHandlingDeserializer(new MyDeserializer())
                )
        );
        factory.getContainerProperties().setAckMode(
                ContainerProperties.AckMode.MANUAL_IMMEDIATE
        );
        DefaultErrorHandler errorHandler = new DefaultErrorHandler();
        errorHandler.removeClassification(DeserializationException.class);
        factory.setCommonErrorHandler(errorHandler);
        return factory;
    }

    // this fakes occasional errors which succeed after a retry
    static class MyDeserializer implements Deserializer<String> {
        int retries = 0;
        @Override
        public String deserialize(String topic, byte[] bytes) {
            String s = new String(bytes);
            if (s.contains("7") && retries == 0) {
                retries = 1;
                throw new RuntimeException();
            }
            retries = 0;
            return s;
        }
    }
}

   @Component
   public class StringListener {
       @KafkaListener(
               topics = {"string-test"},
               groupId = "test",
               batch = "true",
               containerFactory = "myContainerFactory"
       )
       public void listen(List<Message<String>> messages, Acknowledgment acknowledgment) {
           for (Message<String> message: messages) {
               try {
                   String s = message.getPayload();
                   System.out.println(s);
               } catch (Exception e) {
                   // 获取反序列化异常
                   byte[] exceptionBytes = (byte[]) message.getHeaders().get(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER);
                   DeserializationException deserializationException = byteArrayToDeserializationException(exceptionBytes);

                   // 重新抛出异常，触发重试
                   throw new ListenerExecutionFailedException("Deserialization failed", deserializationException);
               }
           }
           acknowledgment.acknowledge();
       }

       private DeserializationException byteArrayToDeserializationException(byte[] bytes) {
            ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
            ObjectInputStream ois;
            try {
                ois = new ObjectInputStream(bais);
                return (DeserializationException) ois.readObject();
            } catch (IOException | ClassNotFoundException e) {
                throw new RuntimeException("Failed to deserialize exception from byte array", e);
            }
        }
   }