apache-kafka - 在日志末尾时，如何停止尝试从 Kafka 中消费消息？

Question

我有一个按计划创建的 Kafka 消费者。它尝试使用自上次提交以来添加的所有新消息。

我想在消耗完日志中的所有新消息后关闭使用者，而不是无限期地等待新消息进入。

我无法通过 Kafka 的文档找到解决方案。

我在 Confluent.Kafka.ConsumerConfig 和 ClientConfig 类中看到许多与超时相关的属性，包括 FetchWaitMaxMs，但我无法破译使用哪个。我正在使用 .NET 客户端。

任何意见，将不胜感激。

Answer 1

我找到了解决办法。 Confluent 的 .NET Kafka 库的 1.0.0-beta2 版本提供了一种名为 .Consume(TimeSpan timeSpan) 的方法。 .如果没有要使用的新消息或者我们在分区 EOF，这将返回 null。我以前使用的是 .Consume(CancellationToken cancellationToken)过载阻塞并阻止我关闭消费者。更多信息:https://github.com/confluentinc/confluent-kafka-dotnet/issues/614#issuecomment-433848857

另一种选择是升级到版本 1.0.0-beta3，它在 ConsumeResult 对象上提供一个名为 IsPartitionEOF 的 bool 标志。这就是我最初寻找的 - 一种知道何时到达分区末尾的方法。