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 <[email protected]>

Author: tanik98

CHANGELOG.md

@@ -7,43 +7,6 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Added
 - Add support for Warm Indices Write Block on Flood Watermark breach ([#18375](https://github.com/opensearch-project/OpenSearch/pull/18375))
 - Ability to run Code Coverage with Gradle and produce the jacoco reports locally ([#18509](https://github.com/opensearch-project/OpenSearch/issues/18509))
-- Add support for linux riscv64 platform ([#18156](https://github.com/opensearch-project/OpenSearch/pull/18156))
-- [Rule based auto-tagging] Add get rule API ([#17336](https://github.com/opensearch-project/OpenSearch/pull/17336))
-- [Rule based auto-tagging] Add Delete Rule API ([#18184](https://github.com/opensearch-project/OpenSearch/pull/18184))
-- Add paginated wlm/stats API ([#17638](https://github.com/opensearch-project/OpenSearch/pull/17638))
-- [Rule based auto-tagging] Add Create rule API ([#17792](https://github.com/opensearch-project/OpenSearch/pull/17792))
-- Implement parallel shard refresh behind cluster settings ([#17782](https://github.com/opensearch-project/OpenSearch/pull/17782))
-- Bump OpenSearch Core main branch to 3.0.0 ([#18039](https://github.com/opensearch-project/OpenSearch/pull/18039))
-- [Rule based Auto-tagging] Add wlm `ActionFilter` ([#17791](https://github.com/opensearch-project/OpenSearch/pull/17791))
-- Update API of Message in index to add the timestamp for lag calculation in ingestion polling ([#17977](https://github.com/opensearch-project/OpenSearch/pull/17977/))
-- Add Warm Disk Threshold Allocation Decider for Warm shards ([#18082](https://github.com/opensearch-project/OpenSearch/pull/18082))
-- Add composite directory factory ([#17988](https://github.com/opensearch-project/OpenSearch/pull/17988))
-- [Rule based auto-tagging] Add refresh based synchronization service for `Rule`s ([#18128](https://github.com/opensearch-project/OpenSearch/pull/18128))
-- Add pull-based ingestion error metrics and make internal queue size configurable ([#18088](https://github.com/opensearch-project/OpenSearch/pull/18088))
-- [Derive Source] Adding support for derive source feature and implementing it for various type of field mappers ([#17759](https://github.com/opensearch-project/OpenSearch/pull/17759))
-- [Derive Source] Adding integration of derived source feature across diff paths ([#18054](https://github.com/opensearch-project/OpenSearch/pull/18054))
-- [Security Manager Replacement] Enhance Java Agent to intercept newByteChannel ([#17989](https://github.com/opensearch-project/OpenSearch/pull/17989))
-- Enabled Async Shard Batch Fetch by default ([#18139](https://github.com/opensearch-project/OpenSearch/pull/18139))
-- Allow to get the search request from the QueryCoordinatorContext ([#17818](https://github.com/opensearch-project/OpenSearch/pull/17818))
-- Reject close index requests, while remote store migration is in progress.([#18327](https://github.com/opensearch-project/OpenSearch/pull/18327))
-- Improve sort-query performance by retaining the default `totalHitsThreshold` for approximated `match_all` queries ([#18189](https://github.com/opensearch-project/OpenSearch/pull/18189))
-- Enable testing for ExtensiblePlugins using classpath plugins ([#16908](https://github.com/opensearch-project/OpenSearch/pull/16908))
-- Introduce system generated ingest pipeline ([#17817](https://github.com/opensearch-project/OpenSearch/pull/17817)))
-- Apply cluster state metadata and routing table diff when building cluster state from remote([#18256](https://github.com/opensearch-project/OpenSearch/pull/18256))
-- Support create mode in pull-based ingestion and add retries for transient failures ([#18250](https://github.com/opensearch-project/OpenSearch/pull/18250)))
-- Decouple the init of Crypto Plugin and KeyProvider in CryptoRegistry ([18270](https://github.com/opensearch-project/OpenSearch/pull18270)))
-- Support cluster write block in pull-based ingestion ([#18280](https://github.com/opensearch-project/OpenSearch/pull/18280)))
-- Use QueryCoordinatorContext for the rewrite in validate API. ([#18272](https://github.com/opensearch-project/OpenSearch/pull/18272))
-- Upgrade crypto kms plugin dependencies for AWS SDK v2.x. ([#18268](https://github.com/opensearch-project/OpenSearch/pull/18268))
-- Add support for `matched_fields` with the unified highlighter ([#18164](https://github.com/opensearch-project/OpenSearch/issues/18164))
-- Add BooleanQuery rewrite for must_not RangeQuery clauses ([#17655](https://github.com/opensearch-project/OpenSearch/pull/17655))
-- [repository-s3] Add support for SSE-KMS and S3 bucket owner verification ([#18312](https://github.com/opensearch-project/OpenSearch/pull/18312))
-- Optimize gRPC perf by passing by reference ([#18303](https://github.com/opensearch-project/OpenSearch/pull/18303))
-- Added File Cache Stats - Involves Block level as well as full file level stats ([#17538](https://github.com/opensearch-project/OpenSearch/issues/17479))
-- Added time_in_execution attribute to /_cluster/pending_tasks response ([#17780](https://github.com/opensearch-project/OpenSearch/pull/17780))
-- Added File Cache Pinning ([#17617](https://github.com/opensearch-project/OpenSearch/issues/13648))
-- Support consumer reset in Resume API for pull-based ingestion. This PR includes a breaking change for the experimental pull-based ingestion feature. ([#18332](https://github.com/opensearch-project/OpenSearch/pull/18332))
-- Add FIPS build tooling ([#4254](https://github.com/opensearch-project/security/issues/4254))
 
 ### Changed
 
@@ -60,3 +23,6 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Security
 
 [Unreleased 3.x]: https://github.com/opensearch-project/OpenSearch/compare/3.1...main
+### Added
+[Derive Source] Adding back "Adding integration of derived source feature across diff paths" with fixes for IT #18565
+([#18565](https://github.com/opensearch-project/OpenSearch/pull/18565))

server/src/internalClusterTest/java/org/opensearch/recovery/RecoveryWhileUnderLoadIT.java

@@ -533,7 +533,11 @@ public void testRecoveryWithDerivedSourceEnabled() throws Exception {
 
         final int totalNumDocs = scaledRandomIntBetween(200, 1000);
         for (int i = 0; i < totalNumDocs; i++) {
-            client().prepareIndex("test").setId(String.valueOf(i)).setSource("name", "test" + i, "age", i).get();
+            if (i % 2 == 0) {
+                client().prepareIndex("test").setId(String.valueOf(i)).setSource("name", "test" + i, "age", i).get();
+            } else {
+                client().prepareIndex("test").setId(String.valueOf(i)).setSource("age", i, "name", "test" + i).get();
+            }
 
             if (i % 100 == 0) {
                 // Occasionally flush to create new segments
@@ -592,10 +596,17 @@ public void testReplicaRecoveryWithDerivedSourceBeforeRefresh() throws Exception
         // Index documents without refresh
         int docCount = randomIntBetween(100, 200);
         for (int i = 0; i < docCount; i++) {
-            client().prepareIndex("test")
-                .setId(String.valueOf(i))
-                .setSource("timestamp", "2023-01-01T01:20:30." + String.valueOf(i % 10).repeat(3) + "Z", "ip", "192.168.1." + i)
-                .get();
+            if (i % 2 == 0) {
+                client().prepareIndex("test")
+                    .setId(String.valueOf(i))
+                    .setSource("timestamp", "2023-01-01T01:20:30." + String.valueOf(i % 10).repeat(3) + "Z", "ip", "192.168.1." + i)
+                    .get();
+            } else {
+                client().prepareIndex("test")
+                    .setId(String.valueOf(i))
+                    .setSource("ip", "192.168.1." + i, "timestamp", "2023-01-01T01:20:30." + String.valueOf(i % 10).repeat(3) + "Z")
+                    .get();
+            }
         }
 
         // Add replica before refresh
@@ -657,11 +668,19 @@ public void testReplicaRecoveryWithDerivedSourceFromTranslog() throws Exception
         // Index documents with immediate visibility
         int docCount = randomIntBetween(100, 200);
         for (int i = 0; i < docCount; i++) {
-            client().prepareIndex("test")
-                .setId(String.valueOf(i))
-                .setSource("coordinates", Geohash.stringEncode(40.0 + i, 75.0 + i) + i, "value", "fox_" + i + " in the field")
-                .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
-                .get();
+            if (i % 2 == 0) {
+                client().prepareIndex("test")
+                    .setId(String.valueOf(i))
+                    .setSource("coordinates", Geohash.stringEncode(40.0 + i, 75.0 + i) + i, "value", "fox_" + i + " in the field")
+                    .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
+                    .get();
+            } else {
+                client().prepareIndex("test")
+                    .setId(String.valueOf(i))
+                    .setSource("value", "fox_" + i + " in the field", "coordinates", Geohash.stringEncode(40.0 + i, 75.0 + i) + i)
+                    .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
+                    .get();
+            }
         }
 
         // Force flush to ensure documents are in segments

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

@@ -29,7 +29,6 @@
 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;
@@ -126,8 +125,7 @@ public void onAfterTranslogSync() {
                 },
                 this,
                 engineConfig.getTranslogFactory(),
-                engineConfig.getStartedPrimarySupplier(),
-                TranslogOperationHelper.create(engineConfig)
+                engineConfig.getStartedPrimarySupplier()
             );
             this.translogManager = translogManagerRef;
             success = true;

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

@@ -20,6 +20,31 @@
  */
 public class InternalTranslogFactory implements TranslogFactory {
 
+    @Override
+    public Translog newTranslog(
+        TranslogConfig translogConfig,
+        String translogUUID,
+        TranslogDeletionPolicy translogDeletionPolicy,
+        LongSupplier globalCheckpointSupplier,
+        LongSupplier primaryTermSupplier,
+        LongConsumer persistedSequenceNumberConsumer,
+        BooleanSupplier startedPrimarySupplier
+    ) throws IOException {
+
+        assert translogConfig.getIndexSettings().isDerivedSourceEnabled() == false; // For derived source supported index, primary method
+                                                                                    // must be used
+
+        return new LocalTranslog(
+            translogConfig,
+            translogUUID,
+            translogDeletionPolicy,
+            globalCheckpointSupplier,
+            primaryTermSupplier,
+            persistedSequenceNumberConsumer,
+            TranslogOperationHelper.EMPTY
+        );
+    }
+
     @Override
     public Translog newTranslog(
         TranslogConfig translogConfig,

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

@@ -56,6 +56,30 @@ public RemoteBlobStoreInternalTranslogFactory(
         this.remoteStoreSettings = remoteStoreSettings;
     }
 
+    @Override
+    public Translog newTranslog(
+        TranslogConfig config,
+        String translogUUID,
+        TranslogDeletionPolicy deletionPolicy,
+        LongSupplier globalCheckpointSupplier,
+        LongSupplier primaryTermSupplier,
+        LongConsumer persistedSequenceNumberConsumer,
+        BooleanSupplier startedPrimarySupplier
+    ) throws IOException {
+        assert config.getIndexSettings().isDerivedSourceEnabled() == false; // For derived source supported index, primary method must be
+                                                                            // used
+        return this.newTranslog(
+            config,
+            translogUUID,
+            deletionPolicy,
+            globalCheckpointSupplier,
+            primaryTermSupplier,
+            persistedSequenceNumberConsumer,
+            startedPrimarySupplier,
+            TranslogOperationHelper.EMPTY
+        );
+    }
+
     @Override
     public Translog newTranslog(
         TranslogConfig config,

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

@@ -162,17 +162,18 @@ public abstract class Translog extends AbstractIndexShardComponent implements In
      * generation referenced from already committed data. This means all operations that have not yet been committed should be in the
      * translog file referenced by this generation. The translog creation will fail if this generation can't be opened.
      *
-     * @param config                   the configuration of this translog
-     * @param translogUUID             the translog uuid to open, null for a new translog
-     * @param deletionPolicy           an instance of {@link TranslogDeletionPolicy} that controls when a translog file can be safely
-     *                                 deleted
-     * @param globalCheckpointSupplier a supplier for the global checkpoint
-     * @param primaryTermSupplier      a supplier for the latest value of primary term of the owning index shard. The latest term value is
-     *                                 examined and stored in the header whenever a new generation is rolled. It's guaranteed from outside
-     *                                 that a new generation is rolled when the term is increased. This guarantee allows to us to validate
-     *                                 and reject operation whose term is higher than the primary term stored in the translog header.
+     * @param config                          the configuration of this translog
+     * @param translogUUID                    the translog uuid to open, null for a new translog
+     * @param deletionPolicy                  an instance of {@link TranslogDeletionPolicy} that controls when a translog file can be safely
+     *                                        deleted
+     * @param globalCheckpointSupplier        a supplier for the global checkpoint
+     * @param primaryTermSupplier             a supplier for the latest value of primary term of the owning index shard. The latest term value is
+     *                                        examined and stored in the header whenever a new generation is rolled. It's guaranteed from outside
+     *                                        that a new generation is rolled when the term is increased. This guarantee allows to us to validate
+     *                                        and reject operation whose term is higher than the primary term stored in the translog header.
      * @param persistedSequenceNumberConsumer a callback that's called whenever an operation with a given sequence number is successfully
      *                                        persisted.
+     * @param translogOperationHelper         a helper method to validate translog operations with the support of derived source
      */
     public Translog(
         final TranslogConfig config,
@@ -199,6 +200,30 @@ public Translog(
         this.translogOperationHelper = translogOperationHelper;
     }
 
+    /**
+     * Secondary constructor, this should only be called if index is normal and not for derived source
+     */
+    public Translog(
+        final TranslogConfig config,
+        final String translogUUID,
+        TranslogDeletionPolicy deletionPolicy,
+        final LongSupplier globalCheckpointSupplier,
+        final LongSupplier primaryTermSupplier,
+        final LongConsumer persistedSequenceNumberConsumer
+    ) throws IOException {
+        this(
+            config,
+            translogUUID,
+            deletionPolicy,
+            globalCheckpointSupplier,
+            primaryTermSupplier,
+            persistedSequenceNumberConsumer,
+            TranslogOperationHelper.EMPTY
+        );
+        assert config.getIndexSettings().isDerivedSourceEnabled() == false; // For derived source supported index, it is incorrect to use
+                                                                            // this constructor
+    }
+
     /** recover all translog files found on disk */
     protected ArrayList<TranslogReader> recoverFromFiles(Checkpoint checkpoint) throws IOException {
         boolean success = false;
@@ -2102,7 +2127,7 @@ public static String createEmptyTranslog(
             },
             BigArrays.NON_RECYCLING_INSTANCE,
             null,
-            null
+            TranslogOperationHelper.EMPTY
         );
         writer.close();
         return uuid;

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

@@ -21,10 +21,19 @@
  *
  * @opensearch.api
  */
-@FunctionalInterface
 @PublicApi(since = "1.0.0")
 public interface TranslogFactory {
 
+    Translog newTranslog(
+        final TranslogConfig config,
+        final String translogUUID,
+        final TranslogDeletionPolicy deletionPolicy,
+        final LongSupplier globalCheckpointSupplier,
+        final LongSupplier primaryTermSupplier,
+        final LongConsumer persistedSequenceNumberConsumer,
+        final BooleanSupplier startedPrimarySupplier
+    ) throws IOException;
+
     Translog newTranslog(
         final TranslogConfig config,
         final String translogUUID,

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

@@ -8,8 +8,6 @@
 
 package org.opensearch.index.translog;
 
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.common.collect.Tuple;
 import org.opensearch.common.xcontent.XContentHelper;
@@ -30,12 +28,11 @@
  *
  * @opensearch.internal
  */
-@PublicApi(since = "1.0.0")
+@PublicApi(since = "3.1.0")
 public abstract class TranslogOperationHelper {
 
     public static final TranslogOperationHelper EMPTY = new TranslogOperationHelper() {
     };
-    private static final Logger logger = LogManager.getLogger(TranslogOperationHelper.class);
 
     private TranslogOperationHelper() {}
 
@@ -70,7 +67,6 @@ public boolean hasSameIndexOperation(Translog.Index op1, Translog.Index op2) {
     }
 
     private static BytesReference deriveSource(Translog.Index op, EngineConfig engineConfig) {
-        logger.warn("Fetching derived source for {}", op.id());
         try (TranslogLeafReader leafReader = new TranslogLeafReader(op, engineConfig)) {
             final FieldsVisitor visitor = new FieldsVisitor(true);
             leafReader.storedFields().document(0, visitor); // As reader will have only a single document, segment level doc id will be 0

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

@@ -168,6 +168,43 @@ private TranslogWriter(
         this.translogOperationHelper = translogOperationHelper;
     }
 
+    public static TranslogWriter create(
+        ShardId shardId,
+        String translogUUID,
+        long fileGeneration,
+        Path file,
+        ChannelFactory channelFactory,
+        ByteSizeValue bufferSize,
+        final long initialMinTranslogGen,
+        long initialGlobalCheckpoint,
+        final LongSupplier globalCheckpointSupplier,
+        final LongSupplier minTranslogGenerationSupplier,
+        final long primaryTerm,
+        TragicExceptionHolder tragedy,
+        final LongConsumer persistedSequenceNumberConsumer,
+        final BigArrays bigArrays,
+        Boolean remoteTranslogEnabled
+    ) throws IOException {
+        return create(
+            shardId,
+            translogUUID,
+            fileGeneration,
+            file,
+            channelFactory,
+            bufferSize,
+            initialMinTranslogGen,
+            initialGlobalCheckpoint,
+            globalCheckpointSupplier,
+            minTranslogGenerationSupplier,
+            primaryTerm,
+            tragedy,
+            persistedSequenceNumberConsumer,
+            bigArrays,
+            remoteTranslogEnabled,
+            TranslogOperationHelper.EMPTY
+        );
+    }
+
     public static TranslogWriter create(
         ShardId shardId,
         String translogUUID,