А кто может подсказать, с чем связана такая ошибка?

INFO tocheck=[Store{name=TriggerEventsStore, rollup=false}, Store{name=MonitoringMessageAggregatorWindows, rollup=true}, Store{name=MonitoringStream, rollup=true}, Store{name=KSTREAM-OUTERTHIS-0000000104-store, rollup=false}, Store{name=AlertHistoryStore, rollup=false}, Store{name=group-aggregate-store, rollup=true}, Store{name=aggregate-topic-partition-store, rollup=false}, Store{name=MonitoringVerifierStore, rollup=false}, Store{name=Group, rollup=true}, Store{name=aggregatedTopicPartitionTableWindows, rollup=true}, Store{name=monitoring-aggregate-rekey-store, rollup=false}, Store{name=TriggerActionsStore, rollup=false}, Store{name=MonitoringTriggerStore, rollup=false}, Store{name=KSTREAM-OUTEROTHER-0000000105-store, rollup=false}, Store{name=MetricsAggregateStore, rollup=false}] (io.confluent.controlcenter.streams.KafkaStreamsManager)
 INFO streams in state=REBALANCING (io.confluent.controlcenter.streams.KafkaStreamsManager)

Сначала cp-control-center запускается и даже веб-морду отдает, потом через некоторое время отваливается коннект до зукипера и до кафки, а потом он с концами падает и постоянно пишет на экран одно и то же.

рестарт сервиса не помог

версия 5,2,0

Это повторяется каждые 1-2 сек. Лаг увеличивается

теперь он снова запустился

теперь опять упал

А что произойдет если брокер умрет? А на нем есть лидеры. Произойдет переизбрание лидера.

Что будет с продьюсером который пишет в лидер в момент переизбрания?
Можно ли повлиять на поведение продьюсера? (чтобы ждал, игнорил или еще что то)

Я так понмаю просто выставить acks

Тогда буффер продьюсера будет наполняться

B выходит что все сообщения уйдут в небытие. Ибо буфер продьюсера переполнен

пока происходит выбор лидера

ну линкедин вон круизконтроль запилил что бы за здоровьем кластеров следить

контролить это одно. Мне интересен механизм когда кластер таки даст сбой

и один брокер упадет

У меня еще и lua выступает продьюсером. А там нет обертки над librdkafka актуальной
Потому создание продьюсера просто «класс»

local kafka_producer = producer:new(broker_table, { producer_type = "async", refresh_interval=30000 })

собери тестовый стенд и проверь, делов на 10 минут максимум

на тестовом стенде тяжело создать нагрузку как на проде

в чем тяжесть ? у тебя есть тесты, сделай больше потоков