Kafka监听器：无法使用SimpleRetryPolicy来阻止特定异常的重试

要解决Kafka监听器无法使用SimpleRetryPolicy来阻止特定异常的重试的问题，可以使用SeekToCurrentErrorHandler结合自定义异常处理器来实现。

首先，需要创建一个自定义的异常处理器，用于判断特定异常是否应该被重试。以下是一个示例的异常处理器：

import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.kafka.listener.SeekUtils;
import org.springframework.kafka.listener.adapter.RecordInterceptor;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.serializer.DeserializationException;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.ErrorMessage;

public class CustomSeekToCurrentErrorHandler extends SeekToCurrentErrorHandler {

    private final Class<? extends Throwable> nonRetryableException;

    public CustomSeekToCurrentErrorHandler(Class<? extends Throwable> nonRetryableException) {
        this.nonRetryableException = nonRetryableException;
    }

    @Override
    public void handle(Exception exception, Consumer<?, ?> consumer) {
        if (exception instanceof DeserializationException && exception.getCause() != null
                && nonRetryableException.isAssignableFrom(exception.getCause().getClass())) {
            super.handle(new ErrorMessage(exception), consumer);
        } else {
            super.handle(exception, consumer);
        }
    }
    
    @Override
    public void handle(Exception exception, Consumer<?, ?> consumer, Message<?> message) {
        if (exception instanceof DeserializationException && exception.getCause() != null
                && nonRetryableException.isAssignableFrom(exception.getCause().getClass())) {
            super.handle(new ErrorMessage(exception), consumer, message);
        } else {
            super.handle(exception, consumer, message);
        }
    }
}

接下来，使用自定义的异常处理器和SimpleRetryPolicy创建一个SeekToCurrentErrorHandler实例。然后，将该实例设置为Kafka监听器的错误处理器。

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.listener.RetryListener;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.kafka.listener.adapter.RecordInterceptor;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.serializer.DeserializationException;
import org.springframework.retry.RetryPolicy;
import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

import java.util.HashMap;
import java.util.Map;

public class KafkaListenerExample {

    public static void main(String[] args) {
        // Kafka consumer configuration
        Map<String, Object> consumerConfig = new HashMap<>();
        consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // Kafka consumer factory
        ConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerConfig);

        // Kafka listener container properties
        ContainerProperties containerProperties = new ContainerProperties("test-topic");

        // Create a retry policy
        RetryPolicy retryPolicy = new SimpleRetryPolicy(3);

        // Create a retry template with fixed backoff policy
        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        backOffPolicy.setBackOffPeriod(1000);
        RetryTemplate retryTemplate = new RetryTemplate();
        retryTemplate.setRetryPolicy(retryPolicy);
        retryTemplate.setBackOffPolicy(backOffPolicy);

        // Create a custom exception handler
        Class<? extends Throwable> nonRetryableException = DeserializationException.class; // specify the non-retryable exception
        SeekToCurrentErrorHandler errorHandler = new CustomSeekToCurrentErrorHandler(nonRetryableException);
        errorHandler.setRetryTemplate(retryTemplate); // set the retry template

        // Create the Kafka message listener container
        KafkaMessageListenerContainer<String, String> container = new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);
        container.setErrorHandler(errorHandler); // set the custom error handler

        // Start the Kafka listener container
        container.start();
    }
}

在上面的示例中，创建了一个自定义的SeekToCurrentErrorHandler，并将其设置为Kafka监听器容器的错误处理器。同时，使用SimpleRetryPolicy创建了一个Retry