kafka-python 1.3.3：带有显式密钥的 KafkaProducer.send 无法向代理发送消息答案

【问题标题】：kafka-python 1.3.3: KafkaProducer.send with explicit key fails to send message to brokerkafka-python 1.3.3：带有显式密钥的 KafkaProducer.send 无法向代理发送消息
【发布时间】：2018-01-10 05:21:28
【问题描述】：

（可能重复Can't send a keyedMessage to brokers with partitioner.class=kafka.producer.DefaultPartitioner，尽管那个问题的 OP 没有提到 kafka-python。无论如何，它从来没有得到答案。）

我有一个 Python 程序已成功（数月）向 Kafka 代理发送消息，基本上使用以下逻辑：

producer = kafka.KafkaProducer(bootstrap_servers=[some_addr],
                               retries=3)
...
msg = json.dumps(some_message)
res = producer.send(some_topic, value=msg)

最近，我尝试将其升级为根据从消息中提取的确定键值将消息发送到不同的分区：

producer = kafka.KafkaProducer(bootstrap_servers=[some_addr],
                               key_serializer=str.encode,
                               retries=3)
...
try: 
    key = some_message[0]
except:
    key = None
msg = json.dumps(some_message)
res = producer.send(some_topic, value=msg, key=key)

但是，使用此代码，不会消息从程序中传递到代理。我已经验证从some_message 提取的键值始终是有效字符串。大概我不需要定义自己的partitioner，因为根据文档：

默认分区器实现使用与 java 客户端相同的 murmur2 算法对每个非 None 键进行哈希处理，以便将具有相同键的消息分配给同一分区。

此外，使用新代码，当我尝试通过调用 res.get（以获取 kafka.FutureRecordMetadata）来确定我的 send 发生了什么时，那个调用会引发 TypeError descriptor 'encode' requires a 'str' object but received a 'unicode' 消息例外。

（作为一个附带问题，如果我真的能够得到它，我不确定我会用FutureRecordMetadata 做什么。基于 kafka-python 源代码，我想我想调用它的succeeded 或failed 方法，但文档没有提及这一点。文档确实说send 的返回值“解析为”RecordMetadata，但是我无法从文档或代码中弄清楚“解析为”在这种情况下的含义。）

无论如何：我不可能是唯一一个使用 kafka-python 1.3.3 的人尝试使用分区键发送消息，而且我在 Intertubes 上没有看到任何描述类似问题的内容（除了 SO 问题我在这篇文章的顶部引用了）。

我当然愿意相信我做错了什么，但我不知道那可能是什么。我需要为KafkaProducer 构造函数提供一些额外的参数吗？

【问题讨论】：

标签： kafka-producer-api kafka-python

【解决方案1】：

事实证明，我的关键值是unicode，尽管我非常确信它是str。因此为我的key_serializer 选择str.encode 是不合适的，并且是导致res.get 异常的原因。省略 key_serializer 并调用 key.encode('utf-8') 足以让我的消息发布并按预期进行分区。

（对我而言）这个问题晦涩难懂的一个重要原因是 kafka-python 1.3.3 documentation 没有详细说明 FutureRecordMetadata 到底是什么，也不应该在其get 方法可以引发的异常方式。文档中唯一的用法示例：

# Asynchronous by default
future = producer.send('my-topic', b'raw_bytes')

# Block for 'synchronous' sends
try:
    record_metadata = future.get(timeout=10)
except KafkaError:
    # Decide what to do if produce request failed...
    log.exception()
    pass

建议它会引发的唯一一种异常是KafkaError，这是不正确的。事实上，get 可以并且将（重新）引发异步发布机制在尝试将消息发送出去时遇到的任何异常。

【讨论】：

【解决方案2】：

我也遇到了同样的错误。一旦我在发送密钥时添加了 json.dumps，它就起作用了。

producer.send(topic="first_topic", key=json.dumps(key)
.encode('utf-8'), value=json.dumps(msg)
.encode('utf-8'))
.add_callback(on_send_success).add_errback(on_send_error)

【讨论】：