Correctly clone IndexInputs to avoid race

In PR opensearch-project#6345 I did remove a duplicate clone, however this resulted in
cloning the IndexInput in the wrong place. When requesting a file that
needs to be downloaded, we have a mechanism to ensure that concurrent
calls do not end up duplicating the download, which results in multiple
threads being given the same instance. The clone must happen _after_
this point to ensure that each thread gets its own clone.

Signed-off-by: Andrew Ross <andrross@amazon.com>

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

@@ -52,11 +52,13 @@ public TransferManager(final BlobContainer blobContainer, final ExecutorService
     /**
      * Given a blobFetchRequest, return it's corresponding IndexInput.
      * @param blobFetchRequest to fetch
-     * @return
+     * @return The IndexInput of the requested data
      */
     public IndexInput fetchBlob(BlobFetchRequest blobFetchRequest) {
         try {
             final IndexInput origin = fetchOriginIndexInput(blobFetchRequest).get();
+            // The origin instances stays in the cache with a ref count of zero
+            // and must be cloned before being returned.
             return origin.clone();
         } catch (InterruptedException | ExecutionException e) {
             if (e instanceof InterruptedException) {
@@ -94,7 +96,7 @@ private CompletableFuture<IndexInput> fetchOriginIndexInput(BlobFetchRequest blo
                 return cachedIndexInput;
             });
 
-            if (Objects.isNull(origin)) {
+            if (origin == null) {
                 // origin is not in file cache, download origin
 
                 // open new origin

server/src/test/java/org/opensearch/index/store/remote/file/OnDemandBlockSnapshotIndexInputTests.java

@@ -30,7 +30,6 @@
 import java.io.EOFException;
 import java.io.IOException;
 import java.nio.file.Path;
-import java.util.concurrent.CompletableFuture;
 
 import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
 
@@ -142,10 +141,8 @@ private OnDemandBlockSnapshotIndexInput createOnDemandBlockSnapshotIndexInput(in
 
         doAnswer(invocation -> {
             BlobFetchRequest blobFetchRequest = invocation.getArgument(0);
-            return CompletableFuture.completedFuture(
-                blobFetchRequest.getDirectory().openInput(blobFetchRequest.getFileName(), IOContext.READ)
-            );
-        }).when(transferManager).asyncFetchBlob(any());
+            return blobFetchRequest.getDirectory().openInput(blobFetchRequest.getFileName(), IOContext.READ);
+        }).when(transferManager).fetchBlob(any());
 
         FSDirectory directory = null;
         try {