Make sure EngineConfig can access whether shard is primary (#92326)

Make sure engine factories can access the `RecoveryState#primary` flag, so plugins can create different engines for primary and replica indices.

Author: arteam

server/src/internalClusterTest/java/org/elasticsearch/indices/IndexingMemoryControllerIT.java

@@ -79,7 +79,8 @@ EngineConfig engineConfigWithLargerIndexingMemory(EngineConfig config) {
                 config.getSnapshotCommitSupplier(),
                 config.getLeafSorter(),
                 config.getRelativeTimeInNanosSupplier(),
-                config.getIndexCommitListener()
+                config.getIndexCommitListener(),
+                config.isRecoveringAsPrimary()
             );
         }
 

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

@@ -128,6 +128,8 @@ public Supplier<RetentionLeases> retentionLeasesSupplier() {
     @Nullable
     private final Engine.IndexCommitListener indexCommitListener;
 
+    private final boolean recoveringAsPrimary;
+
     /**
      * Creates a new {@link org.elasticsearch.index.engine.EngineConfig}
      */
@@ -156,7 +158,8 @@ public EngineConfig(
         IndexStorePlugin.SnapshotCommitSupplier snapshotCommitSupplier,
         Comparator<LeafReader> leafSorter,
         LongSupplier relativeTimeInNanosSupplier,
-        Engine.IndexCommitListener indexCommitListener
+        Engine.IndexCommitListener indexCommitListener,
+        boolean recoveringAsPrimary
     ) {
         this.shardId = shardId;
         this.indexSettings = indexSettings;
@@ -198,6 +201,7 @@ public EngineConfig(
         this.leafSorter = leafSorter;
         this.relativeTimeInNanosSupplier = relativeTimeInNanosSupplier;
         this.indexCommitListener = indexCommitListener;
+        this.recoveringAsPrimary = recoveringAsPrimary;
     }
 
     /**
@@ -405,4 +409,12 @@ public LongSupplier getRelativeTimeInNanosSupplier() {
     public Engine.IndexCommitListener getIndexCommitListener() {
         return indexCommitListener;
     }
+
+    /**
+     * Represents the primary state only in case when a recovery starts.
+     * IMPORTANT: The flag is a temporary solution and should NOT be used outside of Stateless.
+     */
+    public boolean isRecoveringAsPrimary() {
+        return recoveringAsPrimary;
+    }
 }

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

@@ -3274,7 +3274,8 @@ private EngineConfig newEngineConfig(LongSupplier globalCheckpointSupplier) {
             snapshotCommitSupplier,
             isTimeseriesIndex ? TIMESERIES_LEAF_READERS_SORTER : null,
             relativeTimeInNanosSupplier,
-            indexCommitListener
+            indexCommitListener,
+            recoveryState != null && recoveryState.getPrimary()
         );
     }
 

server/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java

@@ -3590,7 +3590,8 @@ public void testRecoverFromForeignTranslog() throws IOException {
             IndexModule.DEFAULT_SNAPSHOT_COMMIT_SUPPLIER,
             null,
             config.getRelativeTimeInNanosSupplier(),
-            null
+            null,
+            false
         );
         expectThrows(EngineCreationFailureException.class, () -> new InternalEngine(brokenConfig));
 
@@ -7260,7 +7261,8 @@ public void testNotWarmUpSearcherInEngineCtor() throws Exception {
                 config.getSnapshotCommitSupplier(),
                 config.getLeafSorter(),
                 config.getRelativeTimeInNanosSupplier(),
-                config.getIndexCommitListener()
+                config.getIndexCommitListener(),
+                config.isRecoveringAsPrimary()
             );
             try (InternalEngine engine = createEngine(configWithWarmer)) {
                 assertThat(warmedUpReaders, empty());

server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java

@@ -2900,11 +2900,15 @@ public void testRecoverFromTranslog() throws IOException {
     }
 
     public void testShardActiveDuringInternalRecovery() throws IOException {
-        IndexShard shard = newStartedShard(true);
+        boolean isPrimary = randomBoolean();
+        IndexShard shard = newStartedShard(isPrimary);
         indexDoc(shard, "_doc", "0");
         shard = reinitShard(shard);
         DiscoveryNode localNode = new DiscoveryNode("foo", buildNewFakeTransportAddress(), emptyMap(), emptySet(), Version.CURRENT);
-        shard.markAsRecovering("for testing", new RecoveryState(shard.routingEntry(), localNode, null));
+        DiscoveryNode sourceNode = isPrimary
+            ? null
+            : new DiscoveryNode("bar", buildNewFakeTransportAddress(), emptyMap(), emptySet(), Version.CURRENT);
+        shard.markAsRecovering("for testing", new RecoveryState(shard.routingEntry(), localNode, sourceNode));
         // Shard is still inactive since we haven't started recovering yet
         assertFalse(shard.isActive());
         shard.prepareForIndexRecovery();
@@ -2914,6 +2918,8 @@ public void testShardActiveDuringInternalRecovery() throws IOException {
         shard.openEngineAndRecoverFromTranslog();
         // Shard should now be active since we did recover:
         assertTrue(shard.isActive());
+        // Recovery state should be propagated to the engine
+        assertEquals(isPrimary, shard.getEngine().config().isRecoveringAsPrimary());
         closeShards(shard);
     }
 
@@ -4520,7 +4526,8 @@ public void testCloseShardWhileEngineIsWarming() throws Exception {
                 IndexModule.DEFAULT_SNAPSHOT_COMMIT_SUPPLIER,
                 config.getLeafSorter(),
                 config.getRelativeTimeInNanosSupplier(),
-                config.getIndexCommitListener()
+                config.getIndexCommitListener(),
+                config.isRecoveringAsPrimary()
             );
             return new InternalEngine(configWithWarmer);
         });

server/src/test/java/org/elasticsearch/index/shard/RefreshListenersTests.java

@@ -153,7 +153,8 @@ public void onFailedEngine(String reason, @Nullable Exception e) {
             IndexModule.DEFAULT_SNAPSHOT_COMMIT_SUPPLIER,
             null,
             System::nanoTime,
-            null
+            null,
+            false
         );
         engine = new InternalEngine(config);
         engine.recoverFromTranslog((e, s) -> 0, Long.MAX_VALUE);

server/src/test/java/org/elasticsearch/indices/IndexingMemoryControllerTests.java

@@ -402,7 +402,8 @@ EngineConfig configWithRefreshListener(EngineConfig config, ReferenceManager.Ref
             config.getSnapshotCommitSupplier(),
             config.getLeafSorter(),
             config.getRelativeTimeInNanosSupplier(),
-            config.getIndexCommitListener()
+            config.getIndexCommitListener(),
+            config.isRecoveringAsPrimary()
         );
     }
 

test/framework/src/main/java/org/elasticsearch/index/engine/EngineTestCase.java

@@ -275,7 +275,8 @@ public EngineConfig copy(EngineConfig config, LongSupplier globalCheckpointSuppl
             config.getSnapshotCommitSupplier(),
             config.getLeafSorter(),
             config.getRelativeTimeInNanosSupplier(),
-            config.getIndexCommitListener()
+            config.getIndexCommitListener(),
+            config.isRecoveringAsPrimary()
         );
     }
 
@@ -305,7 +306,8 @@ public EngineConfig copy(EngineConfig config, Analyzer analyzer) {
             config.getSnapshotCommitSupplier(),
             config.getLeafSorter(),
             config.getRelativeTimeInNanosSupplier(),
-            config.getIndexCommitListener()
+            config.getIndexCommitListener(),
+            config.isRecoveringAsPrimary()
         );
     }
 
@@ -335,7 +337,8 @@ public EngineConfig copy(EngineConfig config, MergePolicy mergePolicy) {
             config.getSnapshotCommitSupplier(),
             config.getLeafSorter(),
             config.getRelativeTimeInNanosSupplier(),
-            config.getIndexCommitListener()
+            config.getIndexCommitListener(),
+            config.isRecoveringAsPrimary()
         );
     }
 
@@ -858,7 +861,8 @@ public EngineConfig config(
             IndexModule.DEFAULT_SNAPSHOT_COMMIT_SUPPLIER,
             null,
             System::nanoTime,
-            indexCommitListener
+            indexCommitListener,
+            false
         );
     }
 
@@ -896,7 +900,8 @@ protected EngineConfig config(EngineConfig config, Store store, Path translogPat
             config.getSnapshotCommitSupplier(),
             config.getLeafSorter(),
             config.getRelativeTimeInNanosSupplier(),
-            config.getIndexCommitListener()
+            config.getIndexCommitListener(),
+            config.isRecoveringAsPrimary()
         );
     }
 

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/index/engine/FollowingEngineTests.java

@@ -288,7 +288,8 @@ public void onFailedEngine(String reason, Exception e) {
             IndexModule.DEFAULT_SNAPSHOT_COMMIT_SUPPLIER,
             null,
             System::nanoTime,
-            null
+            null,
+            false
         );
     }