No translog for search-only shards (#93082)

In due course I expect we will address this more fully, but for now we
want to avoid needing to associate a new translog with a search shard
during startup. This breaks the finalization phase of peer recovery
because we use the translog checkpoint to hold the GCP, so with this
commit we just keep hold of the GCP in memory on search shards.

Author: DaveCTurner

server/src/internalClusterTest/java/org/elasticsearch/cluster/routing/ShardRoutingRoleIT.java

@@ -25,7 +25,12 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.engine.EngineFactory;
+import org.elasticsearch.index.engine.NoOpEngine;
+import org.elasticsearch.index.translog.TranslogStats;
 import org.elasticsearch.plugins.ClusterPlugin;
+import org.elasticsearch.plugins.EnginePlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.PluginsService;
 import org.elasticsearch.snapshots.SnapshotState;
@@ -37,6 +42,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.hamcrest.Matchers.anEmptyMap;
@@ -50,7 +56,7 @@ public class ShardRoutingRoleIT extends ESIntegTestCase {
 
     private static final Logger logger = LogManager.getLogger(ShardRoutingRoleIT.class);
 
-    public static class TestPlugin extends Plugin implements ClusterPlugin {
+    public static class TestPlugin extends Plugin implements ClusterPlugin, EnginePlugin {
 
         volatile int numIndexingCopies = 1;
 
@@ -83,6 +89,17 @@ public Decision canForceAllocatePrimary(ShardRouting shardRouting, RoutingNode n
                 }
             });
         }
+
+        @Override
+        public Optional<EngineFactory> getEngineFactory(IndexSettings indexSettings) {
+            // TODO revert this once replication/recovery ignores unpromotable shards
+            return Optional.of(config -> new NoOpEngine(config, new TranslogStats(0, 0, 0, 0, 0)));
+        }
+    }
+
+    @Override
+    protected boolean addMockInternalEngine() {
+        return false;
     }
 
     @Override
@@ -356,7 +373,7 @@ public AllocationCommand getCancelPrimaryCommand() {
         return null;
     }
 
-    public void testSearchRouting() throws InterruptedException {
+    public void testSearchRouting() {
 
         var routingTableWatcher = new RoutingTableWatcher();
         routingTableWatcher.numReplicas = Math.max(1, routingTableWatcher.numReplicas);
@@ -371,7 +388,7 @@ public void testSearchRouting() throws InterruptedException {
             masterClusterService.addListener(routingTableWatcher);
 
             createIndex(INDEX_NAME, routingTableWatcher.getIndexSettings());
-            indexRandom(true, INDEX_NAME, between(1, 100));
+            // TODO index some documents here once recovery/replication ignore unpromotable shards
             ensureGreen(INDEX_NAME);
 
             final var searchShardProfileKeys = new HashSet<String>();

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

@@ -17,12 +17,14 @@
 import org.apache.lucene.store.Directory;
 import org.elasticsearch.common.lucene.Lucene;
 import org.elasticsearch.common.util.concurrent.ReleasableLock;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.shard.DocsStats;
 import org.elasticsearch.index.store.Store;
 import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.index.translog.TranslogConfig;
 import org.elasticsearch.index.translog.TranslogDeletionPolicy;
+import org.elasticsearch.index.translog.TranslogStats;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -43,7 +45,11 @@ public final class NoOpEngine extends ReadOnlyEngine {
     private final DocsStats docsStats;
 
     public NoOpEngine(EngineConfig config) {
-        super(config, null, null, true, Function.identity(), true, true);
+        this(config, null);
+    }
+
+    public NoOpEngine(EngineConfig config, @Nullable TranslogStats translogStats) {
+        super(config, null, translogStats, true, Function.identity(), true, true);
         this.segmentsStats = new SegmentsStats();
         Directory directory = store.directory();
         try (DirectoryReader reader = openDirectory(directory)) {

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

@@ -291,6 +291,9 @@ Runnable getGlobalCheckpointSyncer() {
     private volatile long startedRelativeTimeInNanos;
     private volatile long indexingTimeBeforeShardStartedInNanos;
 
+    // the translog keeps track of the GCP, but unpromotable shards have no translog so we need to track the GCP here instead
+    private volatile long globalCheckPointIfUnpromotable;
+
     public IndexShard(
         final ShardRouting shardRouting,
         final IndexSettings indexSettings,
@@ -1587,6 +1590,11 @@ static Engine.Searcher wrapSearcher(
         ); // completes stats recording
     }
 
+    public void setGlobalCheckpointIfUnpromotable(long globalCheckpoint) {
+        assert shardRouting.isPromotableToPrimary() == false : "must only call this on unpromotable shards";
+        globalCheckPointIfUnpromotable = globalCheckpoint;
+    }
+
     private static final class NonClosingReaderWrapper extends FilterDirectoryReader {
 
         private NonClosingReaderWrapper(DirectoryReader in) throws IOException {
@@ -1885,12 +1893,16 @@ int runTranslogRecovery(Engine engine, Translog.Snapshot snapshot, Engine.Operat
     }
 
     private void loadGlobalCheckpointToReplicationTracker() throws IOException {
-        // we have to set it before we open an engine and recover from the translog because
-        // acquiring a snapshot from the translog causes a sync which causes the global checkpoint to be pulled in,
-        // and an engine can be forced to close in ctor which also causes the global checkpoint to be pulled in.
-        final String translogUUID = store.readLastCommittedSegmentsInfo().getUserData().get(Translog.TRANSLOG_UUID_KEY);
-        final long globalCheckpoint = Translog.readGlobalCheckpoint(translogConfig.getTranslogPath(), translogUUID);
-        replicationTracker.updateGlobalCheckpointOnReplica(globalCheckpoint, "read from translog checkpoint");
+        if (shardRouting.isPromotableToPrimary()) {
+            // we have to set it before we open an engine and recover from the translog because
+            // acquiring a snapshot from the translog causes a sync which causes the global checkpoint to be pulled in,
+            // and an engine can be forced to close in ctor which also causes the global checkpoint to be pulled in.
+            final String translogUUID = store.readLastCommittedSegmentsInfo().getUserData().get(Translog.TRANSLOG_UUID_KEY);
+            final long globalCheckpoint = Translog.readGlobalCheckpoint(translogConfig.getTranslogPath(), translogUUID);
+            replicationTracker.updateGlobalCheckpointOnReplica(globalCheckpoint, "read from translog checkpoint");
+        } else {
+            replicationTracker.updateGlobalCheckpointOnReplica(globalCheckPointIfUnpromotable, "from CleanFilesRequest");
+        }
     }
 
     /**

server/src/main/java/org/elasticsearch/indices/recovery/PeerRecoveryTargetService.java

@@ -313,14 +313,20 @@ public static StartRecoveryRequest getStartRecoveryRequest(
 
         Store.MetadataSnapshot metadataSnapshot;
         try {
-            metadataSnapshot = recoveryTarget.indexShard().snapshotStoreMetadata();
-            // Make sure that the current translog is consistent with the Lucene index; otherwise, we have to throw away the Lucene index.
-            try {
-                final String expectedTranslogUUID = metadataSnapshot.commitUserData().get(Translog.TRANSLOG_UUID_KEY);
-                final long globalCheckpoint = Translog.readGlobalCheckpoint(recoveryTarget.translogLocation(), expectedTranslogUUID);
-                assert globalCheckpoint + 1 >= startingSeqNo : "invalid startingSeqNo " + startingSeqNo + " >= " + globalCheckpoint;
-            } catch (IOException | TranslogCorruptedException e) {
-                logGlobalCheckpointWarning(logger, startingSeqNo, e);
+            if (recoveryTarget.indexShard().routingEntry().isPromotableToPrimary()) {
+                metadataSnapshot = recoveryTarget.indexShard().snapshotStoreMetadata();
+                // Make sure that the current translog is consistent with the Lucene index; otherwise, we have to throw away the Lucene
+                // index.
+                try {
+                    final String expectedTranslogUUID = metadataSnapshot.commitUserData().get(Translog.TRANSLOG_UUID_KEY);
+                    final long globalCheckpoint = Translog.readGlobalCheckpoint(recoveryTarget.translogLocation(), expectedTranslogUUID);
+                    assert globalCheckpoint + 1 >= startingSeqNo : "invalid startingSeqNo " + startingSeqNo + " >= " + globalCheckpoint;
+                } catch (IOException | TranslogCorruptedException e) {
+                    logGlobalCheckpointWarning(logger, startingSeqNo, e);
+                    metadataSnapshot = Store.MetadataSnapshot.EMPTY;
+                    startingSeqNo = UNASSIGNED_SEQ_NO;
+                }
+            } else {
                 metadataSnapshot = Store.MetadataSnapshot.EMPTY;
                 startingSeqNo = UNASSIGNED_SEQ_NO;
             }

server/src/main/java/org/elasticsearch/indices/recovery/RecoveryTarget.java

@@ -497,15 +497,16 @@ public void cleanFiles(
             try {
                 if (indexShard.routingEntry().isPromotableToPrimary()) {
                     store.cleanupAndVerify("recovery CleanFilesRequestHandler", sourceMetadata);
+                    final String translogUUID = Translog.createEmptyTranslog(
+                        indexShard.shardPath().resolveTranslog(),
+                        globalCheckpoint,
+                        shardId,
+                        indexShard.getPendingPrimaryTerm()
+                    );
+                    store.associateIndexWithNewTranslog(translogUUID);
+                } else {
+                    indexShard.setGlobalCheckpointIfUnpromotable(globalCheckpoint);
                 }
-                final String translogUUID = Translog.createEmptyTranslog(
-                    indexShard.shardPath().resolveTranslog(),
-                    globalCheckpoint,
-                    shardId,
-                    indexShard.getPendingPrimaryTerm()
-                );
-                store.associateIndexWithNewTranslog(translogUUID);
-
                 if (indexShard.getRetentionLeases().leases().isEmpty()) {
                     // if empty, may be a fresh IndexShard, so write an empty leases file to disk
                     indexShard.persistRetentionLeases();