时间戳

首页 > 解决方案 > Kafka 一个特定的消费者群体正在重新平衡

apache-kafka - Kafka 一个特定的消费者群体正在重新平衡

问题描述

几个月前，我在我的 Kafka 链中创建了两个消费者组（每个组有 3 个消费者）：

这没有问题。

最近，我添加了第三组以将并行化添加到我的链中：

问题：新主题 3 并没有真正被消耗（在几个小时后消耗了每个主题中的 20 或 50 条消息......）=> 第三组的 3 个分区的每个分区上的巨大“滞后”（从未消耗）。

在 kafka 日志中，我只能看到第三组的许多重新平衡（似乎达到 5 分钟默认 max.poll.interval.ms 但我不确定：kafka 消息未明确指定：准备重新平衡组 group3...） . 我尝试修改轮询选项，减少max.poll.records的数量等。减少后情况会更好，但我继续获得重新平衡...我尝试添加更多 CPU 内核（我以为很多消费者都在我的CPU 服务器）从 1 到 2，最后是 4 个 vcore：没有真正改变，htop 没有显示 CPU 核心的巨大利用率......

我只遇到新消费者组的滞后/重新平衡问题，并且消费代码与其他消费者相同......

我不知道该怎么做......关于kafka conf，kafka CPU的利用率或其他任何建议？

注意：所有堆栈都在 Docker 容器下

谢谢 ！

编辑：kafka日志的示例1（生成循环不重置为0）：

...
[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)

Kafka日志示例2（生成循环重置为0）：

...
[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))

标签： apache-kafkakafka-consumer-apikafka-python

解决方案

消费者组只消费来自 Kafka 的消息，因此您实际上有一个生产者向 topic1、topic3 和两个消费者组发送消息——一个从 topic1 消费，第二个从 topic3 消费。

将相同的数据发送到多个主题没有意义，我建议您仅将这些消息发送到主题1，并使用两个消费者组：

  1. 来自 topic1 的 3 个消费者。
  2. 来自 topic1 的另外 3 个消费者。

Kafka 中不同的消费者组独立管理偏移量，这就是为什么您可以为数据使用单个主题并创建多个消费者组来使用它，否则具有相同数据的两个主题将占用双倍的存储空间，并且可能会有 CPU 开销来管理它们.

请确保同一组中的消费者具有相同的group.id参数。
例如，6 个消费者——其中 3 个具有 group id g1，另外 3 个具有 group id g2。

推荐阅读