Настройка SeekToCurrentErrorHandler в Spring Kafka для DLT и повторных попыток

Правильная настройка SeekToCurrentErrorHandler в Spring Kafka для работы с Dead Letter Topic требует создания кастомного обработчика ошибок с отслеживанием попыток и использованием DeadLetterPublishingRecoverer. Для старых версий Spring Kafka необходимо сконфигурировать MessagingMessageConverter для сохранения исходного ConsumerRecord в заголовках и реализовать логику подсчета попыток вручную.

Содержание

• Основы SeekToCurrentErrorHandler в Spring Kafka
• Конфигурация для работы с Dead Letter Topic в старых версиях
• Ключевые компоненты для обработки ошибок
• Пошаговая настройка SeekToCurrentErrorHandler
• Отладка и решение распространенных проблем
• Альтернативные подходы и рекомендации по миграции

Основы SeekToCurrentErrorHandler в Spring Kafka

SeekToCurrentErrorHandler - это механизм обработки ошибок в Spring Kafka, который обеспечивает повторную доставку сообщений с сохранением текущей позиции offset. В отличие от более новых версий, где используется DefaultErrorHandler, в старых версиях (до 2.8) этот обработчик требует дополнительной настройки для работы с Dead Letter Topic.

Ключевая особенность SeekToCurrentErrorHandler заключается в том, что он не автоматически подсчитывает количество попыток или не перенаправляет сообщения в DLT. Это требует ручной реализации счетчика попыток и логики перенаправления. Важно понимать, что этот обработчик предназначен для восстановления после временных сбоев, а не для обработки необратимых ошибок.

Конфигурация для работы с Dead Letter Topic в старых версиях

Для реализации счетчика повторных попыток и перенаправления в DLT в старых версиях Spring Kafka необходимо создать комплексную конфигурацию. Вот ключевые компоненты:

Настройка KafkaListenerContainerFactory

@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> factory() {
 ConcurrentKafkaListenerContainerFactory<String, String> factory = 
 new ConcurrentKafkaListenerContainerFactory<>();
 factory.setConsumerFactory(consumerFactory());
 factory.getContainerProperties().setPollTimeout(3000);
 
 // Настройка обработчика ошибок
 SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler(
 new DeadLetterPublishingRecoverer(kafkaTemplate()), 
 new FixedBackOff(1000, 1) // задержка 1 сек, 1 попытка повторной доставки
 );
 factory.setErrorHandler(errorHandler);
 
 return factory;
}

Кастомный обработчик ошибок с подсчетом попыток

@Bean
public KafkaListenerErrorHandler errorHandler(DeadLetterPublishingRecoverer recoverer) {
 return (msg, ex) -> {
 // Получаем количество попыток из заголовков
 Integer attempts = msg.getHeaders()
 .get(KafkaHeaders.DELIVERY_ATTEMPT, Integer.class);
 
 // Если попыток больше лимита, отправляем в DLT
 if (attempts != null && attempts > 9) {
 recoverer.accept(
 msg.getHeaders().get(KafkaHeaders.RAW_DATA, ConsumerRecord.class), 
 ex
 );
 return "FAILED";
 }
 throw ex; // Повторная доставка
 };
}

Ключевые компоненты для обработки ошибок

DeadLetterPublishingRecoverer

Этот компонент отвечает за отправку сообщений в Dead Letter Topic. Он требует настройки KafkaTemplate для отправки сообщений:

@Bean
public DeadLetterPublishingRecoverer recoverer() {
 return new DeadLetterPublishingRecoverer(kafkaTemplate(), 
 (record, ex) -> new TopicPartition("YOUR_DLT_TOPIC", record.partition()));
}

MessagingMessageConverter для сохранения исходных данных

Для работы с DLT необходимо сохранять исходный ConsumerRecord в заголовках:

@Bean
public MessagingMessageConverter messageConverter() {
 MessagingMessageConverter converter = new MessagingMessageConverter();
 converter.setRawRecordHeader(true); // Важно для старых версий!
 return converter;
}

FixedBackOff для контроля повторных попыток

Класс FixedBackOff настраивает задержку между попытками и максимальное количество попыток:

new FixedBackOff(1000, 1) // 1 секунда задержки, 1 дополнительная попытка

Пошаговая настройка SeekToCurrentErrorHandler для DLT

1. Создайте KafkaTemplate для отправки сообщений в DLT:

@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
 return new KafkaTemplate<>(producerFactory());
}

2. Настройте DeadLetterPublishingRecoverer:

@Bean
public DeadLetterPublishingRecoverer deadLetterRecoverer(KafkaTemplate<String, String> template) {
 return new DeadLetterPublishingRecoverer(template, 
 (cr, ex) -> new TopicPartition("dlt-topic", cr.partition()));
}

3. Создайте контейнер с обработчиком ошибок:

@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory(
 ConsumerFactory<String, String> consumerFactory,
 DeadLetterPublishingRecoverer recoverer) {
 
 ConcurrentKafkaListenerContainerFactory<String, String> factory = 
 new ConcurrentKafkaListenerContainerFactory<>();
 factory.setConsumerFactory(consumerFactory);
 
 SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler(
 recoverer, 
 new FixedBackOff(0, 10) // 10 попыток
 );
 factory.setErrorHandler(errorHandler);
 
 return factory;
}

4. Реализуйте KafkaListener с обработкой ошибок:

@KafkaListener(topics = "your-topic", 
 containerFactory = "kafkaListenerContainerFactory")
public void listen(String message, @Header(KafkaHeaders.DELIVERY_ATTEMPT) int attempt) {
 if (attempt > 5) {
 throw new RuntimeException("Максимальное количество попыток превышено");
 }
 // Ваша бизнес-логика
}

Отладка и решение распространенных проблем

Проблема: Сообщения не попадают в Dead Letter Topic

Решение: Проверьте настройку rawRecordHeader в MessagingMessageConverter. В старых версиях Spring Kafka это критически важно:

@Bean
public DefaultKafkaConsumerFactory<String, String> consumerFactory() {
 Map<String, Object> props = new HashMap<>();
 // ... ваши настройки
 
 DefaultKafkaConsumerFactory<String, String> factory = 
 new DefaultKafkaConsumerFactory<>(props);
 
 // Обязательно установите конвертер
 factory.setMessageConverter(new StringJsonMessageConverter());
 return factory;
}

Проблема: Счетчик попыток не увеличивается

Решение: Убедитесь, что у KafkaListener есть параметр @Header(KafkaHeaders.DELIVERY_ATTEMPT):

public void listen(String message, @Header(KafkaHeaders.DELIVERY_ATTEMPT) int attempt) {
 // ...
}

Проблема: Повторные попытки не происходят

Решение: Проверьте конфигурацию FixedBackOff. Убедитесь, что количество попыток больше нуля:

new FixedBackOff(1000, 3) // 3 попытки повторной доставки

Альтернативные подходы и рекомендации по миграции

Копирование реализации из новых версий

Для очень старых версий Spring Kafka можно скопировать реализацию SeekToCurrentErrorHandler из ветки master проекта. В версии 2.2+ этот обработчик уже поддерживает пропуск записей после заданного количества неудачных попыток:

SeekToCurrentErrorHandler errorHandler = 
 new SeekToCurrentErrorHandler((record, exception) -> {
 // Логика обработки неудачной записи
 }, 3); // 3 попытки

Рекомендация по миграции

В долгосрочной перспективе настоятельно рекомендуется обновить Spring Kafka до версии 2.8+ и использовать DefaultErrorHandler:

@Bean
public DefaultErrorHandler errorHandler() {
 return new DefaultErrorHandler(
 new DeadLetterPublishingRecoverer(kafkaTemplate()),
 new FixedBackOff(1000, 3)
 );
}

DefaultErrorHandler предоставляет более надежную обработку ошибок, включая:

• Автоматическое подсчет попыток
• Поддержку разных политик повторных попыток
• Лучшую интеграцию с Spring Boot

Источники

1. Spring Kafka Documentation — Руководство по обработке ошибок в Spring Kafka: https://docs.spring.io/spring-kafka/reference/kafka/annotation-error-handling.html
2. Stack Overflow - Artem Bilan — Ответ о настройке SeekToCurrentErrorHandler: https://stackoverflow.com/questions/52031014/spring-kafka-seektocurrenterrorhandler-find-out-which-record-has-failed
3. GitHub Issue — Пример конфигурации SeekToCurrentErrorHandler с DLT: https://github.com/spring-projects/spring-kafka/issues/1516
4. Spring Kafka GitHub — Исходный код SeekToCurrentErrorHandler: https://github.com/spring-projects/spring-kafka

Заключение

Правильная настройка SeekToCurrentErrorHandler для работы с Dead Letter Topic в старых версиях Spring Kafka требует комплексного подхода: создания кастомного обработчика ошибок с подсчетом попыток, настройки DeadLetterPublishingRecoverer и правильной конфигурации конвертеров сообщений. Хотя такой подход работает, он менее надежен, чем использование DefaultErrorHandler в современных версиях Spring Kafka. Для долгосрочного решения рекомендуется обновить версию фреймворка, что обеспечит более надежную обработку ошибок и автоматическую поддержку счетчиков повторных попыток.