Spring cloud stream-Kafka消费者使用StreamListener消费重复消息

Question

在我们的spring boot应用程序中，我们注意到Kafka消费者偶尔会在prod env中随机消费两次消息。我们在PCF中部署了6个实例和6个分区。我们发现在同一主题中收到两次具有相同偏移量和分区的消息，这会导致重复，对我们来说是业务关键。我们在非生产环境中没有注意到这一点，在非生产环境中很难复制。我们最近转向Kafka，但我们无法找到根本问题。

我们使用的是spring cloud stream/spring cloud stream活页夹kafka-2.1.2，配置如下：

spring:
  cloud:
    stream:
      default.consumer.concurrency: 1 
      default-binder: kafka
      bindings:
        channel:
          destination: topic
          content_type: application/json
          autoCreateTopics: false
          group: group
          consumer:
            maxAttempts: 1
      kafka:
        binder:
          autoCreateTopics: false
          autoAddPartitions: false
          brokers: brokers list
        bindings:
          channel:
            consumer:
              autoCommitOnError: true
              autoCommitOffset: true
              configuration:
                max.poll.interval.ms: 1000000
                max.poll.records: 1 
                group.id: group

我们使用@Streamlisteners来使用消息。

下面是我们收到的重复实例和服务器日志中捕获的错误消息。

错误46---[container-0-C-1]o.a.k.C.C.内部构件。ConsumerCoordinator:[Consumer clientId=Consumer-3，groupId=group]在偏移量1291358处对分区主题0的偏移量提交失败：协调器不知道此成员。错误46---[container-0-C-1]o.s.kafka。听众。LoggingErrorHandler:处理时出错：清空组织。阿帕奇。Kafka。客户。消费者CommitFailedException:无法完成提交，因为组已重新平衡并将分区分配给另一个成员。这意味着后续调用poll（）的间隔时间比配置的max.poll长。间隔ms，这通常意味着轮询循环花费了太多时间来处理消息。您可以通过增加会话超时或使用max.poll减少poll（）中返回的批处理的最大大小来解决此问题。记录。在org。阿帕奇。Kafka。客户。消费者内部。消费者协调员$OffsetCommitResponseHandler。handle（ConsumerCoordinator.java:871）~[kafka-clients-2.0.1.jar！/：na]

没有崩溃，所有实例在重复时都是健康的。此外，由于消息被成功处理了两次，错误日志-处理时错误：null也很混乱。max.poll.interval.ms： 100000大约是16分钟，应该有足够的时间来处理系统的任何消息，会话超时和心位配置是默认的。在大多数实例中，重复在2秒内收到。我们缺少什么配置？非常感谢任何建议/帮助。

Answer 1

此外，请记住StreamListener和基于注释的编程模型已经被弃用了3年，并且已经从当前的main中删除，这意味着下一个版本将不会有它。请将您的解决方案迁移到基于函数的编程模型

Answer 2

提交无法完成，因为组已经重新平衡

重新平衡是因为您的侦听器花费了太长时间；您应该调整max.poll.records和max.poll.interval.ms，以确保您总是能够在时限内处理收到的记录。

在任何情况下，Kafka都不保证完全一次交付，只保证至少一次交付。您需要在应用程序中添加幂等性，并检测/忽略重复。