python - KafkaTimeoutError : Failed to update metadata after 60. 0 秒

Question

我有一个高吞吐量 kafka 生产者的用例，我想每秒推送数千条 json 消息。

我有一个 3 节点的 kafka 集群，我使用的是最新的 kafka-python 库，并有以下方法来生成消息

def publish_to_kafka(topic):
    data = get_data(topic)
    producer = KafkaProducer(bootstrap_servers=['b1', 'b2', 'b3'],
                             value_serializer=lambda x: dumps(x).encode('utf-8'), compression_type='gzip')
    try:
        for obj in data:
           producer.send(topic, value=obj)
    except Exception as e:
            logger.error(e)
    finally:
        producer.close()

我的主题有 3 个分区。

方法有时会正常工作，但会失败并显示错误“KafkaTimeoutError:60.0 秒后无法更新元数据”。

我需要更改哪些设置才能使其顺利运行？

Answer 1

如果某个主题不存在，而您正在尝试生成该主题，并且自动主题创 build 置为 false，则可能会发生这种情况。

可能的分辨率:在代理配置 (server.properties) 中 auto.create.topics.enable=true (注意，这是 Confluent Kafka 中的默认设置)

另一种情况可能是网络拥塞或速度，如果使用 Kafka 代理更新元数据的时间超过 60 秒。

可能的分辨率:生产者配置:max.block.ms = 1200000 (120 秒，例如)

检查您的代理是否由于某种原因(例如，负载过多)而关闭，以及为什么他们无法响应元数据请求。通常，您可以在 server.log 文件中看到它们。