同步 kafka 生产者发送仍然有 0 秒超时？答案

【问题标题】：Sync kafka producer send still got 0 secs timeout?同步 kafka 生产者发送仍然有 0 秒超时？
【发布时间】：2020-11-10 06:31:42
【问题描述】：

我在 Python 脚本中有以下代码

from kafka import KafkaProducer

kafka_producer = KafkaProducer(....)

kafka_producer.send(topic, value=message)
kafka_producer.flush()

logger.info('Done!') # this message is displayed

但是，我仍然看到以下消息。看来消息已成功发送。为什么即使调用flush()，也会显示“0秒超时”的消息？

INFO:root:完成！

INFO:kafka.producer.kafka:以0秒超时关闭Kafka生产者。

INFO:kafka.producer.kafka:Proceeding to force close the producer 因为挂起的请求无法在超时 0 内完成。

INFO:kafka.conn:：正在关闭连接。

【问题讨论】：

标签： python apache-kafka

【解决方案1】：

我认为在这种情况下错误消息是不正确的。相关代码-source on github

        if timeout > 0:
            if invoked_from_callback:
                log.warning("Overriding close timeout %s secs to 0 in order to"
                            " prevent useless blocking due to self-join. This"
                            " means you have incorrectly invoked close with a"
                            " non-zero timeout from the producer call-back.",
                            timeout)
            else:
                # Try to close gracefully.
                if self._sender is not None:
                    self._sender.initiate_close()
                    self._sender.join(timeout)

        if self._sender is not None and self._sender.is_alive():
            log.info("Proceeding to force close the producer since pending"
                     " requests could not be completed within timeout %s.",
                     timeout)
            self._sender.force_close()

我对代码的解释：如果timeout 是0，我们会跳过正常关闭代码并直接进行强制关闭。在记录之前在_sender 上检查的唯一条件是它存在，并且它是is_alive()。当然是存在的，而且它还活着，因为它还没有被告知要关闭。

如果timeout 是0，它永远不会检查是否可以完成任何事情。所以在这种情况下记录是不正确的。

在timeout > 0 的情况下，日志记录是有意义的。 initiate_close() 被调用，join() 也被调用，这表示您只能通过稍后检查 is_alive() 才能知道加入是否成功。如果在尝试join() 之后它仍然存在，则强制关闭它并且请求无法在超时内完成。

【讨论】：