KafkaConsumer poll() 行为理解答案

【问题标题】：KafkaConsumer poll() behavior understandingKafkaConsumer poll() 行为理解
【发布时间】：2020-01-29 20:42:36
【问题描述】：

试图了解（kafka 新手）kafka 中的 poll 事件循环是如何工作的。

用例：关于主题的 25 条记录，最大投票大小设置为 5。

max.poll.interval.ms = 5000 //5 seconds by default max.poll.records = 5

任务顺序

从主题中轮询记录。
在 for 循环中处理记录。
一些处理登录，其中逻辑将通过或失败。
如果逻辑通过（带偏移）将被添加到地图中。
然后它将使用 commitSync 调用提交。
如果失败，则循环将中断，并且在此之前的任何成功都将被提交。问题在此之后开始。
即使出现错误，下一次民意调查仍会以 5 个批次继续进行，这是预期的吗？

我们基本上期望的是循环中断并且偏移直到成功过程消息逻辑应该被提交，然后下一个轮询应该从失败的消息继续。

例如，第一批轮询 5 条消息已轮询，1,2 个偏移量成功并提交，然后第三批失败。因此，如果我们预期之间有任何错误，轮询调用将继续移动到下一批，如 5-10,10-15它在该点停止并且轮询应该在第一种情况下从 3 开始，或者如果它在 8 的第二批中失败，那么下一次轮询应该从第 8 个偏移量开始，而不是从下一个最大轮询批处理设置开始，在这种情况下就像 5。如果使用 SPRING BOOT PROJECT 并启用自动提交是错误的。

我已尝试在文档中找到此内容，但没有任何帮助。

尝试对此进行调整，但没有帮助 max.poll.interval.ms

编辑：不接受答案，因为没有针对客户消费者的直接解决方案。保留此信息以供参考

【问题讨论】：

标签： spring-boot apache-kafka kafka-consumer-api kafka-producer-api

【解决方案1】：

max.poll.interval.ms 是毫秒，而不是秒，所以应该是 5000。

一旦轮询返回记录（并且偏移量未提交），它们将不会再次返回，除非您重新启动消费者或对消费者执行seek() 操作以将偏移量重置为未处理的记录。

Apache Kafka 项目的 Spring 提供了一个 SeekToCurrentErrorHandler 来为您执行此任务。

如果您自己使用消费者（听起来很像），则必须进行搜索。

【讨论】：

对不起，我的意思是 5000 毫秒。你是对的，我们自己使用消费者并进行民意调查、处理和提交。我确实尝试了 seek() 选项，但这对我来说似乎是一个黑客行为。当我们从最后提交的偏移量重新开始轮询时，当出现错误时应该不会发生同样的情况，并且轮询应该指向下一个最大轮询批次中开始的最后提交的偏移量。而且很难找到对这种行为的引用，让它听起来像是一个真正的修复而不是一个黑客:)
这种行为在 Apache Kafka 和 Confluent Kafka 中也有不同吗？在每次轮询之前调用 seek 将需要一个新变量来保存当前成功提交的偏移量并在每个轮询周期中进行搜索，这不会影响性能吗？
>should not the same happen when there is a error 不，不应该。 poll() 对您处理记录时的任何错误一无所知——它只知道发送给消费者的最后一批记录的位置；因此，如果你想倒带，你必须执行搜索。没有要求在下一次轮询之前提交轮询的偏移量。您不需要在每次民意调查中寻找，只有当您有错误并且需要重播该记录（以及同一批次中剩余的任何记录）时。行为与 Confluent 相同。
当我们不使用 ConcurrentKafkaListenerContainerFactory 或 KafkaMessageListenerContainer 时如何使用这个 .setErrorHandler(new SeekToCurrentErrorHandler())，我们使用的是 KafkaConsumer 它没有任何设置的错误处理程序。
你不能——因为你没有使用 spring-kafka 而你自己使用的是KafkaConsumer，你必须自己实现类似的逻辑。我只是举个例子。

【解决方案2】：

您可以在失败时手动寻找所有已分配分区的轮询开始偏移量。我不确定使用弹簧消费者。

为普通消费者寻找偏移量的示例代码。在下面的代码中，我获取每个分区的记录列表，然后获取要查找的第一条记录的偏移量。

def seekBack(records: ConsumerRecords[String, String]) = {
    records.partitions().map(partition => {
      val partitionedRecords = records.records(partition)
      val offset = partitionedRecords.get(0).offset()
      consumer.seek(partition, offset)
    })
  }

在生产中这样做的一个问题是不好的，因为您不希望仅在出现暂时错误的情况下一直进行回溯，否则您最终将无限重试。

【讨论】：