Avro，Kafka 消息键插入无关字符答案

【问题标题】：Avro, Kafka message keys inserting extraneous charactersAvro，Kafka 消息键插入无关字符
【发布时间】：2019-12-09 17:26:40
【问题描述】：

我的主题是这样构建的

producer = AvroProducer(
            config.confluent.connection_config(config.confluent.CONNECTION_MODE_PRODUCER),
            default_value_schema=get_avro_schema_object(),
            default_key_schema=avro.loads('{"type": "string"}'))

producer.produce(topic=topic, value=d, key=str(d['identifier']))

一切看起来都很好，但是当我发布时，我会得到这样的密钥

print 'build_pipe_dev';
Format:AVRO
12/9/19, 12:04:23 PM EST, +$something-unique-1, {"identifier": "something-unique-1", "event_type": "build_queued", "build_request": null, "build_queued": {"jobid": "job.torque.yup"}, "build_started": null, "build_finished": null, "static_analyze_request": null}

请看，在我的消息键字段中，不是something-unique-1，而是+$something-unique-1。我整个早上都在为此挠头，我意识到它的 avro 将其 mojo 插入到序列化器/反序列化器的字段中。问题是，当我不包含密钥架构时，confluent_kafka 库会因此在生产函数中引发错误

if key is not None:
            if key_schema:
                key = self._serializer.encode_record_with_schema(topic, key_schema, key, True)
            else:
                raise KeySerializerError("Avro schema required for key")

我在其他帖子中看到人们猴子修补了confluent_kafka 库，但是有没有“正确”的方法来解决这个问题？

我想，还有一件事要考虑，它只是正常工作而我反应过度了吗？

【问题讨论】：

标签： python apache-kafka avro ksqldb

【解决方案1】：

KSQL 目前只支持 STRING 键。它不会正确反序列化您的 Avro，因此您在关键字段中看到的奇怪字符。

例如，如果您使用 kafkacat，您将能够使用您的消息并正确查看 Avro 密钥。

【讨论】：

那么，这是否意味着一切正常？只要架构永远不会改变（它不应该改变）我认为我很好。
取决于您是否依赖 KSQL 处理中的键。如果你需要使用字符串键，要么在源代码（正如你所说的那样，这在 Python 客户端是不可能的，因为它要么是/或目前是 IIRC），要么是通过在 KSQL 中重新键入数据（@987654321 @)