Kafka Connect offset.storage.topic 未收到消息（即如何访问 Kafka Connect 偏移元数据？）答案

【问题标题】：Kafka Connect offset.storage.topic not receiving messages (i.e. how to access Kafka Connect offset metadata?)Kafka Connect offset.storage.topic 未收到消息（即如何访问 Kafka Connect 偏移元数据？）
【发布时间】：2017-05-20 10:16:18
【问题描述】：

我正在设置一个 Kafka Connect 分布式模式应用程序，该应用程序将是一个 Kafka 到 S3 管道。我正在使用 Kafka 0.10.1.0-1 和 Kafka Connect 3.1.1-1。到目前为止，事情进展顺利，但对于我正在使用的大型系统而言，重要的一个方面需要了解 Kafka -> FileSystem 管道的偏移信息。根据文档，offset.storage.topic 配置将是分布式模式应用程序用于存储偏移信息的位置。考虑到 Kafka 如何在“新”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

我在让这个命令运行一分钟左右后收到这条消息：

Processed a total of 0 messages

总结一下，我有两个问题：

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

感谢您的帮助。

【问题讨论】：

检查消费者是否阅读主题并且没有失败。因为消费者不仅期望消息，而且期望格式正确的消息（标头中有 5 个字节，其中 4 个是模式注册表中模式的 Id）。检查这个：stackoverflow.com/a/41285681/1437693
感谢您的帮助@YuriTceretian。这与 Kafka Connect 有关，它有自己的内置消费者。
Python Kafka 是否兼容 Connect 的消费者？
@YuriTceretian 我不确定。我将在星期一继续我的工作。 Kafka 开发人员在 Kafka 用户组中回复了我的查询，但一旦我完成调查，我将发布答案。

标签： apache-kafka apache-kafka-connect

【解决方案1】：

Liju 是正确的，connect-offsets 用于跟踪源连接器（有生产者但没有消费者）的偏移量。 Sink 连接器有一个消费者并以通常的方式跟踪偏移量 - __consumer_offsets 主题
查看最后提交的偏移量的最佳方法是使用消费者组工具：

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

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

【讨论】：

嗨 Gwen，我们如何使用 Java API 实现读取消费者偏移量。我需要在仪表板中显示它吗？到目前为止，我们正在使用您提到的上述 shell 命令。
您可以查看我们在实现消费者组命令行时编写的代码并了解主要思想......（这就是开源摇滚的原因，对吧？）
感谢您的澄清。您的答案应在文档中提及；源连接器和接收器连接器以不同的方式存储偏移量，并且接收器连接器（即具有 kafka 作为源）使用具有消费者组的常规 kafka 消费者。现在很容易监控我们设置的 kafka-to-s3 管道的延迟。

【解决方案2】：

偏移量可能正在提交到 kafka 默认偏移量提交主题，即 _consumer_offsets

【讨论】：

【解决方案3】：

Confluent 发布的新S3 Connector 可能会引起您的兴趣。

根据您的描述，也许它可以显着简化您将记录从 Kafka 导出到 S3 存储桶的目标。

【讨论】：