Defer unpromotable shard refreshes until index refresh blocks are cleared (#120642)

This update postpones unpromotable refreshes for indices with an active INDEX_REFRESH_BLOCK 
until the block is cleared.
This ensures refresh operations proceed only when the index is no longer blocked.

To avoid indefinite delays, the maximum wait time is governed by the bulk request timeout
whereas for explicit refreshes it relies on the fact that the block will be removed eventually.

Closes ES-10134

Author: fcofdez

docs/changelog/120642.yaml

@@ -0,0 +1,5 @@
+pr: 120642
+summary: Defer unpromotable shard refreshes until index refresh blocks are cleared
+area: Engine
+type: enhancement
+issues: []

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

@@ -74,7 +74,14 @@ public TransportShardRefreshAction(
             ReplicaActionExecution.SubjectToCircuitBreaker
         );
         // registers the unpromotable version of shard refresh action
-        new TransportUnpromotableShardRefreshAction(clusterService, transportService, shardStateAction, actionFilters, indicesService);
+        new TransportUnpromotableShardRefreshAction(
+            clusterService,
+            transportService,
+            shardStateAction,
+            actionFilters,
+            indicesService,
+            threadPool
+        );
         this.refreshExecutor = transportService.getThreadPool().executor(ThreadPool.Names.REFRESH);
     }
 

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

@@ -9,20 +9,31 @@
 
 package org.elasticsearch.action.admin.indices.refresh;
 
+import org.elasticsearch.ElasticsearchTimeoutException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.broadcast.unpromotable.TransportBroadcastUnpromotableAction;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.action.shard.ShardStateAction;
+import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.cluster.metadata.MetadataCreateIndexService;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.injection.guice.Inject;
+import org.elasticsearch.node.NodeClosedException;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_REFRESH_BLOCK;
 
 public class TransportUnpromotableShardRefreshAction extends TransportBroadcastUnpromotableAction<
     UnpromotableShardRefreshRequest,
@@ -36,14 +47,37 @@ public class TransportUnpromotableShardRefreshAction extends TransportBroadcastU
     }
 
     private final IndicesService indicesService;
+    private final ThreadPool threadPool;
+    private final boolean useRefreshBlock;
 
     @Inject
     public TransportUnpromotableShardRefreshAction(
         ClusterService clusterService,
         TransportService transportService,
         ShardStateAction shardStateAction,
         ActionFilters actionFilters,
-        IndicesService indicesService
+        IndicesService indicesService,
+        ThreadPool threadPool
+    ) {
+        this(
+            clusterService,
+            transportService,
+            shardStateAction,
+            actionFilters,
+            indicesService,
+            threadPool,
+            MetadataCreateIndexService.useRefreshBlock(clusterService.getSettings())
+        );
+    }
+
+    public TransportUnpromotableShardRefreshAction(
+        ClusterService clusterService,
+        TransportService transportService,
+        ShardStateAction shardStateAction,
+        ActionFilters actionFilters,
+        IndicesService indicesService,
+        ThreadPool threadPool,
+        boolean useRefreshBlock
     ) {
         super(
             NAME,
@@ -55,6 +89,53 @@ public TransportUnpromotableShardRefreshAction(
             transportService.getThreadPool().executor(ThreadPool.Names.REFRESH)
         );
         this.indicesService = indicesService;
+        this.threadPool = threadPool;
+        this.useRefreshBlock = useRefreshBlock;
+    }
+
+    @Override
+    protected void doExecute(Task task, UnpromotableShardRefreshRequest request, ActionListener<ActionResponse.Empty> listener) {
+        beforeDispatchingRequestToUnpromotableShards(request, listener.delegateFailure((l, unused) -> super.doExecute(task, request, l)));
+    }
+
+    private void beforeDispatchingRequestToUnpromotableShards(UnpromotableShardRefreshRequest request, ActionListener<Void> listener) {
+        if (useRefreshBlock == false) {
+            listener.onResponse(null);
+            return;
+        }
+
+        var clusterStateObserver = new ClusterStateObserver(clusterService, request.getTimeout(), logger, threadPool.getThreadContext());
+
+        if (isIndexBlockedForRefresh(request.shardId().getIndexName(), clusterStateObserver.setAndGetObservedState()) == false) {
+            listener.onResponse(null);
+            return;
+        }
+
+        clusterStateObserver.waitForNextChange(new ClusterStateObserver.Listener() {
+            @Override
+            public void onNewClusterState(ClusterState state) {
+                listener.onResponse(null);
+            }
+
+            @Override
+            public void onClusterServiceClose() {
+                listener.onFailure(new NodeClosedException(clusterService.localNode()));
+            }
+
+            @Override
+            public void onTimeout(TimeValue timeout) {
+                listener.onFailure(
+                    new ElasticsearchTimeoutException(
+                        "shard refresh timed out waiting for index block to be removed",
+                        new ClusterBlockException(Map.of(request.shardId().getIndexName(), Set.of(INDEX_REFRESH_BLOCK)))
+                    )
+                );
+            }
+        }, clusterState -> isIndexBlockedForRefresh(request.shardId().getIndexName(), clusterState) == false);
+    }
+
+    private static boolean isIndexBlockedForRefresh(String index, ClusterState state) {
+        return state.blocks().hasIndexBlock(index, INDEX_REFRESH_BLOCK);
     }
 
     @Override

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

@@ -16,6 +16,8 @@
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.engine.Engine;
 
 import java.io.IOException;
@@ -26,22 +28,36 @@ public class UnpromotableShardRefreshRequest extends BroadcastUnpromotableReques
 
     private final long primaryTerm;
     private final long segmentGeneration;
+    private final TimeValue timeout;
 
     public UnpromotableShardRefreshRequest(
         IndexShardRoutingTable indexShardRoutingTable,
         long primaryTerm,
         long segmentGeneration,
         boolean failShardOnError
+    ) {
+        this(indexShardRoutingTable, primaryTerm, segmentGeneration, failShardOnError, null);
+    }
+
+    public UnpromotableShardRefreshRequest(
+        IndexShardRoutingTable indexShardRoutingTable,
+        long primaryTerm,
+        long segmentGeneration,
+        boolean failShardOnError,
+        @Nullable TimeValue timeout
     ) {
         super(indexShardRoutingTable, failShardOnError);
         this.primaryTerm = primaryTerm;
         this.segmentGeneration = segmentGeneration;
+        this.timeout = timeout;
     }
 
     public UnpromotableShardRefreshRequest(StreamInput in) throws IOException {
         super(in);
         segmentGeneration = in.readVLong();
         primaryTerm = in.getTransportVersion().onOrAfter(TransportVersions.V_8_12_0) ? in.readVLong() : Engine.UNKNOWN_PRIMARY_TERM;
+        // The timeout is only used by the request sender, therefore we don't write it over the wire
+        timeout = null;
     }
 
     @Override
@@ -70,6 +86,11 @@ public long getPrimaryTerm() {
         return primaryTerm;
     }
 
+    @Nullable
+    public TimeValue getTimeout() {
+        return timeout;
+    }
+
     @Override
     public String toString() {
         return Strings.format(

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

@@ -65,7 +65,7 @@ public void onFailure(Exception e) {
                 }
             });
             case IMMEDIATE -> immediate(indexShard, listener.delegateFailureAndWrap((l, r) -> {
-                if (indexShard.getReplicationGroup().getRoutingTable().unpromotableShards().size() > 0) {
+                if (indexShard.getReplicationGroup().getRoutingTable().allUnpromotableShards().size() > 0) {
                     sendUnpromotableRequests(indexShard, r.generation(), true, l, postWriteRefreshTimeout);
                 } else {
                     l.onResponse(true);
@@ -136,7 +136,8 @@ private void sendUnpromotableRequests(
             indexShard.getReplicationGroup().getRoutingTable(),
             indexShard.getOperationPrimaryTerm(),
             generation,
-            true
+            true,
+            postWriteRefreshTimeout
         );
         transportService.sendRequest(
             transportService.getLocalNode(),

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateIndexService.java

@@ -1761,7 +1761,7 @@ public static void validateStoreTypeSetting(Settings indexSettings) {
         }
     }
 
-    private static boolean useRefreshBlock(Settings settings) {
+    public static boolean useRefreshBlock(Settings settings) {
         return DiscoveryNode.isStateless(settings) && settings.getAsBoolean(USE_INDEX_REFRESH_BLOCK_SETTING_NAME, false);
     }
 

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

@@ -52,6 +52,7 @@ public class IndexShardRoutingTable {
     final List<ShardRouting> replicas;
     final List<ShardRouting> activeShards;
     final List<ShardRouting> assignedShards;
+    private final List<ShardRouting> assignedUnpromotableShards;
     private final List<ShardRouting> unpromotableShards;
     /**
      * The initializing list, including ones that are initializing on a target node because of relocation.
@@ -71,8 +72,9 @@ public class IndexShardRoutingTable {
         List<ShardRouting> replicas = new ArrayList<>();
         List<ShardRouting> activeShards = new ArrayList<>();
         List<ShardRouting> assignedShards = new ArrayList<>();
-        List<ShardRouting> unpromotableShards = new ArrayList<>();
+        List<ShardRouting> assignedUnpromotableShards = new ArrayList<>();
         List<ShardRouting> allInitializingShards = new ArrayList<>();
+        List<ShardRouting> unpromotableShards = new ArrayList<>();
         boolean allShardsStarted = true;
         int activeSearchShardCount = 0;
         int totalSearchShardCount = 0;
@@ -95,20 +97,24 @@ public class IndexShardRoutingTable {
             if (shard.initializing()) {
                 allInitializingShards.add(shard);
             }
+            if (shard.isPromotableToPrimary() == false) {
+                unpromotableShards.add(shard);
+            }
             if (shard.relocating()) {
                 // create the target initializing shard routing on the node the shard is relocating to
                 allInitializingShards.add(shard.getTargetRelocatingShard());
                 assert shard.assignedToNode() : "relocating from unassigned " + shard;
                 assert shard.getTargetRelocatingShard().assignedToNode() : "relocating to unassigned " + shard.getTargetRelocatingShard();
                 assignedShards.add(shard.getTargetRelocatingShard());
                 if (shard.getTargetRelocatingShard().isPromotableToPrimary() == false) {
+                    assignedUnpromotableShards.add(shard.getTargetRelocatingShard());
                     unpromotableShards.add(shard.getTargetRelocatingShard());
                 }
             }
             if (shard.assignedToNode()) {
                 assignedShards.add(shard);
                 if (shard.isPromotableToPrimary() == false) {
-                    unpromotableShards.add(shard);
+                    assignedUnpromotableShards.add(shard);
                 }
             }
             if (shard.state() != ShardRoutingState.STARTED) {
@@ -117,10 +123,13 @@ public class IndexShardRoutingTable {
         }
         assert shards.isEmpty() == false : "cannot have an empty shard routing table";
         assert primary != null : shards;
+        assert unpromotableShards.containsAll(assignedUnpromotableShards)
+            : unpromotableShards + " does not contain all assigned unpromotable shards " + assignedUnpromotableShards;
         this.primary = primary;
         this.replicas = CollectionUtils.wrapUnmodifiableOrEmptySingleton(replicas);
         this.activeShards = CollectionUtils.wrapUnmodifiableOrEmptySingleton(activeShards);
         this.assignedShards = CollectionUtils.wrapUnmodifiableOrEmptySingleton(assignedShards);
+        this.assignedUnpromotableShards = CollectionUtils.wrapUnmodifiableOrEmptySingleton(assignedUnpromotableShards);
         this.unpromotableShards = CollectionUtils.wrapUnmodifiableOrEmptySingleton(unpromotableShards);
         this.allInitializingShards = CollectionUtils.wrapUnmodifiableOrEmptySingleton(allInitializingShards);
         this.allShardsStarted = allShardsStarted;
@@ -185,6 +194,15 @@ public List<ShardRouting> assignedShards() {
      * @return a {@link List} of shards
      */
     public List<ShardRouting> unpromotableShards() {
+        return this.assignedUnpromotableShards;
+    }
+
+    /**
+     * Returns a {@link List} of all unpromotable shards, including unassigned shards
+     *
+     * @return a {@link List} of shards
+     */
+    public List<ShardRouting> allUnpromotableShards() {
         return this.unpromotableShards;
     }