问题标签 [s3-kafka-connector]

我已启用

但我没有看到 s3 存储中的标题和键。仅存储值部分。

任何输入都会有很大帮助。

我想实现一个自定义 s3 分区器类以包含一些 avro 消息字段和一些额外的逻辑来生成输出 s3 路径前缀

该项目在 kotlin 中，这是我的课程：

起初我尝试通过创建自定义 shadowJar 任务来生成 Jar 文件：

但这样做jar -tvf filename.jar表明它只包含我自己的代码，而 kafka 连接失败了java.lang.ClassNotFoundException: io.confluent.connect.storage.partitioner.TimeBasedPartitioner。我认为您不应该在自定义 jar 中包含 kakfa 连接代码，还因为如果我尝试使用它配置任务，TimeBasedPartitioner那么该类可用。

然后我尝试通过将自定义 jar 定义更改为：

不幸的是，这包括我所有的应用程序代码，但我可以看到分区程序包含在 jar 文件中。

Kafka 连接现在失败并出现此错误：

更新：我还尝试通过覆盖公共方法来改变函数的覆盖方式encodePartition，但没有改变。

我也尝试添加这样的测试（希望应该尝试强制转换为 Partitioner`：

没有失败

我在 MSK Connect 上部署了一个 s3 连接器，在 github 上有一个带有json连接器配置文件的存储库。我想通过 MSK 的 REST API 按需更新连接器配置。我检查了API 文档，但似乎UpdateConnectorAPI 只允许修改容量配置。API 确实允许提供连接器配置，但如果连接器已经存在，CreateConnector它会返回错误。我可以删除然后重新创建连接器，但这似乎不是一个好方法。

是否有另一种方法来更新正在运行的连接器配置？

Confluent 的 Kafka Connect 的 S3 Sink 连接器可以使用 将主题写入 S3 存储桶中的嵌套（不是顶级）文件夹topics.dir吗？

例如，如果我设置topics.dir为 value thisistoplevel/thisisnested，连接器会工作吗？

配置属性的文档说：topics.dir

用于存储从 Kafka 摄取的数据的顶级目录。

但这似乎是一个奇怪的限制。也许“顶级目录”一词的意思更像是“将在其下编写主题的最顶层目录”。

也许其他人已经对此进行了测试或在生产中使用了嵌套目录？