Рассмотрим следующие реальные запутанные журналы:
19:33:48,409 99733391 (pool-6-thread-11) ERROR [org.apache.kafka.clients.consumer.internals.ConsumerCoordinator] [] [Consumer clientId=app2.maria1.mcdonalnds_service_msg, groupId=mcdonalnds_service_msg] Offset commit failed on partition service_megaman_mt-mcdonalnds_service_msg-1 at offset 75796: This is not the correct coordinator.
19:33:48,410 99733392 (pool-6-thread-11) INFO [org.apache.kafka.clients.consumer.internals.AbstractCoordinator] [] [Consumer clientId=app2.maria1.mcdonalnds_service_msg, groupId=mcdonalnds_service_msg] Group coordinator kafka1.maria4.internal:9092 (id: 2147483646 rack: null) is unavailable or invalid, will attempt rediscovery
19:33:48,414 99733396 (kafka-producer-network-thread | producer-1) WARN [org.apache.kafka.clients.producer.internals.Sender] [] [Producer clientId=producer-1] Got error produce response with correlation id 16386 on topic-partition service_megaman_mo-mcdonalnds_service_msg-1, retrying (99 attempts left). Error: NOT_LEADER_FOR_PARTITION
19:33:48,510 99733492 (pool-6-thread-11) INFO [org.apache.kafka.clients.consumer.internals.AbstractCoordinator] [] [Consumer clientId=app2.maria1.mcdonalnds_service_msg, groupId=mcdonalnds_service_msg] Discovered group coordinator kafka3.maria4.internal:9092 (id: 2147483644 rack: null)
19:33:48,528 99733510 (pool-6-thread-11) ERROR [org.apache.kafka.clients.consumer.internals.ConsumerCoordinator] [] [Consumer clientId=app2.maria1.mcdonalnds_service_msg, groupId=mcdonalnds_service_msg] Offset commit failed on partition service_megaman_mt-mcdonalnds_service_msg-1 at offset 75796: The coordinator is not aware of this member.
19:33:48,528 99733510 (pool-6-thread-11) ERROR [com.bob.kafka.consumer.ListenableKafkaConsumer] [] Aborting consumer [mcdonalnds_service_msg] for topics [[service_megaman_mt-mcdonalnds_service_msg]] operation due to failure! Cause:
org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.
Насколько я понимаю, сообщение об исключении для poll () на самом деле не является причиной.Итак, что случилось: 1. Координатор не был доступен 2. Потребитель нашел нового координатора 3. Новый координатор не распознал смещение, поэтому отклонил коммит
Что я пытаюсь выяснить, это варианты выхода из этой ситуации.Это не случайная проблема, но случается один раз в год, поэтому настройка опроса не помогла бы, если бы лидер умер.
Что происходит сейчас: Исходный код приложения просто закрывал потребителей, что неправильновызвал оповещения и проснулся почти у всех, так как приложение перестало потреблять сообщения: -)
Что я хочу, чтобы произошло: Потребитель перезапускается, не умирает, если теряет связь с координатором
В чем я не уверен:
Почему координатор не знает об этом участнике
ЕслиЯ правильно понимаю проблему.: -)
На стороне сервиса с Java Kafka lib для класса KafkaConsumer я должен позвонить закрыть и подписаться или отписаться и подпишитесь , чтобы полностью выполнить мой сценарий восстановления потребителя.
- Что произойдет с обработанным смещением, которое было отклонено новым координатором?Поскольку смещение не было зафиксировано, я предполагаю, что потребитель будет перечитывать те же сообщения?
После post для Spring-kafka выглядит очень похожая проблема, но служба не использует Spring, так чтоимеет ограниченное использование для меня.