Spring Kafka：由于消费者组元数据不匹配，事务偏移量提交失败答案

【问题标题】：Spring Kafka: Transaction offset Commit failed due to consumer group metadata mismatchSpring Kafka：由于消费者组元数据不匹配，事务偏移量提交失败
【发布时间】：2023-02-19 20:27:09
【问题描述】：

Kafka 消息消费正常。但是我间歇性地收到这个错误。

Exception:org.springframework.kafka.KafkaException: Seek to current after exception; nested exception is org.apache.kafka.clients.consumer.CommitFailedException: Transaction offset Commit failed due to consumer group metadata mismatch: The coordinator is not aware of this member.
at org.springframework.kafka.listener.SeekUtils.seekOrRecover(SeekUtils.java:206) \~\[spring-kafka-2.7.14.jar!/:2.7.14\]
at org.springframework.kafka.listener.SeekToCurrentErrorHandler.handle(SeekToCurrentErrorHandler.java:112) \~\[spring-kafka-2.7.14.jar!/:2.7.14\]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeErrorHandler(KafkaMessageListenerContainer.java:2377) \~\[spring-kafka-2.7.14.jar!/:2.7.14\]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:2254) \~\[spring-kafka-2.7.14.jar!/:2.7.14\]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.access$2300(KafkaMessageListenerContainer.java:518) \~\[spring-kafka-2.7.14.jar!/:2.7.14\]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer$3.doInTransactionWithoutResult(KafkaMessageListenerContainer.java:2092) \~\[spring-kafka-2.7.14.jar!/:2.7.14\]
at org.springframework.transaction.support.TransactionCallbackWithoutResult.doInTransaction(TransactionCallbackWithoutResult.java:36) \~\[spring-tx-5.3.20.jar!/:5.3.20\]
at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:140) \~\[spring-tx-5.3.20.jar!/:5.3.20\]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeInTransaction(KafkaMessageListenerContainer.java:2083) \~\[spring-kafka-2.7.14.jar!/:2.7.14\]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListenerInTx(KafkaMessageListenerContainer.java:2052) \~\[spring-kafka-2.7.14.jar!/:2.7.14\]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:2028) \~\[spring-kafka-2.7.14.jar!/:2.7.14\]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:1704) \~\[spring-kafka-2.7.14.jar!/:2.7.14\]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeIfHaveRecords(KafkaMessageListenerContainer.java:1275) \~\[spring-kafka-2.7.14.jar!/:2.7.14\]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1267) \~\[spring-kafka-2.7.14.jar!/:2.7.14\]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1162) \~\[spring-kafka-2.7.14.jar!/:2.7.14\]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) \~\[?:?\]
at java.util.concurrent.FutureTask.run(FutureTask.java:264) \~\[?:?\]
at java.lang.Thread.run(Thread.java:834) \~\[?:?\]
Caused by: org.apache.kafka.clients.consumer.CommitFailedException: Transaction offset Commit failed due to consumer group metadata mismatch: The coordinator is not aware of this member.
at org.apache.kafka.clients.producer.internals.TransactionManager$TxnOffsetCommitHandler.handleResponse(TransactionManager.java:1732) \~\[kafka-clients-2.7.2.jar!/:?\]
at org.apache.kafka.clients.producer.internals.TransactionManager$TxnRequestHandler.onComplete(TransactionManager.java:1290) \~\[kafka-clients-2.7.2.jar!/:?\]
at org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:109) \~\[kafka-clients-2.7.2.jar!/:?\]
at org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:586) \~\[kafka-clients-2.7.2.jar!/:?\]
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:578) \~\[kafka-clients-2.7.2.jar!/:?\]
at org.apache.kafka.clients.producer.internals.Sender.maybeSendAndPollTransactionalRequest(Sender.java:415) \~\[kafka-clients-2.7.2.jar!/:?\]
at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:313) \~\[kafka-clients-2.7.2.jar!/:?\]
at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:240) \~\[kafka-clients-2.7.2.jar!/:?\]

为什么会出现此错误？消费kafka消息是没有问题的。

对不起。我会说一点点英语。所以这个问题很尴尬。

【问题讨论】：

你确定没有问题吗？看起来消费者组重新平衡了，所以正如错误所说，组协调器不再知道该实例，因此无法提交事务。添加一个再平衡监听器

标签： apache-kafka spring-kafka

【解决方案1】：

此异常是否会导致您的消费应用程序出现任何问题？此异常很可能是警告，而不是错误，因此它不应终止使用者。但是，根据您的偏移重置策略，您可能会丢失消息或可能有重复消息。

最有可能发生的情况是，其中一个消费者已关闭，而另一个消费者已加入该组以取代它。如错误所述，协调器尚未意识到新消费者，因此需要进行重新平衡，以便为新消费者分配公平份额的分区。

如上所述，我会检查您是否缺少消息/是否有重复消息。

【讨论】：

【解决方案2】：

这也发生在我们身上，这是一个错误，并且无法恢复。我们必须重新启动应用程序才能再次运行。

我们的案例是，Kafka 代理开始滚动重启以更新证书，这导致集群中出现大量问题，但通常它会恢复，除了一些超时。

我们使用 confluent kafka 算子。

这只发生过一次，但它发生在生产中所以有点痛..

例外：

org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state
    at org.apache.kafka.clients.producer.internals.TransactionManager.maybeFailWithError(TransactionManager.java:1125)
    at org.apache.kafka.clients.producer.internals.TransactionManager.sendOffsetsToTransaction(TransactionManager.java:420)
    at org.apache.kafka.clients.producer.KafkaProducer.sendOffsetsToTransaction(KafkaProducer.java:719)
    at org.springframework.kafka.core.DefaultKafkaProducerFactory$CloseSafeProducer.sendOffsetsToTransaction(DefaultKafkaProducerFactory.java:1149)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doSendOffsets(KafkaMessageListenerContainer.java:2816)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.sendOffsetsToTransaction(KafkaMessageListenerContainer.java:2807)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.commitOffsetsIfNeeded(KafkaMessageListenerContainer.java:2187)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeBatchListener(KafkaMessageListenerContainer.java:2161)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.access$1900(KafkaMessageListenerContainer.java:553)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer$1.doInTransactionWithoutResult(KafkaMessageListenerContainer.java:2048)
    at org.springframework.transaction.support.TransactionCallbackWithoutResult.doInTransaction(TransactionCallbackWithoutResult.java:36)
    at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:140)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeBatchListenerInTx(KafkaMessageListenerContainer.java:2039)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeBatchListener(KafkaMessageListenerContainer.java:2021)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:2003)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeIfHaveRecords(KafkaMessageListenerContainer.java:1375)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1366)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1257)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.apache.kafka.clients.consumer.CommitFailedException: Transaction offset Commit failed due to consumer group metadata mismatch: The coordinator is not aware of this member.
    at org.apache.kafka.clients.producer.internals.TransactionManager$TxnOffsetCommitHandler.handleResponse(TransactionManager.java:1771)
    at org.apache.kafka.clients.producer.internals.TransactionManager$TxnRequestHandler.onComplete(TransactionManager.java:1322)
    at org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:109)
    at org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:583)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:575)
    at org.apache.kafka.clients.producer.internals.Sender.maybeSendAndPollTransactionalRequest(Sender.java:418)
    at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:316)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:243)
    ... 1 common frames omitted

【讨论】：