apache-kafka - 防止kafka消费者长时间超时

Question

我需要防止 kafka 消费者在应用程序等待特定进程完成时超时。我的方法是暂停分区，然后在该过程完成后恢复它们。

List<TopicPartition> partitionList = new ArrayList<>(); 
partitionList.addAll(kafkaConsumer.assignment());
kafkaConsumer.pause(partitionList);

while(//waiting for the process to complete){
    Thread.sleep(10000);                            
    kafkaConsumer.poll(0);
}
kafkaConsumer.resume(partitionList);

问题

暂停会自动将心跳发送到kafka，还是我仍然需要定期轮询以发送心跳？

我的方法是最好的吗？或者有更好的方法吗？

Answer 1

从 Kafka 0.10.1 开始，消费者确实有一个用于发送心跳的后台线程:https://cwiki.apache.org/confluence/display/KAFKA/KIP-62%3A+Allow+consumer+to+send+heartbeats+from+a+background+thread

因此，您无需调用 poll()向经纪人发送心跳。但是，有第二个超时 max.poll.interval.ms -- 您必须调用 poll()在这段时间内避免第二次超时。默认值为 5 分钟。如果您的等待时间比这更长，您可以增加此超时时间。如果这样做，您也不需要暂停任何分区等。

如果您使用的是旧版本，您可以暂停并调用 poll()定期是发送定期心跳以避免超时的唯一方法。