resolve comments

Signed-off-by: Sandeep Kumawat <skumwt@amazon.com>
opensearch-project · Apr 9, 2024 · b528cfa · b528cfa
1 parent ca99c35
commit b528cfa
Show file tree

Hide file tree

Showing 6 changed files with 19 additions and 19 deletions.
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
@@ -177,15 +177,7 @@ public long readBlobPreferredLength() {
      */
     @Override
     public void writeBlob(String blobName, InputStream inputStream, long blobSize, boolean failIfAlreadyExists) throws IOException {
-        assert inputStream.markSupported() : "No mark support on inputStream breaks the S3 SDK's ability to retry requests";
-        SocketAccess.doPrivilegedIOException(() -> {
-            if (blobSize <= getLargeBlobThresholdInBytes()) {
-                executeSingleUpload(blobStore, buildKey(blobName), inputStream, blobSize, null);
-            } else {
-                executeMultipartUpload(blobStore, buildKey(blobName), inputStream, blobSize, null);
-            }
-            return null;
-        });
+        writeBlobWithMetadata(blobName, inputStream, null, blobSize, failIfAlreadyExists);
     }
 
     /**
@@ -196,7 +188,7 @@ public void writeBlob(String blobName, InputStream inputStream, long blobSize, b
     public void writeBlobWithMetadata(
         String blobName,
         InputStream inputStream,
-        Map<String, String> metadata,
+        @Nullable Map<String, String> metadata,
         long blobSize,
         boolean failIfAlreadyExists
     ) throws IOException {
@@ -611,7 +603,7 @@ void executeSingleUpload(
             .acl(blobStore.getCannedACL())
             .overrideConfiguration(o -> o.addMetricPublisher(blobStore.getStatsMetricPublisher().putObjectMetricPublisher));
 
-        if (metadata != null) {
+        if (metadata != null && !metadata.isEmpty()) {
             putObjectRequestBuilder = putObjectRequestBuilder.metadata(metadata);
         }
         if (blobStore.serverSideEncryption()) {
@@ -668,7 +660,7 @@ void executeMultipartUpload(
             .acl(blobStore.getCannedACL())
             .overrideConfiguration(o -> o.addMetricPublisher(blobStore.getStatsMetricPublisher().multipartUploadMetricCollector));
 
-        if (metadata != null) {
+        if (metadata != null && !metadata.isEmpty()) {
             createMultipartUploadRequestBuilder.metadata(metadata);
         }
 

diff --git a/...ins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3RetryingInputStream.java b/...ins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3RetryingInputStream.java
@@ -269,7 +269,7 @@ boolean isAborted() {
         return isStreamAborted.get();
     }
 
-    public Map<String, String> getMetadata() {
+    Map<String, String> getMetadata() {
         return this.metadata;
     }
 }
diff --git a/...epository-s3/src/main/java/org/opensearch/repositories/s3/async/AsyncTransferManager.java b/...epository-s3/src/main/java/org/opensearch/repositories/s3/async/AsyncTransferManager.java
@@ -129,9 +129,12 @@ private void uploadInParts(
 
         CreateMultipartUploadRequest.Builder createMultipartUploadRequestBuilder = CreateMultipartUploadRequest.builder()
             .bucket(uploadRequest.getBucket())
-            .metadata(uploadRequest.getMetadata())
             .key(uploadRequest.getKey())
             .overrideConfiguration(o -> o.addMetricPublisher(statsMetricPublisher.multipartUploadMetricCollector));
+
+        if (uploadRequest.getMetadata() != null && !uploadRequest.getMetadata().isEmpty()) {
+            createMultipartUploadRequestBuilder.metadata(uploadRequest.getMetadata());
+        }
         if (uploadRequest.doRemoteDataIntegrityCheck()) {
             createMultipartUploadRequestBuilder.checksumAlgorithm(ChecksumAlgorithm.CRC32);
         }
@@ -325,10 +328,13 @@ private void uploadInOneChunk(
     ) {
         PutObjectRequest.Builder putObjectRequestBuilder = PutObjectRequest.builder()
             .bucket(uploadRequest.getBucket())
-            .metadata(uploadRequest.getMetadata())
             .key(uploadRequest.getKey())
             .contentLength(uploadRequest.getContentLength())
             .overrideConfiguration(o -> o.addMetricPublisher(statsMetricPublisher.putObjectMetricPublisher));
+
+        if (uploadRequest.getMetadata() != null && !uploadRequest.getMetadata().isEmpty()) {
+            putObjectRequestBuilder.metadata(uploadRequest.getMetadata());
+        }
         if (uploadRequest.doRemoteDataIntegrityCheck()) {
             putObjectRequestBuilder.checksumAlgorithm(ChecksumAlgorithm.CRC32);
             putObjectRequestBuilder.checksumCRC32(base64StringFromLong(uploadRequest.getExpectedChecksum()));

diff --git a/plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/async/UploadRequest.java b/plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/async/UploadRequest.java
@@ -9,6 +9,7 @@
 package org.opensearch.repositories.s3.async;
 
 import org.opensearch.common.CheckedConsumer;
+import org.opensearch.common.Nullable;
 import org.opensearch.common.blobstore.stream.write.WritePriority;
 
 import java.io.IOException;
@@ -49,7 +50,7 @@ public UploadRequest(
         boolean doRemoteDataIntegrityCheck,
         Long expectedChecksum,
         boolean uploadRetryEnabled,
-        Map<String, String> metadata
+        @Nullable Map<String, String> metadata
     ) {
         this.bucket = bucket;
         this.key = key;

diff --git a/server/src/main/java/org/opensearch/common/blobstore/BlobContainer.java b/server/src/main/java/org/opensearch/common/blobstore/BlobContainer.java
@@ -32,6 +32,7 @@
 
 package org.opensearch.common.blobstore;
 
+import org.opensearch.common.Nullable;
 import org.opensearch.common.annotation.ExperimentalApi;
 import org.opensearch.core.action.ActionListener;
 
@@ -166,7 +167,7 @@ default long readBlobPreferredLength() {
     default void writeBlobWithMetadata(
         String blobName,
         InputStream inputStream,
-        Map<String, String> metadata,
+        @Nullable Map<String, String> metadata,
         long blobSize,
         boolean failIfAlreadyExists
     ) throws IOException {
@@ -219,7 +220,7 @@ default void writeBlobWithMetadata(
     default void writeBlobAtomicWithMetadata(
         String blobName,
         InputStream inputStream,
-        Map<String, String> metadata,
+        @Nullable Map<String, String> metadata,
         long blobSize,
         boolean failIfAlreadyExists
     ) throws IOException {

diff --git a/server/src/main/java/org/opensearch/common/blobstore/stream/write/WriteContext.java b/server/src/main/java/org/opensearch/common/blobstore/stream/write/WriteContext.java
@@ -52,7 +52,7 @@ private WriteContext(
         CheckedConsumer<Boolean, IOException> uploadFinalizer,
         boolean doRemoteDataIntegrityCheck,
         @Nullable Long expectedChecksum,
-        Map<String, String> metadata
+        @Nullable Map<String, String> metadata
     ) {
         this.fileName = fileName;
         this.streamContextSupplier = streamContextSupplier;