Addressed Jun's round 3 review comments - part 2

coderabbit-test · arvi18 · Apr 10, 2025 · Apr 10, 2025 · Apr 12, 2025 · Apr 12, 2025
commit f94f7ca216470a40bf80386b4441692e08456623
diff --git a/core/src/main/java/kafka/server/share/DelayedShareFetch.java b/core/src/main/java/kafka/server/share/DelayedShareFetch.java
@@ -331,8 +331,17 @@ public boolean tryComplete() {
                 releasePartitionLocks(topicPartitionData.keySet());
                 partitionsAcquired.clear();
                 localPartitionsAlreadyFetched.clear();
+                return forceComplete();
+            } else {
+                boolean completedByMe = forceComplete();
+                // If invocation of forceComplete is not successful, then that means the request is already completed
+                // hence release the acquired locks. This can occur in case of remote storage fetch if there is a thread that
+                // completes the operation (due to expiration) right before a different thread is about to enter tryComplete.
+                if (!completedByMe) {
+                    releasePartitionLocks(partitionsAcquired.keySet());
+                }
+                return completedByMe;
             }
-            return forceComplete();
         }
     }