chia7712 commented on a change in pull request #8657:
URL: https://github.com/apache/kafka/pull/8657#discussion_r440576071
##########
File path:
core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorConcurrencyTest.scala
##########
@@ -307,8 +307,14 @@ class GroupCoordinatorConcurrencyTest extends
AbstractCoordinatorConcurrencyTest
override def runWithCallback(member: GroupMember, responseCallback:
CompleteTxnCallback): Unit = {
val producerId = 1000L
val offsetsPartitions = (0 to numPartitions).map(new
TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, _))
- groupCoordinator.groupManager.handleTxnCompletion(producerId,
- offsetsPartitions.map(_.partition).toSet, isCommit =
random.nextBoolean)
+ val isCommit = random.nextBoolean
+ try groupCoordinator.groupManager.handleTxnCompletion(producerId,
+ offsetsPartitions.map(_.partition).toSet, isCommit = isCommit)
+ catch {
+ case e: IllegalStateException if isCommit
+ && e.getMessage.contains("though the offset commit record itself
hasn't been appended to the log")=>
Review comment:
> That seems a bug.
The root cause (changed by this PR) is that the "txn initialization" and
"txn append" are not executed within same lock.
**The test story is shown below.**
```CommitTxnOffsetsOperation``` calls
```GroupMetadata.prepareTxnOffsetCommit``` to add
```CommitRecordMetadataAndOffset(None, offsetAndMetadata)``` to
```pendingTransactionalOffsetCommits``` (this is the link you attached).
```GroupMetadata.completePendingTxnOffsetCommit``` called by
```CompleteTxnOperation``` throws ```IllegalStateException``` if
```CommitRecordMetadataAndOffset.appendedBatchOffset``` is ```None```
(https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala#L664).
**Why it does not cause error before?**
```CommitRecordMetadataAndOffset.appendedBatchOffset``` is updated by the
callback ```putCacheCallback```
(https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala#L407).
```TestReplicManager``` always create ```delayedProduce``` do handle the
```putCacheCallback```
(https://github.com/apache/kafka/blob/trunk/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala#L188).
The condition to complete the ```delayedProduce``` is
```completeAttempts.incrementAndGet() >= 3```. And the condition gets true when
call both ```producePurgatory.tryCompleteElseWatch(delayedProduce,
producerRequestKeys)``` and ```tryCompleteDelayedRequests()``` since the former
calls ```tryComplete``` two times and another calls ```tryComplete``` once. It
means ```putCacheCallback``` is always executed by
```TestReplicManager.appendRecords``` and noted that
```TestReplicManager.appendRecords``` is executed within a group lock
(https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala#L738)
. In short, txn initialization
(https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala#L464)
and txn append
(https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala#L407)
are executed with same group lock. Hence, the following execution order is
impossible.
1. txn initialization
1. txn completion
1. txn append
However, this PR disable to complete delayed requests within group lock held
by caller. The ```putCacheCallback``` which used to append txn needs to require
group lock again.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
