Kafka消费者离开消费群体

Question

我在使用Kafka时遇到了一些问题。非常感谢任何帮助！我在docker swell中分别有zookeeper和kafka集群3个节点。您可以在下面看到Kafka代理配置。

  KAFKA_DEFAULT_REPLICATION_FACTOR: 3
  KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
  KAFKA_MIN_INSYNC_REPLICAS: 2
  KAFKA_NUM_PARTITIONS: 8
  KAFKA_REPLICA_SOCKET_TIMEOUT_MS: 30000
  KAFKA_REQUEST_TIMEOUT_MS: 30000
  KAFKA_COMPRESSION_TYPE: "gzip"
  KAFKA_JVM_PERFORMANCE_OPTS: "-XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:G1HeapRegionSize=16M -XX:MinMetaspaceFreeRatio=50 -XX:MaxMetaspaceFreeRatio=80"
  KAFKA_HEAP_OPTS: "-Xmx768m -Xms768m -XX:MetaspaceSize=96m"

我的情况：

< li > 20x位制片人不断向Kafka主题传达信息 < li>1x消费者读取和记录消息 < li >终止kafka节点(docker容器停止)，因此现在群集有2个Kafka代理节点(第3个节点将自动启动并加入群集) < li >消费者不再使用消息，因为它由于重新平衡而离开了消费者群体

是否存在任何机制来告诉消费者在重新平衡后加入群组？

日志：

 INFO 1 --- [ | loggingGroup] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=kafka-consumer-0, groupId=loggingGroup] Attempt to heartbeat failed since group is rebalancing
 WARN 1 --- [ | loggingGroup] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=kafka-consumer-0, groupId=loggingGroup] This member will leave the group because consumer poll timeout has expired. This means 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 processing messages. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.

Answer 1

@rostyslav每当我们打电话让消费者阅读消息时，它都会打两个主要电话。

1. 投票
2. 提交

轮询基本上是从kafka主题中获取记录，并提交告诉kafka将其保存为已读消息，这样就不会再次读取。虽然轮询几个参数起着主要作用：

< li >最大轮询记录数 < li >最大轮询间隔毫秒数

仅供参考：变量名称是每个python api的。

因此，每当我们尝试读取来自消费者的消息时，它每max_poll_interval_ms进行一次轮询调用，并且仅在处理获取的记录（如max_poll_records中定义）后才会进行相同的调用。因此，无论何时，max_poll_records未在max_poll_inetrval_ms中处理，我们都会收到错误。

为了克服这个问题，我们需要改变这两个变量中的一个。更改max_poll_interval_ms可能会很忙，因为有时处理某些记录可能需要更长的时间，有时则需要更少的记录。我总是建议使用max_poll_records来解决这个问题。这对我有用。