AWS MSK Connect 将数据插入 s3，无需 {topic_name} 的附加文件夹

Question

我正在使用 AWS MSK 连接并发送到 AWS s3 来使用 kafka 主题（例如 topic1）。

以下是简化的 AWS MSK 连接器配置..

topics.dir=folder1/folder2
topics=topic1

该主题可以按预期被消费并发送到 s3。

唯一的问题是在s3存储桶下创建了一个额外的文件夹{topic_name}，即folder1/folder2/topic1。

我的问题是： 有没有办法禁止每个 topic_name 创建新文件夹？

非常感谢任何帮助。

Answer 1

假设您使用 Confluence S3 接收器连接器，则无法避免将主题名称作为对象路径的一部分。

这是连接器的代码：

  private String fileKeyToCommit(String dirPrefix, long startOffset) {
    String name = tp.topic()
                      + fileDelim
                      + tp.partition()
                      + fileDelim
                      + String.format(zeroPadOffsetFormat, startOffset)
                      + extension;
    return fileKey(topicsDir, dirPrefix, name);
  }

如您所见，连接器无条件地将主题名称添加到对象的路径中。

作为解决方法，您可以在 S3 上创建新对象时启用事件，并在 Lambda 中重命名对象。