Addressing comments and adding UT's

Author: Ajay Kumar Movva

CHANGELOG.md

@@ -43,7 +43,6 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Make node duress values cacheable ([#18649](https://github.com/opensearch-project/OpenSearch/pull/18649))
 - Change default value of remote_data_ratio, which is used in Searchable Snapshots and Writeable Warm from 0 to 5 and min allowed value to 1 ([#18767](https://github.com/opensearch-project/OpenSearch/pull/18767))
 - Making multi rate limiters in repository dynamic [#18069](https://github.com/opensearch-project/OpenSearch/pull/18069)
-- Adding changes to support async fetch blob for transfer manager [#18754](https://github.com/opensearch-project/OpenSearch/pull/18754)
 
 ### Dependencies
 - Bump `stefanzweifel/git-auto-commit-action` from 5 to 6 ([#18524](https://github.com/opensearch-project/OpenSearch/pull/18524))

server/src/main/java/org/opensearch/index/store/CompositeDirectory.java

@@ -60,7 +60,7 @@ public class CompositeDirectory extends FilterDirectory {
     protected final RemoteSegmentStoreDirectory remoteDirectory;
     protected final FileCache fileCache;
     protected final TransferManager transferManager;
-    private final ThreadPool threadPool;
+    protected final ThreadPool threadPool;
 
     /**
      * Constructor to initialise the composite directory

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

@@ -36,7 +36,9 @@
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 
 @ThreadLeakFilters(filters = CleanerDaemonThreadLeakFilter.class)
@@ -234,6 +236,39 @@ public void testDownloadFailsAsyncDownload() throws Exception {
         MatcherAssert.assertThat(fileCache.usage(), equalTo(0L));
     }
 
+    public void testDownloadFailsAsyncDownloadException() throws Exception {
+        mockExceptionWhileReading();
+        doThrow(new IllegalArgumentException("Invalid thread pool")).when(threadPool).executor(ThreadPool.Names.REMOTE_RECOVERY);
+        List<BlobFetchRequest.BlobPart> blobParts = new ArrayList<>();
+        blobParts.add(new BlobFetchRequest.BlobPart("failure-blob", 0, EIGHT_MB));
+        expectThrows(IllegalArgumentException.class, () -> {
+            BlobFetchRequest blobFetchRequest = BlobFetchRequest.builder()
+                .fileName("file")
+                .directory(directory)
+                .blobParts(blobParts)
+                .build();
+            transferManager.fetchBlobAsync(blobFetchRequest);
+        });
+        MatcherAssert.assertThat(fileCache.activeUsage(), equalTo(0L));
+    }
+
+    public void testDownloadFailsAsyncDownloadExceptionFileCache() throws Exception {
+        mockExceptionWhileReading();
+
+        doThrow(new IOException("Invalid key")).when(fileCache).compute(any(), any());
+        List<BlobFetchRequest.BlobPart> blobParts = new ArrayList<>();
+        blobParts.add(new BlobFetchRequest.BlobPart("failure-blob", 0, EIGHT_MB));
+        expectThrows(IOException.class, () -> {
+            BlobFetchRequest blobFetchRequest = BlobFetchRequest.builder()
+                .fileName("file")
+                .directory(directory)
+                .blobParts(blobParts)
+                .build();
+            transferManager.fetchBlobAsync(blobFetchRequest);
+        });
+        MatcherAssert.assertThat(fileCache.activeUsage(), equalTo(0L));
+    }
+
     public void testFetchesToDifferentBlobsDoNotBlockOnEachOther() throws Exception {
         // Mock a call for a blob that will block until the latch is released,
         // then start the fetch for that blob on a separate thread