Error after consumer reconnect "Transaction offset Commit failed due to consumer group metadata mismatch: Specified group generation id is not valid."

[damn1kk]

Hey! I have an issue after consumer has reconnected to broker.

This error has appeared since version 1.3.9 after this commit: 2ff59b3

Looks like groupMetadata is not thread-safe in org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.

This error is hard to reproduce, i only got this error in some tests in CI pipeline.

What's happened by logs:
First, the consumer connected with generationId = 1

19:39:55.024 Successfully synced group in generation Generation{generationId=1, memberId='consumer-authTransactionGroup-1-da853b91-5617-4f55-a502-8189e7bf283b', protocol='cooperative-sticky'}
19:39:55.033 Executing onJoinComplete with generation 1 and memberId consumer-authTransactionGroup-1-da853b91-5617-4f55-a502-8189e7bf283b

Then consumer disconnected from kafka and successfully reconnected with generationId = 2:

19:40:55.790 Error sending fetch request (sessionId=1714076453, epoch=120) to node 1001: org.apache.kafka.common.errors.DisconnectException: null

19:41:05.599 Successfully synced group in generation Generation{generationId=2, memberId='consumer-authTransactionGroup-1-da853b91-5617-4f55-a502-8189e7bf283b', protocol='cooperative-sticky'}
19:41:05.610 Executing onJoinComplete with generation 2 and memberId consumer-authTransactionGroup-1-da853b91-5617-4f55-a502-8189e7bf283b

But ConsumerGroupMetadata still has generationId = 1:

19:41:06.426 Enqueuing transactional request FindCoordinatorRequestData(key='12f76c12-64dd-44bc-b916-5beb34c6136b', keyType=1, coordinatorKeys=[])
19:41:06.537 Discovered transaction coordinator kafka:19092 (id: 1001 rack: null)
19:41:06.679 ProducerId set to 1000 with epoch 0
19:41:06.687 Transition from state INITIALIZING to READY
19:41:06.696 Transition from state READY to IN_TRANSACTION
19:41:06.724 Begin adding offsets {testtopic-0=OffsetAndMetadata{offset=1, leaderEpoch=null, metadata=''}} for consumer group GroupMetadata(groupId = authTransactionGroup, generationId = 1, memberId = consumer-authTransactionGroup-1-da853b91-5617-4f55-a502-8189e7bf283b, groupInstanceId = ) to transaction
19:41:06.732 Enqueuing transactional request AddOffsetsToTxnRequestData(transactionalId='12f76c12-64dd-44bc-b916-5beb34c6136b', producerId=1000, producerEpoch=0, groupId='authTransactionGroup')
19:41:06.750 Successfully added partition for consumer group authTransactionGroup to transaction
19:41:06.769 Enqueuing transactional request FindCoordinatorRequestData(key='authTransactionGroup', keyType=0, coordinatorKeys=[])
19:41:06.777 Enqueuing transactional request TxnOffsetCommitRequestData(transactionalId='12f76c12-64dd-44bc-b916-5beb34c6136b', groupId='authTransactionGroup', producerId=1000, producerEpoch=0, generationId=1, memberId='consumer-authTransactionGroup-1-da853b91-5617-4f55-a502-8189e7bf283b', groupInstanceId=null, topics=[TxnOffsetCommitRequestTopic(name='testtopic', partitions=[TxnOffsetCommitRequestPartition(partitionIndex=0, committedOffset=1, committedLeaderEpoch=-1, committedMetadata='')])])
19:41:06.783 Successfully read a message
19:41:06.789 Discovered group coordinator kafka:19092 (id: 1001 rack: null)
19:41:06.908 Received TxnOffsetCommit response for consumer group authTransactionGroup: {testtopic-0=ILLEGAL_GENERATION}
19:41:06.918 Transiting to abortable error state due to org.apache.kafka.clients.consumer.CommitFailedException: Transaction offset Commit failed due to consumer group metadata mismatch: Specified group generation id is not valid.
19:41:06.926 Transition from state IN_TRANSACTION to error state ABORTABLE_ERROR

19:41:06.926 org.apache.kafka.clients.consumer.CommitFailedException: Transaction offset Commit failed due to consumer group metadata mismatch: Specified group generation id is not valid.
 	at org.apache.kafka.clients.producer.internals.TransactionManager$TxnOffsetCommitHandler.handleResponse(TransactionManager.java:1737)
 	at org.apache.kafka.clients.producer.internals.TransactionManager$TxnRequestHandler.onComplete(TransactionManager.java:1288)
 	at org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:109)
 	at org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:574)
	at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:566)
 	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)
 	at java.base/java.lang.Thread.run(Thread.java:834)

Environment

• reactor-kafka 1.3.9 - 1.3.12
• kafka-clients 2.8.1 - 3.0.0