Python Kafka 生产者无法写入主题答案

【问题标题】：Python Kafka producer unable to write to topicPython Kafka 生产者无法写入主题
【发布时间】：2020-06-25 21:42:11
【问题描述】：

我正在使用 python kafka producer 来写一个主题。 Kafka 服务器是外部的，我使用 Kerberos 连接凭据。当我运行生产者脚本时，似乎正在建立与服务器的连接，但由于某种原因，元数据请求失败并显示“请求 1：MetadataRequest_v1(topics=NULL)”。

这里是生产者代码：

from kafka import KafkaProducer

broker = ['mybroker.com:9092']

p = KafkaProducer(bootstrap_servers=broker,security_protocol='SSL',ssl_cafile="truststore.pem",ssl_keyfile='truststore.jks',sasl_mechanism='GSSAPI',api_version=(0, 10, 1))
topic = 'test-topic'
# Write hello world to test topic
p.send(topic, bytes("Hello World", 'utf-8'))
p.flush()

这是我目前遇到的错误：

EBUG:kafka.producer.kafka:Starting the Kafka producer
DEBUG:kafka.metrics.metrics:Added sensor with name connections-closed
DEBUG:kafka.metrics.metrics:Added sensor with name connections-created
DEBUG:kafka.metrics.metrics:Added sensor with name select-time
DEBUG:kafka.metrics.metrics:Added sensor with name io-time
INFO:kafka.client:Bootstrapping cluster metadata from [('fakehost.lef.ceg.com', 9092, <AddressFamily.AF_UNSPEC: 0>)]
DEBUG:kafka.client:Attempting to bootstrap via node at mybroker.com:9092
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-sent-received
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name request-latency
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.latency
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=:)>: creating new socket
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=:)=9092>: setting socket option (6, 1, 1)
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=:)/iphere port=9092>: configuring default SSL Context
INFO:kafka.conn:<BrokerConnection node_id=bootstrap host=:)/iphere port=9092>: Loading SSL CA from /truststore.pem
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=:)/iphere port=9092>: wrapping socket in ssl context
INFO:kafka.conn:<BrokerConnection node_id=bootstrap host=:)/iphere port=9092>: connecting to iphere
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=:)/iphere port=9092>: established TCP connection
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=:)/iphere port=9092>: initiating SSL handshake
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=:)/iphere port=9092>: completed SSL handshake.
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=:)/iphere port=9092>: Connection complete.
DEBUG:kafka.client:Node bootstrap connected
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=:)/iphere port=9092> Request 1: MetadataRequest_v1(topics=NULL)
ERROR:kafka.conn:<BrokerConnection node_id=bootstrap host=:)/iphere port=9092>: socket disconnected
INFO:kafka.conn:<BrokerConnection node_id=bootstrap host=:)/iphere port=9092>: Closing connection. ConnectionError: socket disconnected
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=:)/iphere port=9092>: reconnect backoff 0.04487732169774206 after 1 failures
ERROR:kafka.client:Unable to bootstrap from [('mybroker.com', 9092, <AddressFamily.AF_UNSPEC: 0>)]
DEBUG:kafka.metrics.metrics:Added sensor with name bufferpool-wait-time
DEBUG:kafka.metrics.metrics:Added sensor with name batch-size
DEBUG:kafka.metrics.metrics:Added sensor with name compression-rate
DEBUG:kafka.metrics.metrics:Added sensor with name queue-time
DEBUG:kafka.metrics.metrics:Added sensor with name produce-throttle-time
DEBUG:kafka.metrics.metrics:Added sensor with name records-per-request
DEBUG:kafka.metrics.metrics:Added sensor with name bytes
DEBUG:kafka.metrics.metrics:Added sensor with name record-retries
DEBUG:kafka.metrics.metrics:Added sensor with name errors
DEBUG:kafka.metrics.metrics:Added sensor with name record-size-max
DEBUG:kafka.producer.sender:Starting Kafka producer I/O thread.
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.producer.kafka:Kafka producer started
DEBUG:kafka.producer.kafka:Requesting metadata update for topic test-topic
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available

这实际上是什么意思，我该如何调试呢？

【问题讨论】：

嗨，您是否能够使用 kafka-console-producer 生成测试记录，使用您在 python 脚本中使用的 bootstrap-server（可能是集群中的 Kafka 有问题）。你能检查一下 Kafka 日志吗？
我刚试过kafka-console-producer，谢谢你的建议。我做了 bin/kafka-console-producer.sh --broker-list mybroker.com:9092 security.protocol=SSL \ ssl.key.location=truststore.jks \ ssl.certificate.location=./truststore.pem --主题“test-topic”我收到以下错误：[2020-03-14 00:02:37,584] WARN [Producer clientId=console-producer] Bootstrap broker mybroker.com:9092 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
mybroker.com:9092 可能没有运行代理
你不能像这样在 CLI 上设置安全协议。您需要使用配置文件。另外，python 也需要证书
您的配置有问题。您将security_protocol 设置为SSL，但您也设置了sasl_mechanism='GSSAPI'。你想使用 SASL 还是只使用 SSL？

标签： python apache-kafka kerberos producer

【解决方案1】：

使用以下代码使用python启动生产者

from kafka import KafkaProducer
import time
import json

topic = 'testing'
bootstrap_servers = '<your broker>:9092'
producer = KafkaProducer(bootstrap_servers=bootstrap_servers)
msg=json.dumps({"status":True})  # update your message here
future = producer.send(topic, msg.encode('utf-8'))
result = future.get(timeout=60)
metrics = producer.metrics()
print(metrics)

消息格式独立，可以使用任意格式发送消息

【讨论】：