Fixing assertion in TranslogWritter for derived source case in which recovery source could be different in display form but logically would be same

Signed-off-by: Tanik Pansuriya <panbhai@amazon.com>

Author: tanik98

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

@@ -115,6 +115,7 @@
 import org.opensearch.index.translog.TranslogDeletionPolicy;
 import org.opensearch.index.translog.TranslogException;
 import org.opensearch.index.translog.TranslogManager;
+import org.opensearch.index.translog.TranslogOperationHelper;
 import org.opensearch.index.translog.listener.CompositeTranslogEventListener;
 import org.opensearch.index.translog.listener.TranslogEventListener;
 import org.opensearch.search.suggest.completion.CompletionStats;
@@ -373,7 +374,8 @@ protected TranslogManager createTranslogManager(
             translogEventListener,
             this::ensureOpen,
             engineConfig.getTranslogFactory(),
-            engineConfig.getStartedPrimarySupplier()
+            engineConfig.getStartedPrimarySupplier(),
+            TranslogOperationHelper.create(engineConfig)
         );
     }
 

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

@@ -29,6 +29,7 @@
 import org.opensearch.index.translog.TranslogDeletionPolicy;
 import org.opensearch.index.translog.TranslogException;
 import org.opensearch.index.translog.TranslogManager;
+import org.opensearch.index.translog.TranslogOperationHelper;
 import org.opensearch.index.translog.WriteOnlyTranslogManager;
 import org.opensearch.index.translog.listener.TranslogEventListener;
 import org.opensearch.search.suggest.completion.CompletionStats;
@@ -125,7 +126,8 @@ public void onAfterTranslogSync() {
                 },
                 this,
                 engineConfig.getTranslogFactory(),
-                engineConfig.getStartedPrimarySupplier()
+                engineConfig.getStartedPrimarySupplier(),
+                TranslogOperationHelper.create(engineConfig)
             );
             this.translogManager = translogManagerRef;
             success = true;

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

@@ -51,6 +51,7 @@
 import org.opensearch.index.translog.TranslogDeletionPolicy;
 import org.opensearch.index.translog.TranslogException;
 import org.opensearch.index.translog.TranslogManager;
+import org.opensearch.index.translog.TranslogOperationHelper;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -191,7 +192,8 @@ public void trimUnreferencedTranslogFiles() throws TranslogException {
                                         engineConfig.getGlobalCheckpointSupplier(),
                                         engineConfig.getPrimaryTermSupplier(),
                                         seqNo -> {},
-                                        engineConfig.getStartedPrimarySupplier()
+                                        engineConfig.getStartedPrimarySupplier(),
+                                        TranslogOperationHelper.create(engineConfig)
                                     )
                             ) {
                                 translog.trimUnreferencedReaders();

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

@@ -54,6 +54,7 @@
 import org.opensearch.index.translog.TranslogConfig;
 import org.opensearch.index.translog.TranslogDeletionPolicy;
 import org.opensearch.index.translog.TranslogManager;
+import org.opensearch.index.translog.TranslogOperationHelper;
 import org.opensearch.index.translog.TranslogStats;
 import org.opensearch.search.suggest.completion.CompletionStats;
 import org.opensearch.transport.Transports;
@@ -271,12 +272,14 @@ private static TranslogStats translogStats(final EngineConfig config, final Segm
             Translog translog = config.getTranslogFactory()
                 .newTranslog(
                     translogConfig,
+
                     translogUuid,
                     translogDeletionPolicy,
                     config.getGlobalCheckpointSupplier(),
                     config.getPrimaryTermSupplier(),
                     seqNo -> {},
-                    config.getStartedPrimarySupplier()
+                    config.getStartedPrimarySupplier(),
+                    TranslogOperationHelper.create(config)
                 )
         ) {
             return translog.stats();

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

@@ -153,7 +153,7 @@ public final class TranslogLeafReader extends LeafReader {
     );
     public static Set<String> ALL_FIELD_NAMES = Sets.newHashSet(FAKE_SOURCE_FIELD.name, FAKE_ROUTING_FIELD.name, FAKE_ID_FIELD.name);
 
-    TranslogLeafReader(Translog.Index operation, EngineConfig engineConfig) {
+    public TranslogLeafReader(Translog.Index operation, EngineConfig engineConfig) {
         this.operation = operation;
         this.engineConfig = engineConfig;
     }

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

@@ -28,7 +28,8 @@ public Translog newTranslog(
         LongSupplier globalCheckpointSupplier,
         LongSupplier primaryTermSupplier,
         LongConsumer persistedSequenceNumberConsumer,
-        BooleanSupplier startedPrimarySupplier
+        BooleanSupplier startedPrimarySupplier,
+        TranslogOperationHelper translogOperationHelper
     ) throws IOException {
 
         return new LocalTranslog(
@@ -37,7 +38,8 @@ public Translog newTranslog(
             translogDeletionPolicy,
             globalCheckpointSupplier,
             primaryTermSupplier,
-            persistedSequenceNumberConsumer
+            persistedSequenceNumberConsumer,
+            translogOperationHelper
         );
     }
 }

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

@@ -58,7 +58,8 @@ public InternalTranslogManager(
         TranslogEventListener translogEventListener,
         LifecycleAware engineLifeCycleAware,
         TranslogFactory translogFactory,
-        BooleanSupplier startedPrimarySupplier
+        BooleanSupplier startedPrimarySupplier,
+        TranslogOperationHelper translogOperationHelper
     ) throws IOException {
         this.shardId = shardId;
         this.readLock = readLock;
@@ -71,7 +72,7 @@ public InternalTranslogManager(
             if (tracker != null) {
                 tracker.markSeqNoAsPersisted(seqNo);
             }
-        }, translogUUID, translogFactory, startedPrimarySupplier);
+        }, translogUUID, translogFactory, startedPrimarySupplier, translogOperationHelper);
         assert translog.getGeneration() != null;
         this.translog = translog;
         assert pendingTranslogRecovery.get() == false : "translog recovery can't be pending before we set it";
@@ -368,7 +369,8 @@ protected Translog openTranslog(
         LongConsumer persistedSequenceNumberConsumer,
         String translogUUID,
         TranslogFactory translogFactory,
-        BooleanSupplier startedPrimarySupplier
+        BooleanSupplier startedPrimarySupplier,
+        TranslogOperationHelper translogOperationHelper
     ) throws IOException {
         return translogFactory.newTranslog(
             translogConfig,
@@ -377,7 +379,8 @@ protected Translog openTranslog(
             globalCheckpointSupplier,
             primaryTermSupplier,
             persistedSequenceNumberConsumer,
-            startedPrimarySupplier
+            startedPrimarySupplier,
+            translogOperationHelper
         );
     }
 

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

@@ -49,9 +49,18 @@ public LocalTranslog(
         TranslogDeletionPolicy deletionPolicy,
         final LongSupplier globalCheckpointSupplier,
         final LongSupplier primaryTermSupplier,
-        final LongConsumer persistedSequenceNumberConsumer
+        final LongConsumer persistedSequenceNumberConsumer,
+        final TranslogOperationHelper translogOperationHelper
     ) throws IOException {
-        super(config, translogUUID, deletionPolicy, globalCheckpointSupplier, primaryTermSupplier, persistedSequenceNumberConsumer);
+        super(
+            config,
+            translogUUID,
+            deletionPolicy,
+            globalCheckpointSupplier,
+            primaryTermSupplier,
+            persistedSequenceNumberConsumer,
+            translogOperationHelper
+        );
         try {
             final Checkpoint checkpoint = readCheckpoint(location);
             final Path nextTranslogFile = location.resolve(getFilename(checkpoint.generation + 1));

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

@@ -64,7 +64,8 @@ public Translog newTranslog(
         LongSupplier globalCheckpointSupplier,
         LongSupplier primaryTermSupplier,
         LongConsumer persistedSequenceNumberConsumer,
-        BooleanSupplier startedPrimarySupplier
+        BooleanSupplier startedPrimarySupplier,
+        TranslogOperationHelper translogOperationHelper
     ) throws IOException {
 
         assert repository instanceof BlobStoreRepository : "repository should be instance of BlobStoreRepository";
@@ -81,7 +82,8 @@ public Translog newTranslog(
                 threadPool,
                 startedPrimarySupplier,
                 remoteTranslogTransferTracker,
-                remoteStoreSettings
+                remoteStoreSettings,
+                translogOperationHelper
             );
         } else {
             return new RemoteFsTranslog(
@@ -95,7 +97,8 @@ public Translog newTranslog(
                 threadPool,
                 startedPrimarySupplier,
                 remoteTranslogTransferTracker,
-                remoteStoreSettings
+                remoteStoreSettings,
+                translogOperationHelper
             );
         }
     }

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

@@ -75,7 +75,8 @@ public RemoteFsTimestampAwareTranslog(
         ThreadPool threadPool,
         BooleanSupplier startedPrimarySupplier,
         RemoteTranslogTransferTracker remoteTranslogTransferTracker,
-        RemoteStoreSettings remoteStoreSettings
+        RemoteStoreSettings remoteStoreSettings,
+        TranslogOperationHelper translogOperationHelper
     ) throws IOException {
         super(
             config,
@@ -88,7 +89,8 @@ public RemoteFsTimestampAwareTranslog(
             threadPool,
             startedPrimarySupplier,
             remoteTranslogTransferTracker,
-            remoteStoreSettings
+            remoteStoreSettings,
+            translogOperationHelper
         );
         logger = Loggers.getLogger(getClass(), shardId);
         this.metadataFilePinnedTimestampMap = new HashMap<>();