remove metdata changes for asyn download flow

Signed-off-by: Sandeep Kumawat <[email protected]>
skumawat2025 · Apr 11, 2024 · a368dac · a368dac
1 parent b528cfa
commit a368dac
Show file tree

Hide file tree

Showing 12 changed files with 22 additions and 39 deletions.
diff --git a/libs/common/src/main/java/org/opensearch/common/io/InputStreamContainer.java b/libs/common/src/main/java/org/opensearch/common/io/InputStreamContainer.java
@@ -11,7 +11,6 @@
 import org.opensearch.common.annotation.ExperimentalApi;
 
 import java.io.InputStream;
-import java.util.Map;
 
 /**
  * Model composed of an input stream and the total content length of the stream
@@ -24,20 +23,17 @@ public class InputStreamContainer {
     private final InputStream inputStream;
     private final long contentLength;
     private final long offset;
-    private final Map<String, String> metadata;
 
     /**
      * Construct a new stream object
      *
      * @param inputStream   The input stream that is to be encapsulated
      * @param contentLength The total content length that is to be read from the stream
-     * @param metadata      The metadata of the blob. This will be same for each part download.
      */
-    public InputStreamContainer(InputStream inputStream, long contentLength, long offset, Map<String, String> metadata) {
+    public InputStreamContainer(InputStream inputStream, long contentLength, long offset) {
         this.inputStream = inputStream;
         this.contentLength = contentLength;
         this.offset = offset;
-        this.metadata = metadata;
     }
 
     /**
@@ -60,11 +56,4 @@ public long getContentLength() {
     public long getOffset() {
         return offset;
     }
-
-    /**
-     * @return metadata of the source content.
-     */
-    public Map<String, String> getMetadata() {
-        return metadata;
-    }
 }
diff --git a/plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3BlobContainer.java b/plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3BlobContainer.java
@@ -822,12 +822,11 @@ static InputStreamContainer transformResponseToInputStreamContainer(
         final GetObjectResponse getObjectResponse = streamResponse.response();
         final String contentRange = getObjectResponse.contentRange();
         final Long contentLength = getObjectResponse.contentLength();
-        final Map<String, String> metadata = getObjectResponse.metadata();
         if ((isMultipartObject && contentRange == null) || contentLength == null) {
             throw SdkException.builder().message("Failed to fetch required metadata for blob part").build();
         }
         final long offset = isMultipartObject ? HttpRangeUtils.getStartOffsetFromRangeHeader(getObjectResponse.contentRange()) : 0L;
-        return new InputStreamContainer(streamResponse, getObjectResponse.contentLength(), offset, metadata);
+        return new InputStreamContainer(streamResponse, getObjectResponse.contentLength(), offset);
     }
 
     /**

diff --git a/...itory-s3/src/test/java/org/opensearch/repositories/s3/S3BlobContainerMockClientTests.java b/...itory-s3/src/test/java/org/opensearch/repositories/s3/S3BlobContainerMockClientTests.java
@@ -473,7 +473,7 @@ private void testWriteBlobByStreams(boolean expectException, boolean throwExcept
         StreamContextSupplier streamContextSupplier = partSize -> new StreamContext((partNo, size, position) -> {
             InputStream inputStream = new OffsetRangeIndexInputStream(new ByteArrayIndexInput("desc", bytes), size, position);
             openInputStreams.add(inputStream);
-            return new InputStreamContainer(inputStream, size, position, null);
+            return new InputStreamContainer(inputStream, size, position);
         }, partSize, calculateLastPartSize(bytes.length, partSize), calculateNumberOfParts(bytes.length, partSize));
 
         CheckedConsumer<Boolean, IOException> uploadFinalizer = uploadSuccess -> {
@@ -529,7 +529,7 @@ private void testWriteBlobByStreamsLargeBlob(boolean expectException, boolean th
         StreamContextSupplier streamContextSupplier = partSize1 -> new StreamContext((partNo, size, position) -> {
             InputStream inputStream = new OffsetRangeIndexInputStream(new ZeroIndexInput("desc", blobSize), size, position);
             openInputStreams.add(inputStream);
-            return new InputStreamContainer(inputStream, size, position, null);
+            return new InputStreamContainer(inputStream, size, position);
         }, partSize1, calculateLastPartSize(blobSize, partSize1), calculateNumberOfParts(blobSize, partSize1));
 
         CheckedConsumer<Boolean, IOException> uploadFinalizer = uploadSuccess -> {
@@ -651,7 +651,7 @@ private void testLargeFilesRedirectedToSlowSyncClient(boolean expectException) t
         StreamContextSupplier streamContextSupplier = partSize1 -> new StreamContext((partNo, size, position) -> {
             InputStream inputStream = new OffsetRangeIndexInputStream(new ZeroIndexInput("desc", blobSize), size, position);
             openInputStreams.add(inputStream);
-            return new InputStreamContainer(inputStream, size, position, null);
+            return new InputStreamContainer(inputStream, size, position);
         }, partSize1, calculateLastPartSize(blobSize, partSize1), calculateNumberOfParts(blobSize, partSize1));
 
         WriteContext writeContext = new WriteContext.Builder().fileName("write_large_blob")

diff --git a/...pository-s3/src/test/java/org/opensearch/repositories/s3/S3BlobContainerRetriesTests.java b/...pository-s3/src/test/java/org/opensearch/repositories/s3/S3BlobContainerRetriesTests.java
@@ -352,7 +352,7 @@ public void testWriteBlobByStreamsWithRetries() throws Exception {
         StreamContextSupplier streamContextSupplier = partSize -> new StreamContext((partNo, size, position) -> {
             InputStream inputStream = new OffsetRangeIndexInputStream(new ByteArrayIndexInput("desc", bytes), size, position);
             openInputStreams.add(inputStream);
-            return new InputStreamContainer(inputStream, size, position, null);
+            return new InputStreamContainer(inputStream, size, position);
         }, partSize, calculateLastPartSize(bytes.length, partSize), calculateNumberOfParts(bytes.length, partSize));
 
         WriteContext writeContext = new WriteContext.Builder().fileName("write_blob_by_streams_max_retries")

diff --git a/...tory-s3/src/test/java/org/opensearch/repositories/s3/async/AsyncTransferManagerTests.java b/...tory-s3/src/test/java/org/opensearch/repositories/s3/async/AsyncTransferManagerTests.java
@@ -85,7 +85,7 @@ public void testOneChunkUpload() {
             }, false, null, true, null),
             new StreamContext((partIdx, partSize, position) -> {
                 streamRef.set(new ZeroInputStream(partSize));
-                return new InputStreamContainer(streamRef.get(), partSize, position, null);
+                return new InputStreamContainer(streamRef.get(), partSize, position);
             }, ByteSizeUnit.MB.toBytes(1), ByteSizeUnit.MB.toBytes(1), 1),
             new StatsMetricPublisher()
         );
@@ -129,7 +129,7 @@ public void testOneChunkUploadCorruption() {
                 // do nothing
             }, false, null, true, null),
             new StreamContext(
-                (partIdx, partSize, position) -> new InputStreamContainer(new ZeroInputStream(partSize), partSize, position, null),
+                (partIdx, partSize, position) -> new InputStreamContainer(new ZeroInputStream(partSize), partSize, position),
                 ByteSizeUnit.MB.toBytes(1),
                 ByteSizeUnit.MB.toBytes(1),
                 1
@@ -184,7 +184,7 @@ public void testMultipartUpload() {
             new StreamContext((partIdx, partSize, position) -> {
                 InputStream stream = new ZeroInputStream(partSize);
                 streams.add(stream);
-                return new InputStreamContainer(stream, partSize, position, null);
+                return new InputStreamContainer(stream, partSize, position);
             }, ByteSizeUnit.MB.toBytes(1), ByteSizeUnit.MB.toBytes(1), 5),
             new StatsMetricPublisher()
         );
@@ -242,7 +242,7 @@ public void testMultipartUploadCorruption() {
                 // do nothing
             }, true, 0L, true, null),
             new StreamContext(
-                (partIdx, partSize, position) -> new InputStreamContainer(new ZeroInputStream(partSize), partSize, position, null),
+                (partIdx, partSize, position) -> new InputStreamContainer(new ZeroInputStream(partSize), partSize, position),
                 ByteSizeUnit.MB.toBytes(1),
                 ByteSizeUnit.MB.toBytes(1),
                 5

diff --git a/...ClusterTest/java/org/opensearch/remotestore/multipart/mocks/MockFsAsyncBlobContainer.java b/...ClusterTest/java/org/opensearch/remotestore/multipart/mocks/MockFsAsyncBlobContainer.java
@@ -128,12 +128,7 @@ public void readBlobAsync(String blobName, ActionListener<ReadContext> listener)
                 List<ReadContext.StreamPartCreator> blobPartStreams = new ArrayList<>();
                 for (int partNumber = 0; partNumber < numberOfParts; partNumber++) {
                     long offset = partNumber * partSize;
-                    InputStreamContainer blobPartStream = new InputStreamContainer(
-                        readBlob(blobName, offset, partSize),
-                        partSize,
-                        offset,
-                        null
-                    );
+                    InputStreamContainer blobPartStream = new InputStreamContainer(readBlob(blobName, offset, partSize), partSize, offset);
                     blobPartStreams.add(() -> CompletableFuture.completedFuture(blobPartStream));
                 }
                 ReadContext blobReadContext = new ReadContext.Builder(contentLength, blobPartStreams).build();

diff --git a/...src/main/java/org/opensearch/common/blobstore/AsyncMultiStreamEncryptedBlobContainer.java b/...src/main/java/org/opensearch/common/blobstore/AsyncMultiStreamEncryptedBlobContainer.java
@@ -168,7 +168,7 @@ private InputStreamContainer decryptInputStreamContainer(InputStreamContainer in
             long adjustedLength = decryptedStreamProvider.getAdjustedRange()[1] - adjustedPos + 1;
             final InputStream decryptedStream = decryptedStreamProvider.getDecryptedStreamProvider()
                 .apply(inputStreamContainer.getInputStream());
-            return new InputStreamContainer(decryptedStream, adjustedLength, adjustedPos, null);
+            return new InputStreamContainer(decryptedStream, adjustedLength, adjustedPos);
         }
     }
 }
diff --git a/server/src/main/java/org/opensearch/common/blobstore/EncryptedBlobContainer.java b/server/src/main/java/org/opensearch/common/blobstore/EncryptedBlobContainer.java
@@ -84,7 +84,7 @@ public long readBlobPreferredLength() {
     private void executeWrite(InputStream inputStream, long blobSize, CheckedBiConsumer<InputStream, Long, IOException> writeConsumer)
         throws IOException {
         T cryptoContext = cryptoHandler.initEncryptionMetadata();
-        InputStreamContainer streamContainer = new InputStreamContainer(inputStream, blobSize, 0, null);
+        InputStreamContainer streamContainer = new InputStreamContainer(inputStream, blobSize, 0);
         InputStreamContainer encryptedStream = cryptoHandler.createEncryptingStream(cryptoContext, streamContainer);
         long cryptoLength = cryptoHandler.estimateEncryptedLengthOfEntireContent(cryptoContext, blobSize);
         writeConsumer.accept(encryptedStream.getInputStream(), cryptoLength);

diff --git a/server/src/main/java/org/opensearch/common/blobstore/transfer/RemoteTransferContainer.java b/server/src/main/java/org/opensearch/common/blobstore/transfer/RemoteTransferContainer.java
@@ -221,7 +221,7 @@ private LocalStreamSupplier<InputStreamContainer> getMultipartStreamSupplier(
                     inputStream = offsetRangeInputStream;
                 }
 
-                return new InputStreamContainer(inputStream, size, position, null);
+                return new InputStreamContainer(inputStream, size, position);
             } catch (IOException e) {
                 log.error("Failed to create input stream", e);
                 throw e;

diff --git a/...est/java/org/opensearch/common/blobstore/AsyncMultiStreamEncryptedBlobContainerTests.java b/...est/java/org/opensearch/common/blobstore/AsyncMultiStreamEncryptedBlobContainerTests.java
@@ -53,7 +53,7 @@ public void testReadBlobAsync() throws Exception {
         final byte[] data = new byte[size];
         Randomness.get().nextBytes(data);
 
-        final InputStreamContainer inputStreamContainer = new InputStreamContainer(new ByteArrayInputStream(data), data.length, 0, null);
+        final InputStreamContainer inputStreamContainer = new InputStreamContainer(new ByteArrayInputStream(data), data.length, 0);
         final ListenerTestUtils.CountingCompletionListener<ReadContext> completionListener =
             new ListenerTestUtils.CountingCompletionListener<>();
         final CompletableFuture<InputStreamContainer> streamContainerFuture = CompletableFuture.completedFuture(inputStreamContainer);
@@ -99,7 +99,7 @@ public void testReadBlobAsyncException() throws Exception {
         // Objects needed for API call
         final byte[] data = new byte[size];
         Randomness.get().nextBytes(data);
-        final InputStreamContainer inputStreamContainer = new InputStreamContainer(new ByteArrayInputStream(data), data.length, 0, null);
+        final InputStreamContainer inputStreamContainer = new InputStreamContainer(new ByteArrayInputStream(data), data.length, 0);
         final ListenerTestUtils.CountingCompletionListener<ReadContext> completionListener =
             new ListenerTestUtils.CountingCompletionListener<>();
         final CompletableFuture<InputStreamContainer> streamContainerFuture = CompletableFuture.completedFuture(inputStreamContainer);

diff --git a/...c/test/java/org/opensearch/common/blobstore/stream/read/listener/FilePartWriterTests.java b/...c/test/java/org/opensearch/common/blobstore/stream/read/listener/FilePartWriterTests.java
@@ -32,7 +32,7 @@ public void testFilePartWriter() throws Exception {
         Path segmentFilePath = path.resolve(UUID.randomUUID().toString());
         int contentLength = 100;
         InputStream inputStream = new ByteArrayInputStream(randomByteArrayOfLength(contentLength));
-        InputStreamContainer inputStreamContainer = new InputStreamContainer(inputStream, inputStream.available(), 0, null);
+        InputStreamContainer inputStreamContainer = new InputStreamContainer(inputStream, inputStream.available(), 0);
 
         FilePartWriter.write(segmentFilePath, inputStreamContainer, UnaryOperator.identity());
 
@@ -45,7 +45,7 @@ public void testFilePartWriterWithOffset() throws Exception {
         int contentLength = 100;
         int offset = 10;
         InputStream inputStream = new ByteArrayInputStream(randomByteArrayOfLength(contentLength));
-        InputStreamContainer inputStreamContainer = new InputStreamContainer(inputStream, inputStream.available(), offset, null);
+        InputStreamContainer inputStreamContainer = new InputStreamContainer(inputStream, inputStream.available(), offset);
 
         FilePartWriter.write(segmentFilePath, inputStreamContainer, UnaryOperator.identity());
 
@@ -57,7 +57,7 @@ public void testFilePartWriterLargeInput() throws Exception {
         Path segmentFilePath = path.resolve(UUID.randomUUID().toString());
         int contentLength = 20 * 1024 * 1024;
         InputStream inputStream = new ByteArrayInputStream(randomByteArrayOfLength(contentLength));
-        InputStreamContainer inputStreamContainer = new InputStreamContainer(inputStream, contentLength, 0, null);
+        InputStreamContainer inputStreamContainer = new InputStreamContainer(inputStream, contentLength, 0);
 
         FilePartWriter.write(segmentFilePath, inputStreamContainer, UnaryOperator.identity());
 

diff --git a/...t/java/org/opensearch/common/blobstore/stream/read/listener/ReadContextListenerTests.java b/...t/java/org/opensearch/common/blobstore/stream/read/listener/ReadContextListenerTests.java
@@ -121,7 +121,7 @@ public int available() {
         blobPartStreams.add(
             NUMBER_OF_PARTS,
             () -> CompletableFuture.supplyAsync(
-                () -> new InputStreamContainer(badInputStream, PART_SIZE, PART_SIZE * NUMBER_OF_PARTS, null),
+                () -> new InputStreamContainer(badInputStream, PART_SIZE, PART_SIZE * NUMBER_OF_PARTS),
                 threadPool.generic()
             )
         );
@@ -174,7 +174,7 @@ public int read(byte[] b) throws IOException {
         blobPartStreams.add(
             NUMBER_OF_PARTS,
             () -> CompletableFuture.supplyAsync(
-                () -> new InputStreamContainer(assertingStream, PART_SIZE, PART_SIZE * NUMBER_OF_PARTS, null),
+                () -> new InputStreamContainer(assertingStream, PART_SIZE, PART_SIZE * NUMBER_OF_PARTS),
                 threadPool.generic()
             )
         );
@@ -219,7 +219,7 @@ private List<ReadContext.StreamPartCreator> initializeBlobPartStreams() {
             int finalPartNumber = partNumber;
             blobPartStreams.add(
                 () -> CompletableFuture.supplyAsync(
-                    () -> new InputStreamContainer(testStream, PART_SIZE, (long) finalPartNumber * PART_SIZE, null),
+                    () -> new InputStreamContainer(testStream, PART_SIZE, (long) finalPartNumber * PART_SIZE),
                     threadPool.generic()
                 )
             );