address review comments

Signed-off-by: Sandeep Kumawat <[email protected]>

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

@@ -90,7 +90,7 @@ public interface BlobContainer {
      */
     @ExperimentalApi
     default InputStreamWithMetadata readBlobWithMetadata(String blobName) throws IOException {
-        return null;
+        throw new UnsupportedOperationException("readBlobWithMetadata is not implemented yet");
     };
 
     /**

server/src/main/java/org/opensearch/index/translog/RemoteFsTranslog.java

@@ -23,11 +23,11 @@
 import org.opensearch.index.remote.RemoteStorePathStrategy;
 import org.opensearch.index.remote.RemoteTranslogTransferTracker;
 import org.opensearch.index.seqno.SequenceNumbers;
-import org.opensearch.index.translog.transfer.BaseTranslogTransferManager;
 import org.opensearch.index.translog.transfer.BlobStoreTransferService;
 import org.opensearch.index.translog.transfer.FileTransferTracker;
 import org.opensearch.index.translog.transfer.TransferSnapshot;
 import org.opensearch.index.translog.transfer.TranslogSyncSnapshot;
+import org.opensearch.index.translog.transfer.TranslogTransferManager;
 import org.opensearch.index.translog.transfer.TranslogTransferManagerFactory;
 import org.opensearch.index.translog.transfer.TranslogTransferMetadata;
 import org.opensearch.index.translog.transfer.listener.TranslogTransferListener;
@@ -70,7 +70,7 @@
 public class RemoteFsTranslog extends Translog {
 
     private final Logger logger;
-    private final BaseTranslogTransferManager translogTransferManager;
+    private final TranslogTransferManager translogTransferManager;
     private final FileTransferTracker fileTransferTracker;
     private final BooleanSupplier startedPrimarySupplier;
     private final RemoteTranslogTransferTracker remoteTranslogTransferTracker;
@@ -169,8 +169,7 @@ RemoteTranslogTransferTracker getRemoteTranslogTracker() {
     }
 
     private boolean isCkpAsTranslogMetadata(Supplier<Version> minNodeVersionSupplier, BlobStoreRepository blobStoreRepository) {
-        boolean isBlobMetadataSupported = blobStoreRepository.blobStore().isBlobMetadataSupported();
-        return isBlobMetadataSupported && Version.CURRENT.compareTo(minNodeVersionSupplier.get()) <= 0;
+        return blobStoreRepository.blobStore().isBlobMetadataSupported() && Version.CURRENT.compareTo(minNodeVersionSupplier.get()) <= 0;
     }
 
     public static void download(
@@ -193,7 +192,7 @@ public static void download(
         // TODO: To be revisited as part of https://github.com/opensearch-project/OpenSearch/issues/7567
         RemoteTranslogTransferTracker remoteTranslogTransferTracker = new RemoteTranslogTransferTracker(shardId, 1000);
         FileTransferTracker fileTransferTracker = new FileTransferTracker(shardId, remoteTranslogTransferTracker, false);
-        BaseTranslogTransferManager translogTransferManager = buildTranslogTransferManager(
+        TranslogTransferManager translogTransferManager = buildTranslogTransferManager(
             blobStoreRepository,
             threadPool,
             shardId,
@@ -207,7 +206,7 @@ public static void download(
         logger.trace(remoteTranslogTransferTracker.toString());
     }
 
-    static void download(BaseTranslogTransferManager translogTransferManager, Path location, Logger logger, boolean seedRemote)
+    static void download(TranslogTransferManager translogTransferManager, Path location, Logger logger, boolean seedRemote)
         throws IOException {
         /*
         In Primary to Primary relocation , there can be concurrent upload and download of translog.
@@ -235,7 +234,7 @@ static void download(BaseTranslogTransferManager translogTransferManager, Path l
         throw ex;
     }
 
-    private static void downloadOnce(BaseTranslogTransferManager translogTransferManager, Path location, Logger logger, boolean seedRemote)
+    private static void downloadOnce(TranslogTransferManager translogTransferManager, Path location, Logger logger, boolean seedRemote)
         throws IOException {
         logger.debug("Downloading translog files from remote");
         RemoteTranslogTransferTracker statsTracker = translogTransferManager.getRemoteTranslogTransferTracker();
@@ -300,15 +299,15 @@ private static boolean isEmptyTranslog(Checkpoint checkpoint) {
             && checkpoint.numOps == 0;
     }
 
-    public static BaseTranslogTransferManager buildTranslogTransferManager(
+    public static TranslogTransferManager buildTranslogTransferManager(
         BlobStoreRepository blobStoreRepository,
         ThreadPool threadPool,
         ShardId shardId,
         FileTransferTracker fileTransferTracker,
         RemoteTranslogTransferTracker tracker,
         RemoteStorePathStrategy pathStrategy,
         RemoteStoreSettings remoteStoreSettings,
-        boolean shouldUploadTranslogCkpAsMetadata
+        boolean ckpAsTranslogMetadata
     ) {
         assert Objects.nonNull(pathStrategy);
         String indexUUID = shardId.getIndex().getUUID();
@@ -338,7 +337,7 @@ public static BaseTranslogTransferManager buildTranslogTransferManager(
             fileTransferTracker,
             tracker,
             remoteStoreSettings,
-            shouldUploadTranslogCkpAsMetadata
+            ckpAsTranslogMetadata
         );
     }
 
@@ -423,7 +422,7 @@ private boolean prepareAndUpload(Long primaryTerm, Long generation) throws IOExc
     private boolean upload(long primaryTerm, long generation, long maxSeqNo) throws IOException {
         logger.trace("uploading translog for primary term {} generation {}", primaryTerm, generation);
         try {
-            TranslogSyncSnapshot transferSnapshotProvider = new TranslogSyncSnapshot.Builder(
+            TranslogSyncSnapshot translogSyncSnapshot = new TranslogSyncSnapshot.Builder(
                 primaryTerm,
                 generation,
                 location,
@@ -432,7 +431,7 @@ private boolean upload(long primaryTerm, long generation, long maxSeqNo) throws
                 config.getNodeId()
             ).build();
             return translogTransferManager.transferSnapshot(
-                transferSnapshotProvider,
+                translogSyncSnapshot,
                 new RemoteFsTranslogTransferListener(generation, primaryTerm, maxSeqNo)
             );
         } finally {
@@ -623,7 +622,7 @@ public static void cleanup(
         // TODO: To be revisited as part of https://github.com/opensearch-project/OpenSearch/issues/7567
         RemoteTranslogTransferTracker remoteTranslogTransferTracker = new RemoteTranslogTransferTracker(shardId, 1000);
         FileTransferTracker fileTransferTracker = new FileTransferTracker(shardId, remoteTranslogTransferTracker, false);
-        BaseTranslogTransferManager translogTransferManager = buildTranslogTransferManager(
+        TranslogTransferManager translogTransferManager = buildTranslogTransferManager(
             blobStoreRepository,
             threadPool,
             shardId,

server/src/main/java/org/opensearch/index/translog/transfer/FileSnapshot.java

@@ -42,13 +42,6 @@ public class FileSnapshot implements Closeable {
     @Nullable
     private Map<String, String> metadata;
 
-    private FileSnapshot(Path path) throws IOException {
-        Objects.requireNonNull(path);
-        this.name = path.getFileName().toString();
-        this.path = path;
-        this.fileChannel = FileChannel.open(path, StandardOpenOption.READ);
-    }
-
     private FileSnapshot(Path path, Map<String, String> metadata) throws IOException {
         Objects.requireNonNull(path);
         this.name = path.getFileName().toString();
@@ -125,9 +118,7 @@ public static class TransferFileSnapshot extends FileSnapshot {
         private Long checksum;
 
         public TransferFileSnapshot(Path path, long primaryTerm, Long checksum) throws IOException {
-            super(path);
-            this.primaryTerm = primaryTerm;
-            this.checksum = checksum;
+            this(path, primaryTerm, checksum, null);
         }
 
         private TransferFileSnapshot(Path path, long primaryTerm, Long checksum, Map<String, String> metadata) throws IOException {

server/src/main/java/org/opensearch/index/translog/transfer/FileTransferTracker.java

@@ -14,23 +14,22 @@
 import org.opensearch.index.remote.RemoteTranslogTransferTracker;
 import org.opensearch.index.translog.transfer.listener.FileTransferListener;
 
-import java.io.IOException;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.LongSupplier;
 import java.util.stream.Collectors;
 
 /**
  * FileTransferTracker keeps track of generational translog files uploaded to the remote translog store
  */
 public class FileTransferTracker implements FileTransferListener {
 
-    private final ConcurrentHashMap<String, TransferState> fileTransferTracker;
-    private final ConcurrentHashMap<Long, TransferState> generationTransferTracker;
+    private final Map<String, TransferState> fileTransferTracker;
+    private final Map<Long, TransferState> generationTransferTracker;
     private final RemoteTranslogTransferTracker remoteTranslogTransferTracker;
     private Map<String, Long> bytesForTlogCkpFileToUpload;
     private long fileTransferStartTime = -1;
@@ -64,26 +63,17 @@ void recordBytesForFiles(Set<TranslogCheckpointSnapshot> toUpload) {
         });
     }
 
-    private void recordFileContentLength(String fileName, ThrowingSupplier<Long, IOException> contentLengthSupplier) {
-        try {
-            if (!uploaded(fileName)) {
-                bytesForTlogCkpFileToUpload.put(fileName, contentLengthSupplier.get());
-            }
-        } catch (IOException ignored) {
-            bytesForTlogCkpFileToUpload.put(fileName, 0L);
+    private void recordFileContentLength(String fileName, LongSupplier contentLengthSupplier) {
+        if (!uploaded(fileName)) {
+            bytesForTlogCkpFileToUpload.put(fileName, contentLengthSupplier.getAsLong());
         }
     }
 
-    @FunctionalInterface
-    private interface ThrowingSupplier<T, E extends Exception> {
-        T get() throws E;
-    }
-
     long getTotalBytesToUpload() {
         return bytesForTlogCkpFileToUpload.values().stream().reduce(0L, Long::sum);
     }
 
-    private <K> void updateTransferTracker(ConcurrentHashMap<K, TransferState> tracker, K key, TransferState targetState) {
+    private <K> void updateTransferState(Map<K, TransferState> tracker, K key, TransferState targetState) {
         tracker.compute(key, (k, v) -> {
             if (v == null || v.validateNextState(targetState)) {
                 return targetState;
@@ -98,7 +88,7 @@ void addGeneration(long generation, boolean success) {
     }
 
     private void addGeneration(long generation, TransferState targetState) {
-        updateTransferTracker(generationTransferTracker, generation, targetState);
+        updateTransferState(generationTransferTracker, generation, targetState);
     }
 
     void add(String file, boolean success) {
@@ -107,7 +97,7 @@ void add(String file, boolean success) {
     }
 
     private void add(String file, TransferState targetState) {
-        updateTransferTracker(fileTransferTracker, file, targetState);
+        updateTransferState(fileTransferTracker, file, targetState);
     }
 
     @Override
@@ -132,6 +122,16 @@ public void onFailure(TranslogCheckpointSnapshot fileSnapshot, Exception e) {
         remoteTranslogTransferTracker.addUploadTimeInMillis(durationInMillis);
         updateTransferStats(fileSnapshot, false);
         addGeneration(fileSnapshot.getGeneration(), TransferState.FAILED);
+
+        if (!ckpAsTranslogMetadata) {
+            assert e instanceof TranslogTransferException;
+            TranslogTransferException exception = (TranslogTransferException) e;
+            Set<FileSnapshot.TransferFileSnapshot> failedFiles = exception.getFailedFiles();
+            Set<FileSnapshot.TransferFileSnapshot> successFiles = exception.getSuccessFiles();
+            assert failedFiles.isEmpty() == false;
+            failedFiles.forEach(failedFile -> add(failedFile.getName(), false));
+            successFiles.forEach(successFile -> add(successFile.getName(), true));
+        }
     }
 
     private void updateTransferStats(TranslogCheckpointSnapshot fileSnapshot, boolean isSuccess) {
@@ -181,23 +181,19 @@ Set<TranslogCheckpointSnapshot> exclusionFilter(Set<TranslogCheckpointSnapshot>
     }
 
     Set<Long> allUploadedGeneration() {
-        Set<Long> successGenFileTransferTracker = new HashSet<>();
-        generationTransferTracker.forEach((k, v) -> {
-            if (v == TransferState.SUCCESS) {
-                successGenFileTransferTracker.add(k);
-            }
-        });
-        return successGenFileTransferTracker;
+        return getSuccessfulKeys(generationTransferTracker);
     }
 
     public Set<String> allUploaded() {
-        Set<String> successFileTransferTracker = new HashSet<>();
-        fileTransferTracker.forEach((k, v) -> {
-            if (v == TransferState.SUCCESS) {
-                successFileTransferTracker.add(k);
-            }
-        });
-        return successFileTransferTracker;
+        return getSuccessfulKeys(fileTransferTracker);
+    }
+
+    private <K> Set<K> getSuccessfulKeys(Map<K, TransferState> tracker) {
+        return tracker.entrySet()
+            .stream()
+            .filter(entry -> entry.getValue() == TransferState.SUCCESS)
+            .map(Map.Entry::getKey)
+            .collect(Collectors.toSet());
     }
 
     /**

server/src/main/java/org/opensearch/index/translog/transfer/TranslogCheckpointSnapshot.java

@@ -21,7 +21,7 @@
 import java.util.HashMap;
 import java.util.Map;
 
-import static org.opensearch.index.translog.transfer.BaseTranslogTransferManager.CHECKPOINT_FILE_DATA_KEY;
+import static org.opensearch.index.translog.transfer.TranslogTransferManager.CHECKPOINT_FILE_DATA_KEY;
 
 /**
  * Snapshot of a single translog generational files that gets transferred
@@ -70,15 +70,19 @@ String getCheckpointFileName() {
         return checkpointPath.getFileName().toString();
     }
 
-    long getTranslogFileContentLength() throws IOException {
+    long getTranslogFileContentLength() {
         try (FileChannel fileChannel = FileChannel.open(translogPath, StandardOpenOption.READ)) {
             return fileChannel.size();
+        } catch (IOException ignore) {
+            return 0L;
         }
     }
 
-    long getCheckpointFileContentLength() throws IOException {
+    long getCheckpointFileContentLength() {
         try (FileChannel fileChannel = FileChannel.open(checkpointPath, StandardOpenOption.READ)) {
             return fileChannel.size();
+        } catch (IOException ignore) {
+            return 0L;
         }
     }
 

server/src/main/java/org/opensearch/index/translog/transfer/TranslogCkpAsMetadataFileTransferManager.java

@@ -28,7 +28,7 @@
  *
  * @opensearch.internal
  */
-public class TranslogCkpAsMetadataFileTransferManager extends BaseTranslogTransferManager {
+public class TranslogCkpAsMetadataFileTransferManager extends TranslogTransferManager {
 
     TransferService transferService;
 
@@ -55,14 +55,13 @@ public TranslogCkpAsMetadataFileTransferManager(
 
     @Override
     public void transferTranslogCheckpointSnapshot(
-        Set<TranslogCheckpointSnapshot> generationalSnapshotList,
+        Set<TranslogCheckpointSnapshot> toUpload,
         Map<Long, BlobPath> blobPathMap,
-        LatchedActionListener<TranslogCheckpointSnapshot> latchedActionListener,
-        WritePriority writePriority
+        LatchedActionListener<TranslogCheckpointSnapshot> latchedActionListener
     ) throws Exception {
         Set<TransferFileSnapshot> filesToUpload = new HashSet<>();
         Map<TransferFileSnapshot, TranslogCheckpointSnapshot> fileToGenerationSnapshotMap = new HashMap<>();
-        for (TranslogCheckpointSnapshot translogCheckpointSnapshot : generationalSnapshotList) {
+        for (TranslogCheckpointSnapshot translogCheckpointSnapshot : toUpload) {
             TransferFileSnapshot transferFileSnapshot = translogCheckpointSnapshot.getTranslogFileSnapshotWithMetadata();
             fileToGenerationSnapshotMap.put(transferFileSnapshot, translogCheckpointSnapshot);
             filesToUpload.add(transferFileSnapshot);
@@ -72,16 +71,12 @@ public void transferTranslogCheckpointSnapshot(
             ex -> {
                 assert ex instanceof FileTransferException;
                 FileTransferException e = (FileTransferException) ex;
+                TransferFileSnapshot failedSnapshot = e.getFileSnapshot();
                 latchedActionListener.onFailure(
-                    new TranslogTransferException(fileToGenerationSnapshotMap.get(e.getFileSnapshot()), ex, null, null)
+                    new TranslogTransferException(fileToGenerationSnapshotMap.get(failedSnapshot), ex, Set.of(failedSnapshot), null)
                 );
             }
         );
         transferService.uploadBlobs(filesToUpload, blobPathMap, actionListener, WritePriority.HIGH);
     }
-
-    @Override
-    public boolean updateFileNameTransferTracker() {
-        return false;
-    }
 }

server/src/main/java/org/opensearch/index/translog/transfer/TranslogCkpFilesTransferManager.java

@@ -31,7 +31,7 @@
  *
  * @opensearch.internal
  */
-public class TranslogCkpFilesTransferManager extends BaseTranslogTransferManager {
+public class TranslogCkpFilesTransferManager extends TranslogTransferManager {
 
     TransferService transferService;
     FileTransferTracker fileTransferTracker;
@@ -60,12 +60,11 @@ public TranslogCkpFilesTransferManager(
 
     @Override
     public void transferTranslogCheckpointSnapshot(
-        Set<TranslogCheckpointSnapshot> generationalSnapshotList,
+        Set<TranslogCheckpointSnapshot> toUpload,
         Map<Long, BlobPath> blobPathMap,
-        LatchedActionListener<TranslogCheckpointSnapshot> latchedActionListener,
-        WritePriority writePriority
+        LatchedActionListener<TranslogCheckpointSnapshot> latchedActionListener
     ) throws Exception {
-        for (TranslogCheckpointSnapshot tlogAndCkpTransferFileSnapshot : generationalSnapshotList) {
+        for (TranslogCheckpointSnapshot tlogAndCkpTransferFileSnapshot : toUpload) {
             Set<TransferFileSnapshot> filesToUpload = new HashSet<>();
             Set<Exception> exceptionList = ConcurrentCollections.newConcurrentSet();
 
@@ -111,12 +110,7 @@ public void transferTranslogCheckpointSnapshot(
                     }
                 }
             });
-            transferService.uploadBlobs(filesToUpload, blobPathMap, actionListener, writePriority);
+            transferService.uploadBlobs(filesToUpload, blobPathMap, actionListener, WritePriority.HIGH);
         }
     }
-
-    @Override
-    public boolean updateFileNameTransferTracker() {
-        return true;
-    }
 }