Fast refresh indices should use search shards (#113478) (#114259)

Fast refresh indices should now behave like non fast refresh indices in
how they execute (m)gets and searches. I.e., they should use the search
shards.

For BWC, we define a new transport version. We expect search shards to
be upgraded first, before promotable shards. Until the cluster is fully
upgraded, the promotable shards (whether upgraded or not) will still
receive and execute gets/searches locally.

Relates ES-9573
Relates ES-9579

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -235,6 +235,7 @@ public class TransportVersions {
     public static final TransportVersion SEARCH_FAILURE_STATS = def(8_759_00_0);
     public static final TransportVersion INGEST_GEO_DATABASE_PROVIDERS = def(8_760_00_0);
     public static final TransportVersion DATE_TIME_DOC_VALUES_LOCALES = def(8_761_00_0);
+    public static final TransportVersion FAST_REFRESH_RCO = def(8_762_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportShardRefreshAction.java

@@ -23,7 +23,6 @@ import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.injection.guice.Inject;
@@ -120,27 +119,18 @@ public class TransportShardRefreshAction extends TransportReplicationAction<
             ActionListener<Void> listener
         ) {
             assert replicaRequest.primaryRefreshResult.refreshed() : "primary has not refreshed";
-            boolean fastRefresh = IndexSettings.INDEX_FAST_REFRESH_SETTING.get(
-                clusterService.state().metadata().index(indexShardRoutingTable.shardId().getIndex()).getSettings()
+            UnpromotableShardRefreshRequest unpromotableReplicaRequest = new UnpromotableShardRefreshRequest(
+                indexShardRoutingTable,
+                replicaRequest.primaryRefreshResult.primaryTerm(),
+                replicaRequest.primaryRefreshResult.generation(),
+                false
+            );
+            transportService.sendRequest(
+                transportService.getLocalNode(),
+                TransportUnpromotableShardRefreshAction.NAME,
+                unpromotableReplicaRequest,
+                new ActionListenerResponseHandler<>(listener.safeMap(r -> null), in -> ActionResponse.Empty.INSTANCE, refreshExecutor)
             );
-
-            // Indices marked with fast refresh do not rely on refreshing the unpromotables
-            if (fastRefresh) {
-                listener.onResponse(null);
-            } else {
-                UnpromotableShardRefreshRequest unpromotableReplicaRequest = new UnpromotableShardRefreshRequest(
-                    indexShardRoutingTable,
-                    replicaRequest.primaryRefreshResult.primaryTerm(),
-                    replicaRequest.primaryRefreshResult.generation(),
-                    false
-                );
-                transportService.sendRequest(
-                    transportService.getLocalNode(),
-                    TransportUnpromotableShardRefreshAction.NAME,
-                    unpromotableReplicaRequest,
-                    new ActionListenerResponseHandler<>(listener.safeMap(r -> null), in -> ActionResponse.Empty.INSTANCE, refreshExecutor)
-                );
-            }
         }
     }
 }

server/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportUnpromotableShardRefreshAction.java

@@ -24,6 +24,9 @@ import org.elasticsearch.transport.TransportService;
 
 import java.util.List;
 
+import static org.elasticsearch.TransportVersions.FAST_REFRESH_RCO;
+import static org.elasticsearch.index.IndexSettings.INDEX_FAST_REFRESH_SETTING;
+
 public class TransportUnpromotableShardRefreshAction extends TransportBroadcastUnpromotableAction<
     UnpromotableShardRefreshRequest,
     ActionResponse.Empty> {
@@ -73,6 +76,18 @@ public class TransportUnpromotableShardRefreshAction extends TransportBroadcastU
             return;
         }
 
+        // During an upgrade to FAST_REFRESH_RCO, we expect search shards to be first upgraded before the primary is upgraded. Thus,
+        // when the primary is upgraded, and starts to deliver unpromotable refreshes, we expect the search shards to be upgraded already.
+        // Note that the fast refresh setting is final.
+        // TODO: remove assertion (ES-9563)
+        assert INDEX_FAST_REFRESH_SETTING.get(shard.indexSettings().getSettings()) == false
+            || transportService.getLocalNodeConnection().getTransportVersion().onOrAfter(FAST_REFRESH_RCO)
+            : "attempted to refresh a fast refresh search shard "
+                + shard
+                + " on transport version "
+                + transportService.getLocalNodeConnection().getTransportVersion()
+                + " (before FAST_REFRESH_RCO)";
+
         ActionListener.run(responseListener, listener -> {
             shard.waitForPrimaryTermAndGeneration(
                 request.getPrimaryTerm(),

server/src/main/java/org/elasticsearch/action/get/TransportGetAction.java

@@ -125,11 +125,10 @@ public class TransportGetAction extends TransportSingleShardAction<GetRequest, G
         IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
         IndexShard indexShard = indexService.getShard(shardId.id());
         if (indexShard.routingEntry().isPromotableToPrimary() == false) {
-            assert indexShard.indexSettings().isFastRefresh() == false
-                : "a search shard should not receive a TransportGetAction for an index with fast refresh";
             handleGetOnUnpromotableShard(request, indexShard, listener);
             return;
         }
+        // TODO: adapt assertion to assert only that it is not stateless (ES-9563)
         assert DiscoveryNode.isStateless(clusterService.getSettings()) == false || indexShard.indexSettings().isFastRefresh()
             : "in Stateless a promotable to primary shard can receive a TransportGetAction only if an index has the fast refresh setting";
         if (request.realtime()) { // we are not tied to a refresh cycle here anyway

server/src/main/java/org/elasticsearch/action/get/TransportShardMultiGetAction.java

@@ -124,11 +124,10 @@ public class TransportShardMultiGetAction extends TransportSingleShardAction<Mul
         IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
         IndexShard indexShard = indexService.getShard(shardId.id());
         if (indexShard.routingEntry().isPromotableToPrimary() == false) {
-            assert indexShard.indexSettings().isFastRefresh() == false
-                : "a search shard should not receive a TransportShardMultiGetAction for an index with fast refresh";
             handleMultiGetOnUnpromotableShard(request, indexShard, listener);
             return;
         }
+        // TODO: adapt assertion to assert only that it is not stateless (ES-9563)
         assert DiscoveryNode.isStateless(clusterService.getSettings()) == false || indexShard.indexSettings().isFastRefresh()
             : "in Stateless a promotable to primary shard can receive a TransportShardMultiGetAction only if an index has "
                 + "the fast refresh setting";

server/src/main/java/org/elasticsearch/action/support/replication/PostWriteRefresh.java

@@ -19,7 +19,6 @@ import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.translog.Translog;
@@ -53,9 +52,7 @@ public class PostWriteRefresh {
             case WAIT_UNTIL -> waitUntil(indexShard, location, new ActionListener<>() {
                 @Override
                 public void onResponse(Boolean forced) {
-                    // Fast refresh indices do not depend on the unpromotables being refreshed
-                    boolean fastRefresh = IndexSettings.INDEX_FAST_REFRESH_SETTING.get(indexShard.indexSettings().getSettings());
-                    if (location != null && (indexShard.routingEntry().isSearchable() == false && fastRefresh == false)) {
+                    if (location != null && indexShard.routingEntry().isSearchable() == false) {
                         refreshUnpromotables(indexShard, location, listener, forced, postWriteRefreshTimeout);
                     } else {
                         listener.onResponse(forced);
@@ -68,9 +65,7 @@ public class PostWriteRefresh {
                 }
             });
             case IMMEDIATE -> immediate(indexShard, listener.delegateFailureAndWrap((l, r) -> {
-                // Fast refresh indices do not depend on the unpromotables being refreshed
-                boolean fastRefresh = IndexSettings.INDEX_FAST_REFRESH_SETTING.get(indexShard.indexSettings().getSettings());
-                if (indexShard.getReplicationGroup().getRoutingTable().unpromotableShards().size() > 0 && fastRefresh == false) {
+                if (indexShard.getReplicationGroup().getRoutingTable().unpromotableShards().size() > 0) {
                     sendUnpromotableRequests(indexShard, r.generation(), true, l, postWriteRefreshTimeout);
                 } else {
                     l.onResponse(true);

server/src/main/java/org/elasticsearch/cluster/routing/OperationRouting.java

@@ -32,6 +32,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import static org.elasticsearch.TransportVersions.FAST_REFRESH_RCO;
 import static org.elasticsearch.index.IndexSettings.INDEX_FAST_REFRESH_SETTING;
 
 public class OperationRouting {
@@ -305,8 +306,14 @@ public class OperationRouting {
     }
 
     public static boolean canSearchShard(ShardRouting shardRouting, ClusterState clusterState) {
+        // TODO: remove if and always return isSearchable (ES-9563)
         if (INDEX_FAST_REFRESH_SETTING.get(clusterState.metadata().index(shardRouting.index()).getSettings())) {
-            return shardRouting.isPromotableToPrimary();
+            // Until all the cluster is upgraded, we send searches/gets to the primary (even if it has been upgraded) to execute locally.
+            if (clusterState.getMinTransportVersion().onOrAfter(FAST_REFRESH_RCO)) {
+                return shardRouting.isSearchable();
+            } else {
+                return shardRouting.isPromotableToPrimary();
+            }
         } else {
             return shardRouting.isSearchable();
         }

server/src/main/java/org/elasticsearch/index/cache/bitset/BitsetFilterCache.java

@@ -105,7 +105,7 @@ public final class BitsetFilterCache
         boolean loadFiltersEagerlySetting = settings.getValue(INDEX_LOAD_RANDOM_ACCESS_FILTERS_EAGERLY_SETTING);
         boolean isStateless = DiscoveryNode.isStateless(settings.getNodeSettings());
         if (isStateless) {
-            return DiscoveryNode.hasRole(settings.getNodeSettings(), DiscoveryNodeRole.INDEX_ROLE)
+            return DiscoveryNode.hasRole(settings.getNodeSettings(), DiscoveryNodeRole.SEARCH_ROLE)
                 && loadFiltersEagerlySetting
                 && INDEX_FAST_REFRESH_SETTING.get(settings.getSettings());
         } else {

server/src/test/java/org/elasticsearch/cluster/routing/IndexRoutingTableTests.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.cluster.routing;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.settings.Settings;
@@ -19,6 +20,7 @@ import org.mockito.Mockito;
 
 import java.util.List;
 
+import static org.elasticsearch.TransportVersions.FAST_REFRESH_RCO;
 import static org.elasticsearch.index.IndexSettings.INDEX_FAST_REFRESH_SETTING;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
@@ -27,16 +29,22 @@ import static org.mockito.Mockito.when;
 public class IndexRoutingTableTests extends ESTestCase {
 
     public void testReadyForSearch() {
-        innerReadyForSearch(false);
-        innerReadyForSearch(true);
+        innerReadyForSearch(false, false);
+        innerReadyForSearch(false, true);
+        innerReadyForSearch(true, false);
+        innerReadyForSearch(true, true);
     }
 
-    private void innerReadyForSearch(boolean fastRefresh) {
+    // TODO: remove if (fastRefresh && beforeFastRefreshRCO) branches (ES-9563)
+    private void innerReadyForSearch(boolean fastRefresh, boolean beforeFastRefreshRCO) {
         Index index = new Index(randomIdentifier(), UUIDs.randomBase64UUID());
         ClusterState clusterState = mock(ClusterState.class, Mockito.RETURNS_DEEP_STUBS);
         when(clusterState.metadata().index(any(Index.class)).getSettings()).thenReturn(
             Settings.builder().put(INDEX_FAST_REFRESH_SETTING.getKey(), fastRefresh).build()
         );
+        when(clusterState.getMinTransportVersion()).thenReturn(
+            beforeFastRefreshRCO ? TransportVersion.fromId(FAST_REFRESH_RCO.id() - 1_00_0) : TransportVersion.current()
+        );
         // 2 primaries that are search and index
         ShardId p1 = new ShardId(index, 0);
         IndexShardRoutingTable shardTable1 = new IndexShardRoutingTable(
@@ -55,7 +63,7 @@ public class IndexRoutingTableTests extends ESTestCase {
         shardTable1 = new IndexShardRoutingTable(p1, List.of(getShard(p1, true, ShardRoutingState.STARTED, ShardRouting.Role.INDEX_ONLY)));
         shardTable2 = new IndexShardRoutingTable(p2, List.of(getShard(p2, true, ShardRoutingState.STARTED, ShardRouting.Role.INDEX_ONLY)));
         indexRoutingTable = new IndexRoutingTable(index, new IndexShardRoutingTable[] { shardTable1, shardTable2 });
-        if (fastRefresh) {
+        if (fastRefresh && beforeFastRefreshRCO) {
             assertTrue(indexRoutingTable.readyForSearch(clusterState));
         } else {
             assertFalse(indexRoutingTable.readyForSearch(clusterState));
@@ -91,7 +99,7 @@ public class IndexRoutingTableTests extends ESTestCase {
             )
         );
         indexRoutingTable = new IndexRoutingTable(index, new IndexShardRoutingTable[] { shardTable1, shardTable2 });
-        if (fastRefresh) {
+        if (fastRefresh && beforeFastRefreshRCO) {
             assertTrue(indexRoutingTable.readyForSearch(clusterState));
         } else {
             assertFalse(indexRoutingTable.readyForSearch(clusterState));
@@ -118,8 +126,6 @@ public class IndexRoutingTableTests extends ESTestCase {
         assertTrue(indexRoutingTable.readyForSearch(clusterState));
 
         // 2 unassigned primaries that are index only with some replicas that are all available
-        // Fast refresh indices do not support replicas so this can not practically happen. If we add support we will want to ensure
-        // that readyForSearch allows for searching replicas when the index shard is not available.
         shardTable1 = new IndexShardRoutingTable(
             p1,
             List.of(
@@ -137,8 +143,8 @@ public class IndexRoutingTableTests extends ESTestCase {
             )
         );
         indexRoutingTable = new IndexRoutingTable(index, new IndexShardRoutingTable[] { shardTable1, shardTable2 });
-        if (fastRefresh) {
-            assertFalse(indexRoutingTable.readyForSearch(clusterState)); // if we support replicas for fast refreshes this needs to change
+        if (fastRefresh && beforeFastRefreshRCO) {
+            assertFalse(indexRoutingTable.readyForSearch(clusterState));
         } else {
             assertTrue(indexRoutingTable.readyForSearch(clusterState));
         }

server/src/test/java/org/elasticsearch/index/cache/bitset/BitSetFilterCacheTests.java

@@ -276,7 +276,7 @@ public class BitSetFilterCacheTests extends ESTestCase {
                     for (var isStateless : values) {
                         if (isStateless) {
                             assertEquals(
-                                loadFiltersEagerly && indexFastRefresh && hasIndexRole,
+                                loadFiltersEagerly && indexFastRefresh && hasIndexRole == false,
                                 BitsetFilterCache.shouldLoadRandomAccessFiltersEagerly(
                                     bitsetFilterCacheSettings(isStateless, hasIndexRole, loadFiltersEagerly, indexFastRefresh)
                                 )