Получение ошибки отката при повторных попытках и не перезапуск kafka-потоков на марафоне

Question

У нас есть кафка с высокой доступностью. Мы запускаем несколько потоков kafka и запускаем задания на разные темы.

Но мы обнаружили странную проблему с отправкой записи. Полученная нами ошибка выглядит следующим образом

task [0_4] Ошибка отправки записи (ключ 27314 значение example.jar.class timestamp 1535741326818) в тему TOPIC_NAME из-за {}; Больше записей не будет отправлено, и больше смещений не будет записано для этой задачи. Вы можете увеличить параметр производителя retries и retry.backoff.ms, чтобы избежать этой ошибки. org.apache.kafka.streams.errors.StreamsException: task [0_4] Прервать отправку из-за ошибки, обнаруженной в предыдущей записи (значение ключа 27314 example.jar.class@6f7cb16c отметка времени 1535741326818) в тему TOPIC_NAME из-за записи с истекающим сроком 1 (s) ) для TOPIC_NAME-4: 42708 мс прошло с момента создания пакета плюс время ожидания Вы можете увеличить параметр производителя retries и retry.backoff.ms, чтобы избежать этой ошибки.

Мы также пытались сохранить время задержки 0, чтобы оно обрабатывалось сразу же после получения. но не повезло.

Наша конфигурация указана ниже

retries: 3
retry-backoff-ms: 20000
replication-factor: 1
reconnect-backoff-ms: 60000
max-poll-records: 1000
session-timeout-ms: 300000

Также у нас есть марафон для управления потоками kafka, но он не определяет его как сбой и не перезапускается автоматически.

Итак, я хочу либо правильно запустить потоки kafka, либо автоматически перезапустить марафон при таких ошибках (последний вариант).

Answer 1

Я столкнулся с той же проблемой. У меня 5 брокеров в кластере и 32 раздела темы с коэффициентом репликации 2.

Я забыл, что потоки кафки создают внутренние темы. И по умолчанию они имеют коэффициент репликации 1. И причина была в том, что когда мы пытаемся убить одного брокера, внутренняя тема больше недоступна, так как у нее нет реплик. Поэтому попробуйте установить коэффициент репликации для внутренних тем так же, как и для исходных. Надеюсь, это поможет.

Answer 2

Если в Kafka Stream возникает ошибка, умирает только внутренний StreamThread, но в "основном" потоке нет исключений. Наверное, поэтому ошибка не обнаружена. Рекомендуется зарегистрировать необработанный дескриптор исключения на клиенте KafkaStreams, чтобы получать информацию о умирающих потоках и реагировать соответствующим образом.

О самой ошибке. Это может быть известная проблема (для подробностей сравните KIP-91 ). Вы можете попытаться увеличить параметр request.timeout.ms в качестве обходного пути.