Spring kafka-多个消费者收到相同消息

Question

我正在使用spring kafka来消费来自kafka的消息。消费者监听器如下。

@KafkaListener(topics = "topicName",
            groupId = "groupId",
            containerFactory = "kafkaListenerFactory")
    public void onMessage(ConsumerRecord record) {
       logger.info("Received Message from kafka topic " + record.topic() + " with record key " + kafkaRecordKey + " partition " + record.partition() + " offset " +record.offset());

   }

应用程序的单个实例，并发数为6。

该主题有6个分区。

Time: 5/27/22 6:28:52.864 PM
message: Received Message from kafka topic payment-topic with record key ti9:a1956769-28d2-4329-a0ff-9003003a3cde partition 4 offset 325
thread: org.springframework.kafka.KafkaListenerEndpointContainer#0-4-C-1
threadId: 69

Time: 5/27/22 6:28:52.864 PM
message: Received Message from kafka topic payment-topic with record key ti9:a1956769-28d2-4329-a0ff-9003003a3cde partition 4 offset 325
thread: org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1
threadId: 66

从上面的日志中可以清楚地看到，两个用户在完全相同的时间收到了来自分区和偏移量的相同消息。

每个线程继续处理消息。最后，其中一个消费者失败了，错误如下

Time: 5/27/22 6:28:52.887 PM
message: [Consumer clientId=consumer-payment-consumer-5, groupId=payment-consumer] Offset commit failed on partition payment-topic-4 at offset 326: The coordinator is not aware of this member.
thread: org.springframework.kafka.KafkaListenerEndpointContainer#0-4-C-1
threadId: 69 

Time: 5/27/22 6:28:53.902 PM
message: Error handler threw an exception
thread: org.springframework.kafka.KafkaListenerEndpointContainer#0-4-C-1
threadId: 69
   threadPriority: 5
   thrown: { [-]
     cause: { [+]
     }
     commonElementCount: 0
     extendedStackTrace: [ [+]
     ]
     localizedMessage: Seek to current after exception; nested exception is org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.
     message: Seek to current after exception; nested exception is org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.
     name: org.springframework.kafka.KafkaException

我知道上面的错误会在有负载或消息处理需要时间时出现。在这种情况下，处理不到一秒钟，kafka主题中的消息不到10条。

请告知为什么多个消费者收到相同的消息。

此外，错误日志显示“偏移量为 326 处的分区付款主题 4 上的偏移量提交失败”，用于偏移量为 325 处的消息

库版本

Spring boot - 2.5.7
org.springframework.kafka.spring-kafka - 2.7.8
org.apache.kafka.kafka-clients - 2.8.1

Answer 1

记录的处理时间必须小于max.poll.interval。ms＜/code＞否则会发生重新平衡，很可能当前处理的记录偏移量未提交，因此另一个分配的使用者仅从该分区的先前提交的偏移量中提取。

https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#consumerconfigs_max.poll.interval.ms