You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ju...@apache.org on 2024/11/20 17:26:52 UTC
(kafka) branch trunk updated: KAFKA-18027: MINOR: Correct DelayedOperationPurgatory code around adding of an already completed operation (#17842)
This is an automated email from the ASF dual-hosted git repository.
junrao pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new aa7a3dbd303 KAFKA-18027: MINOR: Correct DelayedOperationPurgatory code around adding of an already completed operation (#17842)
aa7a3dbd303 is described below
commit aa7a3dbd3037bbf0de703361c4655335ce6402c4
Author: Abhinav Dixit <ad...@confluent.io>
AuthorDate: Wed Nov 20 22:56:44 2024 +0530
KAFKA-18027: MINOR: Correct DelayedOperationPurgatory code around adding of an already completed operation (#17842)
Reviewers: Apoorv Mittal <ap...@gmail.com>, Jun Rao <ju...@gmail.com>
---
.../org/apache/kafka/server/purgatory/DelayedOperationPurgatory.java | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git a/server-common/src/main/java/org/apache/kafka/server/purgatory/DelayedOperationPurgatory.java b/server-common/src/main/java/org/apache/kafka/server/purgatory/DelayedOperationPurgatory.java
index 3491aee139e..380f22c9c8e 100644
--- a/server-common/src/main/java/org/apache/kafka/server/purgatory/DelayedOperationPurgatory.java
+++ b/server-common/src/main/java/org/apache/kafka/server/purgatory/DelayedOperationPurgatory.java
@@ -147,7 +147,10 @@ public class DelayedOperationPurgatory<T extends DelayedOperation> {
// any exclusive lock. Since DelayedOperationPurgatory.checkAndComplete() completes delayed operations asynchronously,
// holding an exclusive lock to make the call is often unnecessary.
if (operation.safeTryCompleteOrElse(() -> {
- watchKeys.forEach(key -> watchForOperation(key, operation));
+ watchKeys.forEach(key -> {
+ if (!operation.isCompleted())
+ watchForOperation(key, operation);
+ });
if (!watchKeys.isEmpty())
estimatedTotalOperations.incrementAndGet();
})) {