只读 kafka 主题中的新消息

Question

我正在 python 中使用 confluent-kafka 创建一个消费者，我想以这样一种方式创建它，如果消费者重新启动，它会从主题中的最后一条可用消息（每个分区）开始，它不会不管它是否在没有提交的情况下留下消息。

这是为了避免处理数以百万计的消息，这些消息在消费者关闭时生成并且不再需要处理。

我尝试设置参数 auto.offset.reset 的不同选项，但最多从上次提交的偏移量开始。这是我的配置：

consumer = Consumer({"bootstrap.servers": "localhost:9092",
                     "group.id": group_id,
                     "auto.offset.reset": "latest",
                     "isolation.level": "read_committed",
                     "default.topic.config": {"enable.auto.commit": False}})

有没有办法实现这种行为？

注意：我可能有多个消费者，但没有一个手动分配给特定分区

Answer 1

auto.offset.reset 配置仅在没有提交的偏移量时应用。

如果您想始终从头开始，您可以使用enable.auto.commit=false 禁用自动提交（并确保也不要显式提交），并将auto.offset.reset 设置为latest。

另一种选择是使用get_watermark_offsets() 和seek() 的组合将分区分配给消费者（使用on_assigned()）时显式搜索到末尾