junrao commented on a change in pull request #8657:
URL: https://github.com/apache/kafka/pull/8657#discussion_r441177055
##########
File path: core/src/main/scala/kafka/server/DelayedOperation.scala
##########
@@ -100,40 +99,27 @@ abstract class DelayedOperation(override val delayMs: Long,
def tryComplete(): Boolean
/**
- * Thread-safe variant of tryComplete() that attempts completion only if the
lock can be acquired
- * without blocking.
+ * Thread-safe variant of tryComplete() that attempts completion after it
succeed to hold the lock.
*
- * If threadA acquires the lock and performs the check for completion before
completion criteria is met
- * and threadB satisfies the completion criteria, but fails to acquire the
lock because threadA has not
- * yet released the lock, we need to ensure that completion is attempted
again without blocking threadA
- * or threadB. `tryCompletePending` is set by threadB when it fails to
acquire the lock and at least one
- * of threadA or threadB will attempt completion of the operation if this
flag is set. This ensures that
- * every invocation of `maybeTryComplete` is followed by at least one
invocation of `tryComplete` until
- * the operation is actually completed.
+ * There is a long story about using "lock" or "tryLock".
+ *
+ * 1) using lock - There were a lot of cases that a thread holds a group
lock and then it tries to hold more group
+ * locks to complete delayed requests. Unfortunately, the scenario causes
deadlock and so we had introduced the
+ * "tryLock" to avoid deadlock.
+ *
+ * 2) using tryLock - However, the "tryLock" causes another issue that the
delayed requests may be into
+ * oblivion if the thread, which should complete the delayed requests, fails
to get the lock.
+ *
+ * Now, we go back to use "lock" and make sure the thread which tries to
complete delayed requests does NOT hold lock.
+ * We introduces a flag "completeDelayedRequests" to
ReplicaManager.appendRecords() (and related methods).
+ *
+ * 1) If "completeDelayedRequests" is true, the delayed requests may be
completed as much as possible.
Review comment:
the delayed requests may be completed as much as possible => the
delayed requests may be completed inside the call with the expectation that no
conflicting locks are held by the caller
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -560,19 +603,23 @@ class ReplicaManager(val config: KafkaConfig,
* Append messages to leader replicas of the partition, and wait for them to
be replicated to other replicas;
* the callback function will be triggered either when timeout or the
required acks are satisfied;
* if the callback function itself is already synchronized on some object
then pass this object to avoid deadlock.
+ * @return Returning a map of successfully appended topic partitions and a
flag indicting whether the HWM has been
+ * incremented. If the caller no longer passed in
completeDelayedRequests, the caller is expected to complete
Review comment:
"If the caller no longer passed in completeDelayedRequests" The caller
still passes this in, just as false.
##########
File path: core/src/main/scala/kafka/server/DelayedOperation.scala
##########
@@ -100,40 +99,27 @@ abstract class DelayedOperation(override val delayMs: Long,
def tryComplete(): Boolean
/**
- * Thread-safe variant of tryComplete() that attempts completion only if the
lock can be acquired
- * without blocking.
+ * Thread-safe variant of tryComplete() that attempts completion after it
succeed to hold the lock.
*
- * If threadA acquires the lock and performs the check for completion before
completion criteria is met
- * and threadB satisfies the completion criteria, but fails to acquire the
lock because threadA has not
- * yet released the lock, we need to ensure that completion is attempted
again without blocking threadA
- * or threadB. `tryCompletePending` is set by threadB when it fails to
acquire the lock and at least one
- * of threadA or threadB will attempt completion of the operation if this
flag is set. This ensures that
- * every invocation of `maybeTryComplete` is followed by at least one
invocation of `tryComplete` until
- * the operation is actually completed.
+ * There is a long story about using "lock" or "tryLock".
+ *
+ * 1) using lock - There were a lot of cases that a thread holds a group
lock and then it tries to hold more group
+ * locks to complete delayed requests. Unfortunately, the scenario causes
deadlock and so we had introduced the
+ * "tryLock" to avoid deadlock.
+ *
+ * 2) using tryLock - However, the "tryLock" causes another issue that the
delayed requests may be into
+ * oblivion if the thread, which should complete the delayed requests, fails
to get the lock.
+ *
+ * Now, we go back to use "lock" and make sure the thread which tries to
complete delayed requests does NOT hold lock.
+ * We introduces a flag "completeDelayedRequests" to
ReplicaManager.appendRecords() (and related methods).
+ *
+ * 1) If "completeDelayedRequests" is true, the delayed requests may be
completed as much as possible.
+ * 2) If "completeDelayedRequests" is false, the delayed requests are NOT
completed and the watch key (for example,
+ * group key) are returned. Callers can complete the delayed requests
manually.
Review comment:
Callers can complete the delayed requests manually => Callers can
complete the delayed requests after releasing any conflicting lock.
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1108,7 +1157,11 @@ class GroupCoordinator(val brokerId: Int,
joinPurgatory.tryCompleteElseWatch(delayedRebalance, Seq(groupKey))
}
- private def removeMemberAndUpdateGroup(group: GroupMetadata, member:
MemberMetadata, reason: String): Unit = {
+ /**
+ * @return the group which is preparing to rebalacne. Callers should use
this group key to complete delayed requests
Review comment:
typo rebalacne
##########
File path:
core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
##########
@@ -239,9 +239,13 @@ class GroupMetadataManager(brokerId: Int,
}
}
+ /**
+ * @return Returning a map of successfully appended topic partitions and a
flag indicting whether the HWM has been
+ * incremented. The caller ought to to complete delayed requests for
those returned partitions.
Review comment:
to to => to
##########
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:
Thanks for the great explanation. I understand the issue now.
Essentially, this exposed a limitation of the existing test. The existing test
happens to work because the producer callbacks are always completed in the same
ReplicaManager.appendRecords() call under the group lock. However, this is not
necessarily the general case.
Your fix works, but may hide other real problems. I was thinking that
another way to fix this is to change the test a bit. For example, we expect
CompleteTxnOperation to happen after CommitTxnOffsetsOperation. So, instead of
letting them run in parallel, we can change the test to make sure that
CompleteTxnOperation only runs after CommitTxnOffsetsOperation completes
successfully. JoinGroupOperation and SyncGroupOperation might need a similar
consideration.
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1118,33 +1171,38 @@ class GroupCoordinator(val brokerId: Int,
group.removeStaticMember(member.groupInstanceId)
group.currentState match {
- case Dead | Empty =>
- case Stable | CompletingRebalance => maybePrepareRebalance(group, reason)
- case PreparingRebalance =>
joinPurgatory.checkAndComplete(GroupKey(group.groupId))
+ case Dead | Empty => None
+ case Stable | CompletingRebalance =>
+ maybePrepareRebalance(group, reason)
+ None
+ case PreparingRebalance => Some(GroupKey(group.groupId))
}
}
- private def removePendingMemberAndUpdateGroup(group: GroupMetadata,
memberId: String): Unit = {
+ /**
+ * remove the pending member and then return the group key which is in
PreparingRebalance,
+ * @param group group
+ * @param memberId member id
+ * @return group key if it is in PreparingRebalance. Otherwise, None
+ */
+ private def removePendingMemberAndUpdateGroup(group: GroupMetadata,
memberId: String): Option[GroupKey] = {
group.removePendingMember(memberId)
- if (group.is(PreparingRebalance)) {
- joinPurgatory.checkAndComplete(GroupKey(group.groupId))
- }
- }
-
- def tryCompleteJoin(group: GroupMetadata, forceComplete: () => Boolean) = {
- group.inLock {
- if (group.hasAllMembersJoined)
- forceComplete()
- else false
- }
+ if (group.is(PreparingRebalance)) Some(GroupKey(group.groupId))
+ else None
}
def onExpireJoin(): Unit = {
// TODO: add metrics for restabilize timeouts
}
- def onCompleteJoin(group: GroupMetadata): Unit = {
+ /**
+ * @return Returning a map of successfully appended topic partitions and a
flag indicting whether the HWM has been
+ * incremented. If the caller no longer passed in
completeDelayedRequests, the caller is expected to complete
Review comment:
" If the caller no longer passed in completeDelayedRequests" => There is
no completeDelayedRequests passed in.
----------------------------------------------------------------
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
