Fix fetch Blob Async Decrement Reference (#19942)

Signed-off-by: Harsh Kothari <techarsh@amazon.com>

Author: Harsh-87

server/src/main/java/org/opensearch/index/store/remote/utils/TransferManager.java

@@ -232,6 +232,7 @@ public IndexInput getIndexInput() throws IOException {
         @ExperimentalApi
         public CompletableFuture<IndexInput> asyncLoadIndexInput(Executor executor) {
             if (isClosed.get()) {
+                fileCache.decRef(request.getFilePath());
                 return CompletableFuture.failedFuture(new IllegalStateException("Already closed"));
             }
             if (isStarted.getAndSet(true) == false) {
@@ -252,6 +253,9 @@ public CompletableFuture<IndexInput> asyncLoadIndexInput(Executor executor) {
                     }
                     return null;
                 });
+            } else {
+                // Decreasing the extra ref count introduced by compute
+                fileCache.decRef(request.getFilePath());
             }
             return result;
         }

server/src/test/java/org/opensearch/index/store/remote/utils/TransferManagerTestCase.java

@@ -324,6 +324,18 @@ public void testRefCount() throws Exception {
         assertEquals(Optional.of(1), Optional.of(fileCache.getRef(blobFetchRequest.getFilePath())));
     }
 
+    public void testRefMultipleCount() throws Exception {
+        List<BlobFetchRequest.BlobPart> blobParts = new ArrayList<>();
+        String blobname = "test-blob";
+        blobParts.add(new BlobFetchRequest.BlobPart("blob", 0, EIGHT_MB));
+        BlobFetchRequest blobFetchRequest = BlobFetchRequest.builder().fileName(blobname).directory(directory).blobParts(blobParts).build();
+        transferManager.fetchBlob(blobFetchRequest);
+        assertNotNull(fileCache.getRef(blobFetchRequest.getFilePath()));
+        transferManager.fetchBlobAsync(blobFetchRequest).join();
+        waitUntil(() -> fileCache.getRef(blobFetchRequest.getFilePath()) == 1, 10, TimeUnit.SECONDS);
+        assertEquals(Optional.of(1), Optional.of(fileCache.getRef(blobFetchRequest.getFilePath())));
+    }
+
     protected abstract void initializeTransferManager() throws IOException;
 
     protected abstract void mockExceptionWhileReading() throws IOException;