За несколько месяцев я создал go две группы потребителей (каждая группа имеет 3 потребителя) в моей цепочке Kafka:
- Первая группа: взять данные из внешнего источника и отправить их в мою тему1
- Вторая группа: потребляйте из topic1
Это без проблем.
Недавно я добавил третью группу, чтобы добавить парралелизацию в свою цепочку:
- Влияние на первую группу: также отправить в тему3 (поверх отправки в тему1)
- Новая третья группа (также с 3 потребителями): потреблять из темы3
Проблема: новая тема3 на самом деле не используется (несколько 20 или 50 сообщений в каждой теме израсходованы через много часов ...) => огромное «отставание» (никогда не используется) на eah c из 3 разделов третьей группы.
В журналах kafka я вижу, что многие ребалансируются ТОЛЬКО для третьей группы (кажется, достигает 5 минут по умолчанию max.poll.interval.ms, но я не уверен: сообщение kafka четко не указано: Подготовка к ребалансировке группа group3 ... ). Я попытался изменить параметры опроса, уменьшив количество max.poll.records et c .. с уменьшением дела обстоят лучше, но я продолжаю перебалансировать ... Я попытался добавить больше ядер ЦП (я думал, что многие потребители были на моем сервере для ЦП) от 1 до 2 и, наконец, 4 виртуальных ядра: на самом деле ничего не меняется, и htop не показывает огромную загрузку ядра ЦП ...
* 1026 ребалансировка с новой группой потребителей, а код потребления такой же, как и у других потребителей ...
Я не знаю, что делать ... Любые предложения по kafka conf, использованию процессора kafka или чему-то еще?
Примечание: весь стек находится под Docker контейнерами
Спасибо!
РЕДАКТИРОВАТЬ: Пример 1 (l oop генерации БЕЗ сброса на 0) журналов kafka:
...
[2020-07-15 10:52:32,029] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 365 (__consumer_offsets-38) (reason: Adding new member kafka-python-2.0.1-319d2d85-ad11-4012-9e30-d8c94f706bbe with group instance id None) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 10:52:32,036] INFO [GroupCoordinator 1001]: Stabilized group group3 generation 366 (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 10:52:32,040] INFO [GroupCoordinator 1001]: Assignment received from leader for group group3 for generation 366 (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 10:54:00,552] INFO [GroupMetadataManager brokerId=1001] Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-07-15 10:57:32,144] INFO [GroupCoordinator 1001]: Member[group.instance.id None, member.id kafka-python-2.0.1-319d2d85-ad11-4012-9e30-d8c94f706bbe] in group group3 has left, removing it from the group (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 10:57:32,149] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 366 (__consumer_offsets-38) (reason: removing member kafka-python-2.0.1-319d2d85-ad11-4012-9e30-d8c94f706bbe on LeaveGroup) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 10:57:32,157] INFO [GroupCoordinator 1001]: Group group3 with generation 367 is now empty (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:04:00,553] INFO [GroupMetadataManager brokerId=1001] Removed 0 expired offsets in 2 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-07-15 11:06:14,173] INFO [GroupCoordinator 1001]: Dynamic Member with unknown member id joins group group3 in Empty state. Created a new member id kafka-python-2.0.1-9eae5a06-c39f-4b66-b9c6-e2e520d787c1 for this member and add to the group. (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:06:14,182] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 367 (__consumer_offsets-38) (reason: Adding new member kafka-python-2.0.1-9eae5a06-c39f-4b66-b9c6-e2e520d787c1 with group instance id None) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:06:14,205] INFO [GroupCoordinator 1001]: Stabilized group group3 generation 368 (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:06:14,210] INFO [GroupCoordinator 1001]: Assignment received from leader for group group3 for generation 368 (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:11:14,233] INFO [GroupCoordinator 1001]: Member[group.instance.id None, member.id kafka-python-2.0.1-9eae5a06-c39f-4b66-b9c6-e2e520d787c1] in group group3 has left, removing it from the group (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:11:14,234] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 368 (__consumer_offsets-38) (reason: removing member kafka-python-2.0.1-9eae5a06-c39f-4b66-b9c6-e2e520d787c1 on LeaveGroup) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:11:14,235] INFO [GroupCoordinator 1001]: Group group3 with generation 369 is now empty (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:14:00,552] INFO [GroupMetadataManager brokerId=1001] Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-07-15 11:15:42,480] INFO [GroupCoordinator 1001]: Dynamic Member with unknown member id joins group group3 in Empty state. Created a new member id kafka-python-2.0.1-560ebdf5-4c6f-4f12-85d5-bf6e7194b712 for this member and add to the group. (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:15:42,483] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 369 (__consumer_offsets-38) (reason: Adding new member kafka-python-2.0.1-560ebdf5-4c6f-4f12-85d5-bf6e7194b712 with group instance id None) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:15:42,489] INFO [GroupCoordinator 1001]: Stabilized group group3 generation 370 (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:15:42,505] INFO [GroupCoordinator 1001]: Assignment received from leader for group group3 for generation 370 (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:20:42,655] INFO [GroupCoordinator 1001]: Member[group.instance.id None, member.id kafka-python-2.0.1-560ebdf5-4c6f-4f12-85d5-bf6e7194b712] in group group3 has left, removing it from the group (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:20:42,657] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 370 (__consumer_offsets-38) (reason: removing member kafka-python-2.0.1-560ebdf5-4c6f-4f12-85d5-bf6e7194b712 on LeaveGroup) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:20:42,657] INFO [GroupCoordinator 1001]: Group group3 with generation 371 is now empty (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:24:00,552] INFO [GroupMetadataManager brokerId=1001] Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-07-15 11:34:00,552] INFO [GroupMetadataManager brokerId=1001] Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-07-15 11:36:26,840] INFO [GroupCoordinator 1001]: Dynamic Member with unknown member id joins group group3 in Empty state. Created a new member id kafka-python-2.0.1-6c853061-c4e8-4066-8940-2c7173709a96 for this member and add to the group. (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:36:26,860] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 371 (__consumer_offsets-38) (reason: Adding new member kafka-python-2.0.1-6c853061-c4e8-4066-8940-2c7173709a96 with group instance id None) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:36:26,869] INFO [GroupCoordinator 1001]: Stabilized group group3 generation 372 (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:36:26,879] INFO [GroupCoordinator 1001]: Assignment received from leader for group group3 for generation 372 (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:41:26,949] INFO [GroupCoordinator 1001]: Member[group.instance.id None, member.id kafka-python-2.0.1-6c853061-c4e8-4066-8940-2c7173709a96] in group group3 has left, removing it from the group (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:41:26,952] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 372 (__consumer_offsets-38) (reason: removing member kafka-python-2.0.1-6c853061-c4e8-4066-8940-2c7173709a96 on LeaveGroup) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:41:26,953] INFO [GroupCoordinator 1001]: Group group3 with generation 373 is now empty (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:44:00,552] INFO [GroupMetadataManager brokerId=1001] Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-07-15 11:49:45,548] INFO [GroupCoordinator 1001]: Dynamic Member with unknown member id joins group group3 in Empty state. Created a new member id kafka-python-2.0.1-834458a8-67f7-4a0d-9056-52b6b61e79db for this member and add to the group. (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:49:45,576] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 373 (__consumer_offsets-38) (reason: Adding new member kafka-python-2.0.1-834458a8-67f7-4a0d-9056-52b6b61e79db with group instance id None) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:49:45,580] INFO [GroupCoordinator 1001]: Stabilized group group3 generation 374 (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:49:45,586] INFO [GroupCoordinator 1001]: Assignment received from leader for group group3 for generation 374 (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:54:00,552] INFO [GroupMetadataManager brokerId=1001] Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-07-15 11:54:45,676] INFO [GroupCoordinator 1001]: Member[group.instance.id None, member.id kafka-python-2.0.1-834458a8-67f7-4a0d-9056-52b6b61e79db] in group group3 has left, removing it from the group (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:54:45,685] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 374 (__consumer_offsets-38) (reason: removing member kafka-python-2.0.1-834458a8-67f7-4a0d-9056-52b6b61e79db on LeaveGroup) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:54:45,686] INFO [GroupCoordinator 1001]: Group group3 with generation 375 is now empty (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:59:01,403] INFO [GroupCoordinator 1001]: Dynamic Member with unknown member id joins group group3 in Empty state. Created a new member id kafka-python-2.0.1-f8f4d080-73df-43bd-b49b-5b5403d15e00 for this member and add to the group. (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:59:01,432] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 375 (__consumer_offsets-38) (reason: Adding new member kafka-python-2.0.1-f8f4d080-73df-43bd-b49b-5b5403d15e00 with group instance id None) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:59:01,471] INFO [GroupCoordinator 1001]: Stabilized group group3 generation 376 (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 11:59:01,481] INFO [GroupCoordinator 1001]: Assignment received from leader for group group3 for generation 376 (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:04:00,552] INFO [GroupMetadataManager brokerId=1001] Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-07-15 12:04:01,515] INFO [GroupCoordinator 1001]: Member[group.instance.id None, member.id kafka-python-2.0.1-f8f4d080-73df-43bd-b49b-5b5403d15e00] in group group3 has left, removing it from the group (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:04:01,518] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 376 (__consumer_offsets-38) (reason: removing member kafka-python-2.0.1-f8f4d080-73df-43bd-b49b-5b5403d15e00 on LeaveGroup) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:04:01,519] INFO [GroupCoordinator 1001]: Group group3 with generation 377 is now empty (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:14:00,552] INFO [GroupMetadataManager brokerId=1001] Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-07-15 12:21:11,498] INFO [GroupCoordinator 1001]: Dynamic Member with unknown member id joins group group3 in Empty state. Created a new member id kafka-python-2.0.1-43259283-cdeb-4738-aaa4-44e83542c7dc for this member and add to the group. (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:21:11,502] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 377 (__consumer_offsets-38) (reason: Adding new member kafka-python-2.0.1-43259283-cdeb-4738-aaa4-44e83542c7dc with group instance id None) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:21:11,505] INFO [GroupCoordinator 1001]: Stabilized group group3 generation 378 (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:21:11,508] INFO [GroupCoordinator 1001]: Assignment received from leader for group group3 for generation 378 (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:24:00,554] INFO [GroupMetadataManager brokerId=1001] Removed 0 expired offsets in 2 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-07-15 12:26:11,680] INFO [GroupCoordinator 1001]: Member[group.instance.id None, member.id kafka-python-2.0.1-43259283-cdeb-4738-aaa4-44e83542c7dc] in group group3 has left, removing it from the group (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:26:11,811] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 378 (__consumer_offsets-38) (reason: removing member kafka-python-2.0.1-43259283-cdeb-4738-aaa4-44e83542c7dc on LeaveGroup) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:26:11,812] INFO [GroupCoordinator 1001]: Group group3 with generation 379 is now empty (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:34:00,553] INFO [GroupMetadataManager brokerId=1001] Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-07-15 12:34:30,547] INFO [GroupCoordinator 1001]: Dynamic Member with unknown member id joins group group3 in Empty state. Created a new member id kafka-python-2.0.1-0775e634-fe08-40e9-9ae3-7080ec21dd8a for this member and add to the group. (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:34:30,597] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 379 (__consumer_offsets-38) (reason: Adding new member kafka-python-2.0.1-0775e634-fe08-40e9-9ae3-7080ec21dd8a with group instance id None) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:34:30,613] INFO [GroupCoordinator 1001]: Stabilized group group3 generation 380 (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:34:30,628] INFO [GroupCoordinator 1001]: Assignment received from leader for group group3 for generation 380 (kafka.coordinator.group.GroupCoordinator)
Пример 2 (l oop поколения СО сбросом до 0) журналов Kafka:
...
[2020-07-15 12:10:02,637] INFO [GroupMetadataManager brokerId=1001] Group group3 transitioned to Dead in generation 2 (kafka.coordinator.group.GroupMetadataManager)
[2020-07-15 12:10:02,637] INFO [GroupMetadataManager brokerId=1001] Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-07-15 12:19:22,210] INFO [GroupCoordinator 1001]: Dynamic Member with unknown member id joins group group3 in Empty state. Created a new member id kafka-python-2.0.1-4ecb3653-0db8-4233-9842-710db5270491 for this member and add to the group. (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:19:22,211] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 0 (__consumer_offsets-38) (reason: Adding new member kafka-python-2.0.1-4ecb3653-0db8-4233-9842-710db5270491 with group instance id None) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:19:22,212] INFO [GroupCoordinator 1001]: Stabilized group group3 generation 1 (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:19:22,214] INFO [GroupCoordinator 1001]: Assignment received from leader for group group3 for generation 1 (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:20:02,637] INFO [GroupMetadataManager brokerId=1001] Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-07-15 12:24:22,271] INFO [GroupCoordinator 1001]: Member[group.instance.id None, member.id kafka-python-2.0.1-4ecb3653-0db8-4233-9842-710db5270491] in group group3 has left, removing it from the group (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:24:22,272] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 1 (__consumer_offsets-38) (reason: removing member kafka-python-2.0.1-4ecb3653-0db8-4233-9842-710db5270491 on LeaveGroup) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:24:22,272] INFO [GroupCoordinator 1001]: Group group3 with generation 2 is now empty (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:25:29,398] INFO [GroupCoordinator 1001]: Dynamic Member with unknown member id joins group group3 in Empty state. Created a new member id kafka-python-2.0.1-d927d9d4-1885-4430-a84b-4cc92cb7a04e for this member and add to the group. (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:25:29,398] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 2 (__consumer_offsets-38) (reason: Adding new member kafka-python-2.0.1-d927d9d4-1885-4430-a84b-4cc92cb7a04e with group instance id None) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:25:29,399] INFO [GroupCoordinator 1001]: Stabilized group group3 generation 3 (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:25:29,401] INFO [GroupCoordinator 1001]: Assignment received from leader for group group3 for generation 3 (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:30:02,637] INFO [GroupMetadataManager brokerId=1001] Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-07-15 12:30:29,420] INFO [GroupCoordinator 1001]: Member[group.instance.id None, member.id kafka-python-2.0.1-d927d9d4-1885-4430-a84b-4cc92cb7a04e] in group group3 has left, removing it from the group (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:30:29,420] INFO [GroupCoordinator 1001]: Preparing to rebalance group group3 in state PreparingRebalance with old generation 3 (__consumer_offsets-38) (reason: removing member kafka-python-2.0.1-d927d9d4-1885-4430-a84b-4cc92cb7a04e on LeaveGroup) (kafka.coordinator.group.GroupCoordinator)
[2020-07-15 12:30:29,421] INFO [GroupCoordinator 1001]: Group group3 with generation 4 is now empty (__consumer_offsets-38) (kafka.coordinator.group.GroupCoordinator)
EDIT2: добавить код детали потребителя
self.consumer = KafkaConsumer('topic1',
bootstrap_servers=brokers_env_var,
auto_offset_reset='earliest',
enable_auto_commit=True,
session_timeout_ms=14000,
heartbeat_interval_ms=2000,
group_id='group3',
max_poll_records=100,
key_deserializer=lambda x: loads(x.decode('utf-8')),
value_deserializer=lambda x: loads(x.decode('utf-8')))
try:
for message in self.consumer:
self.consumeMessage(message.value)
except Exception as error:
print('Run error - error : ' + str(error))