Kafka python优雅关闭消费者答案

【问题标题】：Kafka python graceful shutdown of consumerKafka python优雅关闭消费者
【发布时间】：2020-08-11 22:41:09
【问题描述】：

我正在尝试优雅地关闭 kafka 消费者，但脚本会因 Stopping HeartBeat 线程而阻塞。如何使用 kafka-python 在 SIGTERM 上优雅地关闭消费者。这就是我所做的

import logger as logging
import time
import sys
from kafka import KafkaConsumer
import numpy as np
import signal


log = logging.getLogger(__name__)


class Cons:
    def __init__(self):

        signal.signal(signal.SIGINT, self.sigterm_handler)
        signal.signal(signal.SIGTERM, self.sigterm_handler)
        self.consumer = KafkaConsumer('dummy-topic', group_id='poll-test', bootstrap_servers=['b1'])

    def sigterm_handler(self, signum, frame):
        log.info("Sigterm handler")
        self.consumer.close(autocommit=False)
        sys.exit(0)



    def consume(self):
        try:
            while True:
                records = self.consumer.poll(timeout_ms=500, max_records=500)
                for topic_partition, consumer_records in records.items():
                    for record in consumer_records:
                        log.info("Got Record - {}".format(record))
                    #code to manually commit


        except ValueError as e:
            log.exception("exception")


if __name__ == '__main__':
    c=Cons()
    c.consume()

启用调试日志后，这是我得到的输出，代码在此被阻止。

^C2020-04-28 07:18:33,050 - MainThread - __main__ - INFO - Sigterm handler
2020-04-28 07:18:33,050 - MainThread - kafka.consumer.group - DEBUG - Closing the KafkaConsumer.
2020-04-28 07:18:33,051 - MainThread - kafka.coordinator - INFO - Stopping heartbeat thread

这背后的原因是什么？在 SIGTERM 或 SIGINT 上关闭消费者的正确方法是什么？

【问题讨论】：

您可能需要更新 config/server.properties 文件中的“controlled.shutdown.enable=true”。
但这不适合经纪人吗？这是消费者

标签： python apache-kafka python-3.7 kafka-python

【解决方案1】：

看起来这个问题最终在 2020 年得到了修复，所以任何最新版本都应该没问题：

https://github.com/dpkp/kafka-python/commit/91daea329bb40ed80bddef4635770d24b670b0c6

【讨论】：