Kafka：max.poll.interval.ms 设置为 Integer.MAX_VALUE 但应用程序仍然超时

Question

我正在使用 kafka-clients 2.5.1 并且有一个具有以下属性的 kafka 流应用程序：

properties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, "1");
properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, String.valueOf(Integer.MAX_VALUE));
properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
properties.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "3000");

处理一条记录平均需要 30 分钟，但每次之后，都会出现以下错误，并且应用程序开始重新读取它刚刚处理的消息：

02-07-2021 02:09:31.026 [employee_groups-client-StreamThread-1] WARN  
org.apache.kafka.streams.processor.internals.StreamThread.runLoop - stream-thread 
[employee_groups-client-StreamThread-1] Detected task 0_0 that got migrated to another thread. 
This implies that this thread missed a rebalance and dropped out of the consumer group. Will 
try to rejoin the consumer group. Below is the detailed description of the task:
>TaskId: 0_0
>>  ProcessorTopology: 
>       KSTREAM-SOURCE-0000000000:
>           topics:     [us4lat-employee_group_migration1]
>           children:   [KSTREAM-FOREACH-0000000001]
>       KSTREAM-FOREACH-0000000001:
>Partitions [us4lat-employee_group_migration1-0]
org.apache.kafka.streams.errors.TaskMigratedException: Client request for task 0_0 has been 
fenced due to a rebalance
at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:530) ~ 
[kafka-streams-2.5.1.jar!/:?]
at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:478) ~ 
[kafka-streams-2.5.1.jar!/:?]
at org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:226) ~ 
[kafka-streams-2.5.1.jar!/:?]
at org.apache.kafka.streams.processor.internals.TaskManager.commitAll(TaskManager.java:543) ~ 
[kafka-streams-2.5.1.jar!/:?]
at 
org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:977) ~ 
[kafka-streams-2.5.1.jar!/:?]
at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:823) ~ 
[kafka-streams-2.5.1.jar!/:?]
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:697) 
[kafka-streams-2.5.1.jar!/:?]
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:670) 
[kafka-streams-2.5.1.jar!/:?]
Caused by: 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 max.poll.interval.ms or by    
reducing the maximum size of batches returned in poll() with max.poll.records.

如果最大间隔设置为 Integer.MAX_VALUE，那么为什么在记录处理完成后每隔 30 多分钟就会出现上述异常？我怎样才能解决这个问题？我错过了什么吗？ 我还从 Kafka-Stream 日志中验证了 max.poll.interval.ms 的值设置是否正确。

Answer 1

我认为这几乎可以肯定是这个溢出错误的结果：KAFKA-13126

通常，当max.poll.interval.ms 设置为 MAX_VALUE 时，成员不应错过重新平衡，因为组协调员将等待所有成员重新加入组的时间。但是这个joinGroupTimeout实际上被计算为max.poll.interval.ms + JOIN_GROUP_TIMEOUT_LAPSE——它增加了额外的5秒，让消费者在点击max.poll.interval.ms并再次调用poll()之后有一些回旋的空间来将他们的JoinGroup请求发送给代理。

不幸的是，当间隔已经设置为MAX_VALUE时，这当然会溢出。在这种情况下，joinGroupTimeout 被固定为request.timeout.ms，默认为 30 秒。因此，一旦出于某种原因触发了重新平衡，消费者只有 30 多岁的时间重新加入该组。但由于轮询间隔太长，他们只有在处理完记录后才会进行轮询。

通常情况可能不会那么糟糕，因为批次通常足够小，可以在几秒钟内处理完毕，而且消费者最终需要在 30 秒内进行 poll()。如果每条记录需要 30 分钟来处理，就像您的情况一样，那么就没有这样的运气——几乎可以肯定，在这 30 秒内不会 poll() 。更糟糕的是，一旦它最终确实调用 poll()，它会意识到自己退出并请求重新加入，从而触发新的重新平衡。这可能会导致另一个成员退出，然后谁会在轮询时触发另一个重新平衡，等等。

幸运的是，我为此打开了一个修复程序，解决方法很简单：只需减少 max.poll.interval.ms。将其设置为MAX_VALUE - 5000 应该就足够了，但您可能想尝试将其再降低几个档位以使其更有用。像 2 * max_time_process_record * max.poll.records 这样的东西——如果出现问题并且消费者在处理过程中由于某种原因卡住了，那么组协调员会注意到并将其踢出组，以便另一个消费者可以继续工作它的分区。无需等待 Integer.MAX_VALUE 毫秒过去，即。