Обработка исключений в Java-приложении KafkaItemWriter Spring Batch Spring Boot

Получение этой ошибки:

2020-11-12 20: 16: 35.463 ОШИБКА 17552 --- [extractor-3] osksupport.LoggingProducerListener: исключение при отправке сообщения с ключом = 'abcdefghi' и полезной нагрузкой = '{заголовок: {id: ururirehdjd, Env : 88} и т.д ... »в тему ‹TOPIC_NAME›:

org.apache.kafka.common.errors.TimeoutException: Тема ‹TOPIC_NAME› отсутствует в метаданных через 60000 мс.

Мое весеннее пакетное задание передает эту ошибку и обновляет writeCount, даже если сообщение не опубликовано в теме. Как выбросить исключение в таких сценариях.?


spring-boot apache-kafka spring-batch kafka-producer-api
person Sonia    schedule 12.11.2020    source источник
comment
Я думаю, это связано с github.com/spring-projects/spring-batch/ вопросов / 3773. А пока вы можете использовать настраиваемый писатель и делать что-то вроде stackoverflow.com/a/64648499/5019386.   -  person Mahmoud Ben Hassine    schedule 13.11.2020
comment
Если в случае выполнения задания между двумя датами, у нас есть 100 сообщений для публикации в теме. Мы извлекаем сообщения из базы данных, и на 50-м сообщении я вижу, что сообщение не публикуется успешно, и мое задание выдает исключение и помечено как сбойное. и я снова запускаю свою работу вручную между теми же датами. Собирается ли она снова опубликовать первые 50 сообщений, которые были успешно опубликованы?   -  person Sonia    schedule 17.11.2020
comment
Какого размера ваш кусок? Вы используете KafkaTransactionManager в своей работе Spring Batch?   -  person Mahmoud Ben Hassine    schedule 17.11.2020
comment
Размер блока составляет 500, и я не использую KafkaTransactionManager в своей работе. Это просто базовый KafkaItemWriter с названием темы для записи в указанную тему.   -  person Sonia    schedule 17.11.2020
comment
Поскольку размер вашего блока составляет 500 и у вас есть 100 сообщений, это означает, что у вас будет один блок. Теперь, если задание не выполняется на элементе 50, откат будет выполнен для всего фрагмента, а при следующем запуске будет повторно прочитан первый фрагмент и повторно опубликованы эти элементы. То, что вы пытаетесь сделать, требует настройки Spring Batch для использования JtaTransactionManager, который координирует DataSourceTransactionManager (для таблиц метаданных) и KafkaTransactionManager (для вашего элемента записи), чтобы при откате транзакции ваши элементы не публиковались.   -  person Mahmoud Ben Hassine    schedule 18.11.2020
comment
Примером было 100 сообщений. У нас будут сообщения в сотнях тысяч. Итак, как вы объяснили выше, в случае сбоя будет выполнен откат всего фрагмента. Для этого мне нужны JtaTransactionManager и KafkaTransactionManager в моей работе?   -  person Sonia    schedule 18.11.2020