Spring Kafka Error Handler как избежать бесконечного цикла

Question

Я надеюсь, что кто-то может дать мне подсказку, если я здесь не так. Я написал собственный обработчик ошибок для пакетного прослушивателя, который должен искать за полученными записями, отправлять их в dlq. Я очень устал, но не получается. Моя текущая реализация будет висеть в бесконечном цикле, получая записи снова и снова. Вот код обработчика ошибок:

@Service("consumerAwareListenerErrorHandlerImpl")
public class ConsumerAwareListenerErrorHandlerImpl implements ConsumerAwareListenerErrorHandler {


    private final Executor executor;

    private final KafkaListenerEndpointRegistry registry;

    private final TaskScheduler scheduler;


    @Autowired
    public ConsumerAwareListenerErrorHandlerImpl(KafkaListenerEndpointRegistry registry, TaskScheduler scheduler) {
        this.scheduler = scheduler;
        this.executor = new SimpleAsyncTaskExecutor();
        this.registry = registry;
    }


    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {

        MessageHeaders headers = message.getHeaders();
        List<String> topics = headers.get(KafkaHeaders.RECEIVED_TOPIC, List.class);
        List<Integer> partitions = headers.get(KafkaHeaders.RECEIVED_PARTITION_ID, List.class);
        List<Long> offsets = headers.get(KafkaHeaders.OFFSET, List.class);
        Acknowledgment acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);


        Map<TopicPartition, Long> offsetsToReset = new HashMap<>();

        for (int i = 0; i < topics.size(); i++) {
            int index = i;
            offsetsToReset.compute(new TopicPartition(topics.get(i), partitions.get(i)),
                    (k, v) -> (v == null) ? offsets.get(index) : Math.max(v, offsets.get(index)));
        }
        offsetsToReset.forEach((k, v) -> consumer.seek(k, v));

        if (!(exception.getCause() instanceof DeserializationException)) {
            //pauseAndRestartContainer();
        }

        acknowledgment.acknowledge();
        consumer.commitSync();

        return null;
    }

Answer 1

Вы должны стремиться к смещению + 1, чтобы получить «прошлое».Поиск offset приведет к его воспроизведению.