Недавно я попытался убедиться, что мое приложение Kafka Streams (версия 2.0.0) все еще будет работать (потоки будут возобновлены) после того, как какой-либо брокер Kafka начнет работать после неуказанного простоя всех брокеров (время простоя больше похоже на несколько часов, чем несколько секунд).
Он не подошел с конфигурацией по умолчанию (retries на 0), так как после того, как все брокеры были убиты, все мои потоки изменили состояние на ERROR или даже DEAD и перестали работать:
INFO o.a.k.s.p.internals.StreamThread - stream-thread [MyStream-0936f6a6-c9f4-4591-9b25-534abc65b8d1-StreamThread-24] State transition from RUNNING to PENDING_SHUTDOWN
и затем:
INFO o.a.k.s.p.internals.StreamThread - stream-thread [MyStream-0936f6a6-c9f4-4591-9b25-534abc65b8d1-StreamThread-24] State transition from PENDING_SHUTDOWN to DEAD
И некоторые из потоков перешли к ОШИБКЕ:
stream-client [MyStream-88a8fe9a-d565-43e3-acb5-20cccc6b4a88] State transition from RUNNING to ERROR
В результате, учитывая изменение состояния (https://static.javadoc.io/org.apache.kafka/kafka-streams/2.0.0/org/apache/kafka/streams/KafkaStreams.State.html) Я не могу возобновить работу каким-либо иным способом, кроме перезапуска приложения.
То, что я нашел, было этой темой:
Kafka Streams прекращает прослушивание темы и обработку сообщений, когда брокер отключается
Он отвечает на мой вопрос советом по увеличению моих потоков retries, а также retry.backoff.ms config. Это то, что я сделал (увеличил retries до Int.MaxNumber и retry.backoff.ms до 1000) и обнаружил, что есть некоторые проблемы с производительностью при таком подходе, и я случайно получил повторяющуюся ошибку в журналах: FETCH_SESSION_ID_NOT_FOUND, которую я не могу найти информацию о.
Существуют ли другие способы достижения устойчивости брокеров, кроме увеличения числа повторных попыток? Я могу принять некоторые сообщения, потерянные после сбоя Kafka Brokers, и мне не нужно повторять попытку, если сообщение не может быть создано / использовано. Я думал о ручном перезапуске потока после сбоя брокера, но я не совсем уверен, как отловить «исключение брокера», как вы думаете?
Это моя конфигурация потоков:
buffered.records.per.partition = 1000
cache.max.bytes.buffering = 10485760
commit.interval.ms = 30000
connections.max.idle.ms = 540000
default.key.serde = ... Serdes$StringSerde
default.production.exception.handler = ... DefaultProductionExHandler
default.timestamp.extractor = ... FailOnInvalidTimestamp
default.value.serde = ... Serdes$StringSerde
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
num.standby.replicas = 0
num.stream.threads = 8
partition.grouper = ... DefaultPartitionGrouper
poll.ms = 100
processing.guarantee = at_least_once
receive.buffer.bytes = 32768
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
replication.factor = 1
request.timeout.ms = 40000
retries = 0
retry.backoff.ms = 1000
rocksdb.config.setter = null
security.protocol = PLAINTEXT
send.buffer.bytes = 131072
state.cleanup.delay.ms = 600000
state.dir = /tmp/kafka-state
topology.optimization = all
upgrade.from = null
windowstore.changelog.additional.retention.ms = 86400000