apache-kafka - Kafka Connect offset.storage.topic 没有收到消息(即如何访问 Kafka Connect 偏移量元数据？)

Question

我正在设置 Kafka Connect 分布式模式应用程序，它将成为 Kafka 到 S3 的管道。我正在使用 Kafka 0.10.1.0-1 和 Kafka Connect 3.1.1-1。到目前为止，一切进展顺利，但对我正在使用的大型系统来说很重要的一个方面需要了解 Kafka -> FileSystem 管道的偏移信息。根据文档，offset.storage.topic 配置将成为分布式模式应用程序用于存储偏移量信息的位置。考虑到 Kafka 如何在"new"Kafka 中存储消费者偏移量，这是有道理的。但是，在使用 FileStreamSinkConnector 进行一些测试后，没有任何内容写入我的 offset.storage.topic，这是默认值:connect-offsets。

具体来说，我使用 Python Kafka 生产者将数据推送到主题，并使用 Kafka Connect 和 FileStreamSinkConnect 将数据从主题输出到文件。它的工作和行为与我期望连接器的行为一样。此外，当我停止连接器并启动连接器时，应用程序会记住主题中的状态并且没有数据重复。但是，当我转到 offset.storage.topic 查看存储了哪些偏移元数据时，主题中没有任何内容。

这是我使用的命令:

kafka-console-consumer --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 --topic connect-offsets --from-beginning

让此命令运行一分钟左右后，我收到此消息:

总共处理了 0 条消息

总而言之，我有两个问题:

1. 为什么即使我的分布式应用程序正确地保持状态，偏移元数据也没有写入应该存储它的主题？
2. 如何访问 Kafka Connect 分布式模式应用程序的偏移量元数据信息？这对于我的团队的系统的 Lambda 架构实现是 100% 必要的。

感谢您的帮助。

Answer 1

1. Liju 是正确的，connect-offsets 用于跟踪源连接器(有生产者但没有消费者)的偏移量。接收器连接器有一个消费者并以通常的方式跟踪偏移量 - __consumer_offsets 主题

2. 查看最后提交的偏移量的最佳方式是使用消费者组工具:

   bin/kafka-consumer-groups.sh --group connect-elastic-login-connector --bootstrap-server localhost:9092 --describe

组名始终是“connect-”和连接器名称(在我的例子中是 elastic-login-connector)。这将显示该组提交的最新偏移量，这基本上承认所有达到此偏移量的消息都已写入 Elastic。