[Remote Store] Avoiding Translog Deletion in case of relocating shards.

server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteIndexPrimaryRelocationIT.java

@@ -64,7 +64,6 @@ protected Settings featureFlagSettings() {
             .build();
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/opensearch-project/OpenSearch/issues/9191")
     public void testPrimaryRelocationWhileIndexing() throws Exception {
         super.testPrimaryRelocationWhileIndexing();
     }

server/src/main/java/org/opensearch/index/engine/EngineConfig.java

@@ -53,6 +53,7 @@
 import org.opensearch.index.codec.CodecSettings;
 import org.opensearch.index.mapper.ParsedDocument;
 import org.opensearch.index.seqno.RetentionLeases;
+import org.opensearch.index.shard.IndexShard;
 import org.opensearch.index.store.Store;
 import org.opensearch.index.translog.InternalTranslogFactory;
 import org.opensearch.index.translog.TranslogConfig;
@@ -65,7 +66,6 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
-import java.util.function.BooleanSupplier;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
 
@@ -105,7 +105,7 @@ public final class EngineConfig {
     private final LongSupplier globalCheckpointSupplier;
     private final Supplier<RetentionLeases> retentionLeasesSupplier;
     private final boolean isReadOnlyReplica;
-    private final BooleanSupplier primaryModeSupplier;
+    private final IndexShard.IndexShardConfigSupplier indexShardConfigSupplier;
     private final Comparator<LeafReader> leafSorter;
 
     /**
@@ -266,7 +266,7 @@ private EngineConfig(Builder builder) {
         this.primaryTermSupplier = builder.primaryTermSupplier;
         this.tombstoneDocSupplier = builder.tombstoneDocSupplier;
         this.isReadOnlyReplica = builder.isReadOnlyReplica;
-        this.primaryModeSupplier = builder.primaryModeSupplier;
+        this.indexShardConfigSupplier = builder.indexShardConfigSupplier;
         this.translogFactory = builder.translogFactory;
         this.leafSorter = builder.leafSorter;
     }
@@ -477,8 +477,8 @@ public boolean isReadOnlyReplica() {
      * Returns the underlying primaryModeSupplier.
      * @return the primary mode supplier.
      */
-    public BooleanSupplier getPrimaryModeSupplier() {
-        return primaryModeSupplier;
+    public IndexShard.IndexShardConfigSupplier getIndexShardConfigSupplier() {
+        return indexShardConfigSupplier;
     }
 
     /**
@@ -555,7 +555,7 @@ public static class Builder {
         private TombstoneDocSupplier tombstoneDocSupplier;
         private TranslogDeletionPolicyFactory translogDeletionPolicyFactory;
         private boolean isReadOnlyReplica;
-        private BooleanSupplier primaryModeSupplier;
+        private IndexShard.IndexShardConfigSupplier indexShardConfigSupplier;
         private TranslogFactory translogFactory = new InternalTranslogFactory();
         Comparator<LeafReader> leafSorter;
 
@@ -679,8 +679,8 @@ public Builder readOnlyReplica(boolean isReadOnlyReplica) {
             return this;
         }
 
-        public Builder primaryModeSupplier(BooleanSupplier primaryModeSupplier) {
-            this.primaryModeSupplier = primaryModeSupplier;
+        public Builder indexShardConfig(IndexShard.IndexShardConfigSupplier indexShardConfigSupplier) {
+            this.indexShardConfigSupplier = indexShardConfigSupplier;
             return this;
         }
 

server/src/main/java/org/opensearch/index/engine/EngineConfigFactory.java

@@ -27,6 +27,7 @@
 import org.opensearch.index.codec.CodecServiceFactory;
 import org.opensearch.index.mapper.MapperService;
 import org.opensearch.index.seqno.RetentionLeases;
+import org.opensearch.index.shard.IndexShard;
 import org.opensearch.index.store.Store;
 import org.opensearch.index.translog.TranslogConfig;
 import org.opensearch.index.translog.TranslogDeletionPolicyFactory;
@@ -40,7 +41,6 @@
 import java.util.Comparator;
 import java.util.List;
 import java.util.Optional;
-import java.util.function.BooleanSupplier;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
 
@@ -152,7 +152,7 @@ public EngineConfig newEngineConfig(
         LongSupplier primaryTermSupplier,
         EngineConfig.TombstoneDocSupplier tombstoneDocSupplier,
         boolean isReadOnlyReplica,
-        BooleanSupplier primaryModeSupplier,
+        IndexShard.IndexShardConfigSupplier indexShardConfigSupplier,
         TranslogFactory translogFactory,
         Comparator<LeafReader> leafSorter
     ) {
@@ -185,7 +185,7 @@ public EngineConfig newEngineConfig(
             .primaryTermSupplier(primaryTermSupplier)
             .tombstoneDocSupplier(tombstoneDocSupplier)
             .readOnlyReplica(isReadOnlyReplica)
-            .primaryModeSupplier(primaryModeSupplier)
+            .indexShardConfig(indexShardConfigSupplier)
             .translogFactory(translogFactory)
             .leafSorter(leafSorter)
             .build();

server/src/main/java/org/opensearch/index/engine/InternalEngine.java

@@ -292,7 +292,7 @@ public void onFailure(String reason, Exception ex) {
                     new CompositeTranslogEventListener(Arrays.asList(internalTranslogEventListener, translogEventListener), shardId),
                     this::ensureOpen,
                     engineConfig.getTranslogFactory(),
-                    engineConfig.getPrimaryModeSupplier()
+                    engineConfig.getIndexShardConfigSupplier()
                 );
                 this.translogManager = translogManagerRef;
                 this.softDeletesPolicy = newSoftDeletesPolicy();

server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java

@@ -118,7 +118,7 @@ public void onAfterTranslogSync() {
                 },
                 this,
                 engineConfig.getTranslogFactory(),
-                engineConfig.getPrimaryModeSupplier()
+                engineConfig.getIndexShardConfigSupplier()
             );
             this.translogManager = translogManagerRef;
         } catch (IOException e) {

server/src/main/java/org/opensearch/index/engine/NoOpEngine.java

@@ -203,7 +203,7 @@ public void trimUnreferencedTranslogFiles() throws TranslogException {
                                         engineConfig.getGlobalCheckpointSupplier(),
                                         engineConfig.getPrimaryTermSupplier(),
                                         seqNo -> {},
-                                        engineConfig.getPrimaryModeSupplier()
+                                        engineConfig.getIndexShardConfigSupplier()
                                     )
                             ) {
                                 translog.trimUnreferencedReaders();

server/src/main/java/org/opensearch/index/engine/ReadOnlyEngine.java

@@ -278,7 +278,7 @@ private static TranslogStats translogStats(final EngineConfig config, final Segm
                     config.getGlobalCheckpointSupplier(),
                     config.getPrimaryTermSupplier(),
                     seqNo -> {},
-                    config.getPrimaryModeSupplier()
+                    config.getIndexShardConfigSupplier()
                 )
         ) {
             return translog.stats();

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -216,6 +216,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
+import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.LongSupplier;
@@ -337,6 +338,29 @@ Runnable getGlobalCheckpointSyncer() {
 
     private final List<ReferenceManager.RefreshListener> internalRefreshListener = new ArrayList<>();
 
+    /**
+     * An OpenSearch index shard config supplier
+     *
+     * @opensearch.internal
+     */
+    static public class IndexShardConfigSupplier {
+        BooleanSupplier primaryModeSupplier;
+        BooleanSupplier relocatingSupplier;
+
+        public IndexShardConfigSupplier(BooleanSupplier primaryModeSupplier, BooleanSupplier relocatingSupplier) {
+            this.primaryModeSupplier = primaryModeSupplier;
+            this.relocatingSupplier = relocatingSupplier;
+        }
+
+        public BooleanSupplier getPrimaryModeSupplier() {
+            return primaryModeSupplier;
+        }
+
+        public BooleanSupplier getRelocatingSupplier() {
+            return relocatingSupplier;
+        }
+    }
+
     public IndexShard(
         final ShardRouting shardRouting,
         final IndexSettings indexSettings,
@@ -3733,7 +3757,7 @@ private EngineConfig newEngineConfig(LongSupplier globalCheckpointSupplier) thro
             () -> getOperationPrimaryTerm(),
             tombstoneDocSupplier(),
             isReadOnlyReplica,
-            replicationTracker::isPrimaryMode,
+            new IndexShardConfigSupplier(replicationTracker::isPrimaryMode, () -> shardRouting.relocating()),
             translogFactorySupplier.apply(indexSettings, shardRouting),
             isTimeSeriesDescSortOptimizationEnabled() ? DataStream.TIMESERIES_LEAF_SORTER : null // DESC @timestamp default order for
                                                                                                  // timeseries

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

@@ -8,8 +8,9 @@
 
 package org.opensearch.index.translog;
 
+import org.opensearch.index.shard.IndexShard;
+
 import java.io.IOException;
-import java.util.function.BooleanSupplier;
 import java.util.function.LongConsumer;
 import java.util.function.LongSupplier;
 
@@ -28,7 +29,7 @@ public Translog newTranslog(
         LongSupplier globalCheckpointSupplier,
         LongSupplier primaryTermSupplier,
         LongConsumer persistedSequenceNumberConsumer,
-        BooleanSupplier primaryModeSupplier
+        IndexShard.IndexShardConfigSupplier indexShardConfigSupplier
     ) throws IOException {
 
         return new LocalTranslog(

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

@@ -17,12 +17,12 @@
 import org.opensearch.core.index.shard.ShardId;
 import org.opensearch.index.engine.LifecycleAware;
 import org.opensearch.index.seqno.LocalCheckpointTracker;
+import org.opensearch.index.shard.IndexShard;
 import org.opensearch.index.translog.listener.TranslogEventListener;
 
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.BooleanSupplier;
 import java.util.function.LongConsumer;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
@@ -57,7 +57,7 @@ public InternalTranslogManager(
         TranslogEventListener translogEventListener,
         LifecycleAware engineLifeCycleAware,
         TranslogFactory translogFactory,
-        BooleanSupplier primaryModeSupplier
+        IndexShard.IndexShardConfigSupplier indexShardConfigSupplier
     ) throws IOException {
         this.shardId = shardId;
         this.readLock = readLock;
@@ -70,7 +70,7 @@ public InternalTranslogManager(
             if (tracker != null) {
                 tracker.markSeqNoAsPersisted(seqNo);
             }
-        }, translogUUID, translogFactory, primaryModeSupplier);
+        }, translogUUID, translogFactory, indexShardConfigSupplier);
         assert translog.getGeneration() != null;
         this.translog = translog;
         assert pendingTranslogRecovery.get() == false : "translog recovery can't be pending before we set it";
@@ -357,7 +357,7 @@ protected Translog openTranslog(
         LongConsumer persistedSequenceNumberConsumer,
         String translogUUID,
         TranslogFactory translogFactory,
-        BooleanSupplier primaryModeSupplier
+        IndexShard.IndexShardConfigSupplier indexShardConfigSupplier
     ) throws IOException {
         return translogFactory.newTranslog(
             translogConfig,
@@ -366,7 +366,7 @@ protected Translog openTranslog(
             globalCheckpointSupplier,
             primaryTermSupplier,
             persistedSequenceNumberConsumer,
-            primaryModeSupplier
+            indexShardConfigSupplier
         );
     }
 

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

@@ -8,14 +8,14 @@
 
 package org.opensearch.index.translog;
 
+import org.opensearch.index.shard.IndexShard;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.repositories.Repository;
 import org.opensearch.repositories.RepositoryMissingException;
 import org.opensearch.repositories.blobstore.BlobStoreRepository;
 import org.opensearch.threadpool.ThreadPool;
 
 import java.io.IOException;
-import java.util.function.BooleanSupplier;
 import java.util.function.LongConsumer;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
@@ -54,7 +54,7 @@ public Translog newTranslog(
         LongSupplier globalCheckpointSupplier,
         LongSupplier primaryTermSupplier,
         LongConsumer persistedSequenceNumberConsumer,
-        BooleanSupplier primaryModeSupplier
+        IndexShard.IndexShardConfigSupplier indexShardConfigSupplier
     ) throws IOException {
 
         assert repository instanceof BlobStoreRepository : "repository should be instance of BlobStoreRepository";
@@ -68,7 +68,7 @@ public Translog newTranslog(
             persistedSequenceNumberConsumer,
             blobStoreRepository,
             threadPool,
-            primaryModeSupplier
+            indexShardConfigSupplier
         );
     }
 

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

@@ -17,6 +17,7 @@
 import org.opensearch.common.util.io.IOUtils;
 import org.opensearch.core.index.shard.ShardId;
 import org.opensearch.core.util.FileSystemUtils;
+import org.opensearch.index.shard.IndexShard;
 import org.opensearch.index.translog.transfer.BlobStoreTransferService;
 import org.opensearch.index.translog.transfer.FileTransferTracker;
 import org.opensearch.index.translog.transfer.TransferSnapshot;
@@ -55,6 +56,7 @@ public class RemoteFsTranslog extends Translog {
     private final TranslogTransferManager translogTransferManager;
     private final FileTransferTracker fileTransferTracker;
     private final BooleanSupplier primaryModeSupplier;
+    private final BooleanSupplier relocatingSupplier;
     private volatile long maxRemoteTranslogGenerationUploaded;
 
     private volatile long minSeqNoToKeep;
@@ -80,12 +82,13 @@ public RemoteFsTranslog(
         LongConsumer persistedSequenceNumberConsumer,
         BlobStoreRepository blobStoreRepository,
         ThreadPool threadPool,
-        BooleanSupplier primaryModeSupplier
+        IndexShard.IndexShardConfigSupplier indexShardConfigSupplier
     ) throws IOException {
         super(config, translogUUID, deletionPolicy, globalCheckpointSupplier, primaryTermSupplier, persistedSequenceNumberConsumer);
         logger = Loggers.getLogger(getClass(), shardId);
         this.blobStoreRepository = blobStoreRepository;
-        this.primaryModeSupplier = primaryModeSupplier;
+        this.primaryModeSupplier = indexShardConfigSupplier.getPrimaryModeSupplier();
+        this.relocatingSupplier = indexShardConfigSupplier.getRelocatingSupplier();
         fileTransferTracker = new FileTransferTracker(shardId);
         this.translogTransferManager = buildTranslogTransferManager(blobStoreRepository, threadPool, shardId, fileTransferTracker);
         try {
@@ -385,6 +388,10 @@ public void trimUnreferencedReaders() throws IOException {
             return;
         }
 
+        if (relocatingSupplier.getAsBoolean() == true) {
+            return;
+        }
+
         // cleans up remote translog files not referenced in latest uploaded metadata.
         // This enables us to restore translog from the metadata in case of failover or relocation.
         Set<Long> generationsToDelete = new HashSet<>();

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

@@ -8,8 +8,9 @@
 
 package org.opensearch.index.translog;
 
+import org.opensearch.index.shard.IndexShard;
+
 import java.io.IOException;
-import java.util.function.BooleanSupplier;
 import java.util.function.LongConsumer;
 import java.util.function.LongSupplier;
 
@@ -29,6 +30,6 @@ Translog newTranslog(
         final LongSupplier globalCheckpointSupplier,
         final LongSupplier primaryTermSupplier,
         final LongConsumer persistedSequenceNumberConsumer,
-        final BooleanSupplier primaryModeSupplier
+        final IndexShard.IndexShardConfigSupplier indexShardConfigSupplier
     ) throws IOException;
 }

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

@@ -12,10 +12,10 @@
 import org.opensearch.core.index.shard.ShardId;
 import org.opensearch.index.engine.LifecycleAware;
 import org.opensearch.index.seqno.LocalCheckpointTracker;
+import org.opensearch.index.shard.IndexShard;
 import org.opensearch.index.translog.listener.TranslogEventListener;
 
 import java.io.IOException;
-import java.util.function.BooleanSupplier;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
 
@@ -38,7 +38,7 @@ public WriteOnlyTranslogManager(
         TranslogEventListener translogEventListener,
         LifecycleAware engineLifecycleAware,
         TranslogFactory translogFactory,
-        BooleanSupplier primaryModeSupplier
+        IndexShard.IndexShardConfigSupplier indexShardConfigSupplier
     ) throws IOException {
         super(
             translogConfig,
@@ -52,7 +52,7 @@ public WriteOnlyTranslogManager(
             translogEventListener,
             engineLifecycleAware,
             translogFactory,
-            primaryModeSupplier
+            indexShardConfigSupplier
         );
     }