Add interface changes for async repository downloads

Signed-off-by: Kunal Kotwani <kkotwani@amazon.com>

CHANGELOG.md

@@ -86,6 +86,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Disallow compression level to be set for default and best_compression index codecs ([#8737]()https://github.com/opensearch-project/OpenSearch/pull/8737)
 - Prioritize replica shard movement during shard relocation ([#8875](https://github.com/opensearch-project/OpenSearch/pull/8875))
 - Introducing Default and Best Compression codecs as their algorithm name ([#9123]()https://github.com/opensearch-project/OpenSearch/pull/9123)
+- New API for supporting async downloads within VerifyingMultiStreamBlobContainer ([#9182](https://github.com/opensearch-project/OpenSearch/pull/9182))
 
 ### Dependencies
 - Bump `org.apache.logging.log4j:log4j-core` from 2.17.1 to 2.20.0 ([#8307](https://github.com/opensearch-project/OpenSearch/pull/8307))
@@ -129,4 +130,4 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Security
 
 [Unreleased 3.0]: https://github.com/opensearch-project/OpenSearch/compare/2.x...HEAD
-[Unreleased 2.x]: https://github.com/opensearch-project/OpenSearch/compare/2.10...2.x
+[Unreleased 2.x]: https://github.com/opensearch-project/OpenSearch/compare/2.10...2.x

plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3BlobContainer.java

@@ -46,6 +46,7 @@
 import org.opensearch.common.blobstore.BlobStoreException;
 import org.opensearch.common.blobstore.DeleteResult;
 import org.opensearch.common.blobstore.VerifyingMultiStreamBlobContainer;
+import org.opensearch.common.blobstore.stream.read.ReadContext;
 import org.opensearch.common.blobstore.stream.write.WriteContext;
 import org.opensearch.common.blobstore.stream.write.WritePriority;
 import org.opensearch.common.blobstore.support.AbstractBlobContainer;
@@ -210,6 +211,11 @@ public void asyncBlobUpload(WriteContext writeContext, ActionListener<Void> comp
         }
     }
 
+    @Override
+    public CompletableFuture<ReadContext> asyncBlobDownload(String blobName, boolean forceSingleStream) throws IOException {
+        throw new UnsupportedOperationException("S3 BlobContainer currently does not support async blob downloads");
+    }
+
     // package private for testing
     long getLargeBlobThresholdInBytes() {
         return blobStore.bufferSizeInBytes();

plugins/repository-s3/src/test/java/org/opensearch/repositories/s3/S3BlobStoreContainerTests.java

@@ -879,6 +879,19 @@ public void onFailure(Exception e) {}
         }
     }
 
+    public void testAsyncBlobDownload() {
+        final S3BlobStore blobStore = mock(S3BlobStore.class);
+        final BlobPath blobPath = mock(BlobPath.class);
+        final String blobName = "test-blob";
+
+        final UnsupportedOperationException e = expectThrows(UnsupportedOperationException.class, () -> {
+            final S3BlobContainer blobContainer = new S3BlobContainer(blobPath, blobStore);
+            blobContainer.asyncBlobDownload(blobName, false);
+        });
+
+        assertEquals("S3 BlobContainer currently does not support async blob downloads", e.getMessage());
+    }
+
     public void testListBlobsByPrefixInLexicographicOrderWithNegativeLimit() throws IOException {
         testListBlobsByPrefixInLexicographicOrder(-5, 0, BlobContainer.BlobNameSortOrder.LEXICOGRAPHIC);
     }

server/src/internalClusterTest/java/org/opensearch/remotestore/multipart/mocks/MockFsVerifyingBlobContainer.java

@@ -11,6 +11,7 @@
 import org.apache.lucene.index.CorruptIndexException;
 import org.opensearch.action.ActionListener;
 import org.opensearch.common.blobstore.VerifyingMultiStreamBlobContainer;
+import org.opensearch.common.blobstore.stream.read.ReadContext;
 import org.opensearch.common.io.InputStreamContainer;
 import org.opensearch.common.StreamContext;
 import org.opensearch.common.blobstore.BlobPath;
@@ -24,7 +25,12 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -114,6 +120,32 @@ public void asyncBlobUpload(WriteContext writeContext, ActionListener<Void> comp
 
     }
 
+    @Override
+    public CompletableFuture<ReadContext> asyncBlobDownload(String blobName, boolean forceSingleStream) throws IOException {
+        final int numStreams = forceSingleStream ? 1 : 5;
+        final ExecutorService executorService = Executors.newFixedThreadPool(numStreams);
+
+        // Fetch blob metadata
+        final InputStream blobInputStream = readBlob(blobName);
+        final long blobSize = blobInputStream.available();
+        blobInputStream.close();
+
+        // Create input streams for the blob
+        final List<InputStream> blobInputStreams = new ArrayList<>();
+        long streamSize = (int) Math.ceil(blobSize * 1.0 / numStreams);
+        for (int streamNumber = 0; streamNumber < numStreams; streamNumber++) {
+            long start = streamNumber * streamSize;
+            blobInputStreams.add(readBlob(blobName, start, streamSize));
+        }
+
+        CompletableFuture<ReadContext> readContextFuture = CompletableFuture.supplyAsync(
+            () -> new ReadContext(blobInputStreams, null, numStreams, blobSize),
+            executorService
+        );
+        executorService.shutdown();
+        return readContextFuture;
+    }
+
     private boolean isSegmentFile(String filename) {
         return !filename.endsWith(".tlog") && !filename.endsWith(".ckp");
     }

server/src/main/java/org/opensearch/common/blobstore/VerifyingMultiStreamBlobContainer.java

@@ -9,9 +9,11 @@
 package org.opensearch.common.blobstore;
 
 import org.opensearch.action.ActionListener;
+import org.opensearch.common.blobstore.stream.read.ReadContext;
 import org.opensearch.common.blobstore.stream.write.WriteContext;
 
 import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * An extension of {@link BlobContainer} that adds {@link VerifyingMultiStreamBlobContainer#asyncBlobUpload} to allow
@@ -31,4 +33,17 @@ public interface VerifyingMultiStreamBlobContainer extends BlobContainer {
      * @throws IOException if any of the input streams could not be read, or the target blob could not be written to
      */
     void asyncBlobUpload(WriteContext writeContext, ActionListener<Void> completionListener) throws IOException;
+
+    /**
+     * Creates and populates a list of {@link java.io.InputStream} from the blob stored within the repository, returned
+     * within an async callback using the {@link ReadContext} object.
+     * Defaults to using multiple streams, when feasible, unless a single stream is forced using @param forceSingleStream.
+     * An {@link IOException} is thrown if requesting any of the input streams fails, or reading metadata for the
+     * requested blob fails
+     * @param blobName          Name of the blob to be read using the async mechanism
+     * @param forceSingleStream Value to denote forced use of a single stream within the returned object
+     * @return A future of {@link ReadContext} object which serves the input streams and other metadata for the blob
+     * @throws IOException if any of the input streams could not be requested, or reading metadata for requested blob fails
+     */
+    CompletableFuture<ReadContext> asyncBlobDownload(String blobName, boolean forceSingleStream) throws IOException;
 }

server/src/main/java/org/opensearch/common/blobstore/stream/read/ReadContext.java

@@ -0,0 +1,47 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.common.blobstore.stream.read;
+
+import java.io.InputStream;
+import java.util.List;
+
+/**
+ * ReadContext is used to encapsulate all data needed by <code>VerifyingMultiStreamBlobContainer#asyncBlobDownload</code>
+ *
+ * @opensearch.experimental
+ */
+public class ReadContext {
+    private final List<InputStream> blobInputStreams;
+    private final String blobChecksum;
+    private final int numStreams;
+    private final long blobSize;
+
+    public ReadContext(List<InputStream> blobInputStreams, String blobChecksum, int numStreams, long blobSize) {
+        this.blobInputStreams = blobInputStreams;
+        this.blobChecksum = blobChecksum;
+        this.numStreams = numStreams;
+        this.blobSize = blobSize;
+    }
+
+    public List<InputStream> getBlobInputStreams() {
+        return blobInputStreams;
+    }
+
+    public String getBlobChecksum() {
+        return blobChecksum;
+    }
+
+    public int getNumStreams() {
+        return numStreams;
+    }
+
+    public long getBlobSize() {
+        return blobSize;
+    }
+}

server/src/main/java/org/opensearch/common/blobstore/stream/read/package-info.java

@@ -0,0 +1,10 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+/** Abstractions for stream based file reads */
+package org.opensearch.common.blobstore.stream.read;