Addressed Jun's round 3 review comments - part 2

Author: adixitconfluent

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();
         }
     }