Kafka连接S3源连接器忽略键

Question

我在 AWS S3 中备份了以下文件，这些文件由 Kafka 连接接收器连接器备份：

test_yordan_kafka_connect+0+0000023000.json
test_yordan_kafka_connect+0+0000023000.keys.json

当使用Kafka connect S3源恢复主题时，密钥文件被忽略，我在日志中看到以下调试消息:

DEBUG [source-s3|task-0] Removing test-bucket/topics/test_yordan_kafka_connect/partition=1/test_yordan_kafka_connect+1+0000000000.keys.json from common prefixes. (io.confluent.connect.s3.source.S3Storage:333)

我的源配置如下所示：

"connector.class":"io.confluent.connect.s3.source.S3SourceConnector",
"tasks.max":"1",
"s3.region":"eu-central-1",
"s3.bucket.name":"test-bucket",
"topics.dir":"test-bucket/topics",
"format.class": "io.confluent.connect.s3.format.json.JsonFormat",
"partitioner.class":"io.confluent.connect.storage.partitioner.DefaultPartitioner",
"schema.compatibility":"NONE",
"confluent.topic.bootstrap.servers": "blockchain-kafka-kafka-0.blockchain-kafka-kafka-headless.default.svc.cluster.local:9092",
"transforms":"AddPrefix",
"transforms.AddPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",
"transforms.AddPrefix.regex":".*",
"transforms.AddPrefix.replacement":"$0_copy

我应该做什么改变才能让密钥和消息一起存储在Kafka中？

Answer 1

我忽略了如何确保在从S3读取数据时考虑到.keys.json来构造Kafka键。

不过，我用来实现同样的事情的一个技巧是简单地依赖ValueToKeySMT转换器：如果键值也是值有效负载的一部分，那么将其注入键值就像在源连接器配置中添加org.apache.kafka.connect.transforms.ValueToKey转换器的实例一样简单。

下面是相关的文档:https://docs . confluent . io/platform/current/connect/transforms/value tokey . html

这里有一篇关于变压器的很棒的博客文章（12篇同样很棒的文章的一部分）https://rmoff.net/2020/12/09/twelve-days-of-smt-day-2-valuetokey-and-extractfield/

在我的具体情况下，ValueToKey的内置行为并不完全符合我的要求，所以我最终编写了我自己的CUomKeyToValue转换器，由我的接收器连接器使用，以确保密钥存储在S3中，然后在源连接器上编写CUomValueToKey以将内容重建到Kafka中。

编写这样的自定义转换器实际上非常简单，它基本上只是在单个 java 文件中编写 20 行左右的代码，然后您可以使用 kafka-connect-maven-plugin 打包。例如，以下是可以作为灵感的内置 ValueToKey 的代码：

https://github . com/Apache/Kafka/blob/28f 013708 FFE 8 e 48 e 46 f 408 c 7 f 570 BF 2c D5 c 54 b 2/connect/transforms/src/main/Java/org/Apache/Kafka/connect/transforms/value tokey . Java/Java/org/Apache/Kafka/connect/transforms/value tokey . Java

我们也可以应用相同的原理来保存/检索Kafka标题。