Merge branch 'test-s3-plugin' of https://github.com/skumawat2025/Open…

…Search into test-s3-plugin
skumawat2025 · Apr 9, 2024 · dec088b · dec088b
2 parents fc9c273 + 3332318
commit dec088b
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 39 deletions.
diff --git a/server/src/main/java/org/opensearch/index/translog/transfer/BlobStoreTransferService.java b/server/src/main/java/org/opensearch/index/translog/transfer/BlobStoreTransferService.java
@@ -78,9 +78,8 @@ public void uploadBlob(
     public void uploadBlob(final TransferFileSnapshot fileSnapshot, Iterable<String> remoteTransferPath, WritePriority writePriority)
         throws IOException {
         BlobPath blobPath = (BlobPath) remoteTransferPath;
-        Map<String, String> metadata = prepareFileMetadata(fileSnapshot);
         try (InputStream inputStream = fileSnapshot.inputStream()) {
-            blobStore.blobContainer(blobPath).writeBlobAtomicWithMetadata(fileSnapshot.getName(), inputStream, metadata, fileSnapshot.getContentLength(), true);
+            blobStore.blobContainer(blobPath).writeBlobAtomic(fileSnapshot.getName(), inputStream, fileSnapshot.getContentLength(), true);
         }
     }
 
@@ -96,14 +95,7 @@ public void uploadBlobs(
             if (!(blobStore.blobContainer(blobPath) instanceof AsyncMultiStreamBlobContainer)) {
                 uploadBlob(ThreadPool.Names.TRANSLOG_TRANSFER, fileSnapshot, blobPath, listener, writePriority);
             } else {
-                if(!(fileSnapshot instanceof FileSnapshot.CheckpointFileSnapshot)) {
-                    logger.info("uploading file = {}", fileSnapshot.getName());
-                    try {
-                        uploadBlob(fileSnapshot, listener, blobPath, writePriority);
-                    } catch (IOException e) {
-                        throw new RuntimeException(e);
-                    }
-                }
+                uploadBlob(fileSnapshot, listener, blobPath, writePriority);
             }
         });
 
@@ -128,9 +120,7 @@ private void uploadBlob(
         ActionListener<TransferFileSnapshot> listener,
         BlobPath blobPath,
         WritePriority writePriority
-    ) throws IOException {
-
-        Map<String, String> metadata = prepareFileMetadata(fileSnapshot);
+    ) {
 
         try {
             ChannelFactory channelFactory = FileChannel::open;
@@ -151,8 +141,7 @@ private void uploadBlob(
                 writePriority,
                 (size, position) -> new OffsetRangeFileInputStream(fileSnapshot.getPath(), size, position),
                 Objects.requireNonNull(fileSnapshot.getChecksum()),
-                remoteIntegrityEnabled,
-                metadata
+                remoteIntegrityEnabled
             );
             ActionListener<Void> completionListener = ActionListener.wrap(resp -> listener.onResponse(fileSnapshot), ex -> {
                 logger.error(() -> new ParameterizedMessage("Failed to upload blob {}", fileSnapshot.getName()), ex);
@@ -171,7 +160,6 @@ private void uploadBlob(
             ((AsyncMultiStreamBlobContainer) blobStore.blobContainer(blobPath)).asyncBlobUpload(writeContext, completionListener);
 
         } catch (Exception e) {
-            logger.info("Exception while uploading file = {} with metadata", fileSnapshot.getName());
             logger.error(() -> new ParameterizedMessage("Failed to upload blob {}", fileSnapshot.getName()), e);
             listener.onFailure(new FileTransferException(fileSnapshot, e));
         } finally {

diff --git a/server/src/main/java/org/opensearch/index/translog/transfer/TranslogTransferManager.java b/server/src/main/java/org/opensearch/index/translog/transfer/TranslogTransferManager.java
@@ -14,7 +14,6 @@
 import org.apache.lucene.store.OutputStreamIndexOutput;
 import org.opensearch.action.LatchedActionListener;
 import org.opensearch.common.SetOnce;
-import org.opensearch.common.blobstore.BlobDownloadResponse;
 import org.opensearch.common.blobstore.BlobMetadata;
 import org.opensearch.common.blobstore.BlobPath;
 import org.opensearch.common.blobstore.stream.write.WritePriority;
@@ -37,12 +36,7 @@
 import java.io.InputStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -114,9 +108,7 @@ public boolean transferSnapshot(TransferSnapshot transferSnapshot, TranslogTrans
 
         try {
             toUpload.addAll(fileTransferTracker.exclusionFilter(transferSnapshot.getTranslogFileSnapshots()));
-            //skip checkpoint files...
-            //toUpload.addAll(fileTransferTracker.exclusionFilter((transferSnapshot.getCheckpointFileSnapshots())));
-
+            toUpload.addAll(fileTransferTracker.exclusionFilter((transferSnapshot.getCheckpointFileSnapshots())));
             if (toUpload.isEmpty()) {
                 logger.trace("Nothing to upload for transfer");
                 return true;
@@ -242,15 +234,15 @@ public boolean downloadTranslog(String primaryTerm, String generation, Path loca
             location
         );
         // Download Checkpoint file from remote to local FS
-        //String ckpFileName = Translog.getCommitCheckpointFileName(Long.parseLong(generation));
-        //downloadToFS(ckpFileName, location, primaryTerm);
+        String ckpFileName = Translog.getCommitCheckpointFileName(Long.parseLong(generation));
+        downloadToFS(ckpFileName, location, primaryTerm);
         // Download translog file from remote to local FS
         String translogFilename = Translog.getFilename(Long.parseLong(generation));
-        downloadToFS(translogFilename, location, primaryTerm, generation);
+        downloadToFS(translogFilename, location, primaryTerm);
         return true;
     }
 
-    private void downloadToFS(String fileName, Path location, String primaryTerm, String generation) throws IOException {
+    private void downloadToFS(String fileName, Path location, String primaryTerm) throws IOException {
         Path filePath = location.resolve(fileName);
         // Here, we always override the existing file if present.
         // We need to change this logic when we introduce incremental download
@@ -260,16 +252,8 @@ private void downloadToFS(String fileName, Path location, String primaryTerm, St
 
         boolean downloadStatus = false;
         long bytesToRead = 0, downloadStartTime = System.nanoTime();
-        BlobDownloadResponse downloaded = transferService.downloadBlobWithMetadata(remoteDataTransferPath.add(primaryTerm), fileName);
-        try {
+        try (InputStream inputStream = transferService.downloadBlob(remoteDataTransferPath.add(primaryTerm), fileName)) {
             // Capture number of bytes for stats before reading
-            InputStream inputStream = downloaded.getInputStream();
-            Map<String, String> metadata = downloaded.getMetadata();
-
-            logger.info("downloaded translog for fileName = {}, metadata = {}", fileName, metadata);
-
-            applyMetadataToCkpFile(metadata, location, generation, fileName);
-
             bytesToRead = inputStream.available();
             Files.copy(inputStream, filePath);
             downloadStatus = true;