Ошибка «Не удалось выделить память» и «Истекающая запись» после того, как брокер Kafka не работает более 5 минут, а затем возвращается в кластер

Question

Мы проводим некоторое тестирование на брокерах Kafka 1.1 в AWS EC2.В частности, мы аккуратно отключаем брокеров на 5 минут и более, а затем перезапускаем их, одновременно производя и потребляя из кластера.Теоретически это должно быть относительно беспрепятственно как для производителей, так и для потребителей.

Тем не менее, мы получаем эти ошибки в приложении производителя при тестировании неработающего брокера в течение более 5 минут:

 ERROR - 2019-02-27 04:34:24.946 - message size [2494] 
       -Expiring 7 record(s) for topic2-0: 30033 ms has passed since last append

И тут первая из многих ошибок, похожих на эту.

 ERROR - 2019-02-27 04:35:13.098; Topic [topic2], message size [2494] 
       -Failed to allocate memory within the configured max blocking time 60000 ms. 

В этот момент долгосрочные производители перестают отвечать на запросы, и каждый запрос производителя не выполняется с тем же Failed to allocate memory.Я пытался найти в Интернете похожие проблемы, но все, что я мог найти, - это старый билет Kafka JIRA, который был разрешен в 0.10.1.1, поэтому он не должен применяться к более новой версии 1.1, которую мы используем.

https://issues.apache.org/jira/browse/KAFKA-3651

Мы пробовали множество различных сценариев, включая изменение конфигурации производителя обратно на значения по умолчанию Kafka, чтобы посмотреть, поможет ли что-нибудь, но мы всегда сталкиваемся с этой проблемой, когда брокер возвращается в кластер.после 5 минут или более.

Answer 1

Наконец-то разобрался, как это исправить.Публикация здесь на случай, если это будет полезно кому-то еще.

Наш вызов producer.send определил обратный вызов, который в случае ошибки попытается повторно отправить сообщение, используя того же производителя, чтобыдругая тема.Этот механизм был построен таким образом, чтобы, если бы вызов send привел к ошибке, мы, по крайней мере, сохранили сообщение в теме ошибок, где мы могли бы проверить данные и выяснить, что происходит.

Этот механизм обычно работал хорошо, но всякий раз, когда мы временно теряли брокера (приходилось снимать его для обслуживания / обновления / и т. Д.), Все наши приложения зависали и в конечном итоге уничтожались ELB.

ИзОфициальный Javadoc для KafkaProducer.send(), я нашел это:

Обратите внимание, что обратные вызовы обычно выполняются в потоке ввода-вывода производителя и поэтому должны быть достаточно быстрыми, или они будут задерживать отправку сообщений отдругие темы.Если вы хотите выполнить блокирующие или вычислительно дорогие обратные вызовы, рекомендуется использовать своего собственного исполнителя в теле обратного вызова для распараллеливания обработки.

https://kafka.apache.org/11/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html

producer.send внутриобратный вызов квалифицируется как дорогой звонок, ИМХО.После некоторой дальнейшей отладки и профилирования выясняется, что все потоки XNIO застряли во время ожидания обратного вызова producer.send() для устаревших метаданных, я полагаю, с момента, когда брокер присоединился к кластеру.В конечном итоге это привело к тому, что приложение перестало отвечать.

Исправление в нашем случае состояло в том, чтобы просто поставить сообщение об ошибке в ConcurrentLinkedDeque и обработать эти сообщения отдельным потоком, единственной задачей которого было take сообщения от deque и попытайтесь отправить их заново.

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