Add RemoveBlock API to allow `DELETE /{index}/_block/{block}` (#129128)

Introduces a new `RemoveBlock` API that complements the existing `AddBlock` API by allowing users to remove index blocks using `DELETE /{index}/_block/{block}`.

Resolves #128966

---------

Co-authored-by: Niels Bauman <nielsbauman@gmail.com>

docs/changelog/129128.yaml

@@ -0,0 +1,5 @@
+pr: 129128
+summary: Add RemoveBlock API to allow `DELETE /{index}/_block/{block}`
+area: Indices APIs
+type: enhancement
+issues: []

docs/reference/elasticsearch/index-settings/index-block.md

@@ -143,3 +143,106 @@ The API returns following response:
 }
 ```
 
+
+## Remove index block API [remove-index-block]
+
+Removes an index block from an index.
+
+```console
+DELETE /my-index-000001/_block/write
+```
+
+
+### {{api-request-title}} [remove-index-block-api-request]
+
+`DELETE /<index>/_block/<block>`
+
+
+### {{api-path-parms-title}} [remove-index-block-api-path-params]
+
+`<index>`
+:   (Optional, string) Comma-separated list or wildcard expression of index names used to limit the request.
+
+    By default, you must explicitly name the indices you are removing blocks from. To allow the removal of blocks from indices with `_all`, `*`, or other wildcard expressions, change the `action.destructive_requires_name` setting to `false`. You can update this setting in the `elasticsearch.yml` file or using the [cluster update settings](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-cluster-put-settings) API.
+
+
+`<block>`
+:   (Required, string) Block type to remove from the index.
+
+    **Valid values**:
+
+    `metadata`
+    :   Remove metadata block, allowing metadata changes.
+
+    `read`
+    :   Remove read block, allowing read operations.
+
+    `read_only`
+    :   Remove read-only block, allowing write operations and metadata changes.
+
+    `write`
+    :   Remove write block, allowing write operations.
+
+    ::::
+
+
+
+### {{api-query-parms-title}} [remove-index-block-api-query-params]
+
+`allow_no_indices`
+:   (Optional, Boolean) If `false`, the request returns an error if any wildcard expression, [index alias](docs-content://manage-data/data-store/aliases.md), or `_all` value targets only missing or closed indices. This behavior applies even if the request targets other open indices. For example, a request targeting `foo*,bar*` returns an error if an index starts with `foo` but no index starts with `bar`.
+
+    Defaults to `true`.
+
+
+`expand_wildcards`
+:   (Optional, string) Type of index that wildcard patterns can match. If the request can target data streams, this argument determines whether wildcard expressions match hidden data streams. Supports comma-separated values, such as `open,hidden`. Valid values are:
+
+`all`
+:   Match any data stream or index, including [hidden](/reference/elasticsearch/rest-apis/api-conventions.md#multi-hidden) ones.
+
+`open`
+:   Match open, non-hidden indices. Also matches any non-hidden data stream.
+
+`closed`
+:   Match closed, non-hidden indices. Also matches any non-hidden data stream. Data streams cannot be closed.
+
+`hidden`
+:   Match hidden data streams and hidden indices. Must be combined with `open`, `closed`, or both.
+
+`none`
+:   Wildcard patterns are not accepted.
+
+Defaults to `open`.
+
+
+`ignore_unavailable`
+:   (Optional, Boolean) If `false`, the request returns an error if it targets a missing or closed index. Defaults to `false`.
+
+`master_timeout`
+:   (Optional, [time units](/reference/elasticsearch/rest-apis/api-conventions.md#time-units)) Period to wait for the master node. If the master node is not available before the timeout expires, the request fails and returns an error. Defaults to `30s`. Can also be set to `-1` to indicate that the request should never timeout.
+
+`timeout`
+:   (Optional, [time units](/reference/elasticsearch/rest-apis/api-conventions.md#time-units)) Period to wait for a response from all relevant nodes in the cluster after updating the cluster metadata. If no response is received before the timeout expires, the cluster metadata update still applies but the response will indicate that it was not completely acknowledged. Defaults to `30s`. Can also be set to `-1` to indicate that the request should never timeout.
+
+
+### {{api-examples-title}} [remove-index-block-api-example]
+
+The following example shows how to remove an index block:
+
+```console
+DELETE /my-index-000001/_block/write
+```
+
+The API returns following response:
+
+```console-result
+{
+  "acknowledged" : true,
+  "indices" : [ {
+    "name" : "my-index-000001",
+    "unblocked" : true
+  } ]
+}
+```
+

rest-api-spec/src/main/resources/rest-api-spec/api/indices.remove_block.json

@@ -0,0 +1,65 @@
+{
+  "indices.remove_block": {
+    "documentation": {
+      "url": "https://www.elastic.co/guide/en/elasticsearch/reference/master/index-modules-blocks.html",
+      "description": "Removes a block from an index."
+    },
+    "stability": "stable",
+    "visibility": "public",
+    "headers": {
+      "accept": [
+        "application/json"
+      ]
+    },
+    "url": {
+      "paths": [
+        {
+          "path": "/{index}/_block/{block}",
+          "methods": [
+            "DELETE"
+          ],
+          "parts": {
+            "index": {
+              "type": "list",
+              "description": "A comma separated list of indices to remove a block from"
+            },
+            "block": {
+              "type": "string",
+              "description": "The block to remove (one of read, write, read_only or metadata)"
+            }
+          }
+        }
+      ]
+    },
+    "params": {
+      "timeout": {
+        "type": "time",
+        "description": "Explicit operation timeout"
+      },
+      "master_timeout": {
+        "type": "time",
+        "description": "Specify timeout for connection to master"
+      },
+      "ignore_unavailable": {
+        "type": "boolean",
+        "description": "Whether specified concrete indices should be ignored when unavailable (missing or closed)"
+      },
+      "allow_no_indices": {
+        "type": "boolean",
+        "description": "Whether to ignore if a wildcard indices expression resolves into no concrete indices. (This includes `_all` string or when no indices have been specified)"
+      },
+      "expand_wildcards": {
+        "type": "enum",
+        "options": [
+          "open",
+          "closed",
+          "hidden",
+          "none",
+          "all"
+        ],
+        "default": "open",
+        "description": "Whether to expand wildcard expression to concrete indices that are open, closed or both."
+      }
+    }
+  }
+}

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/indices.blocks/10_basic.yml

@@ -28,3 +28,46 @@
         index: test_index
         body:
           index.blocks.write: false
+
+---
+"Test removing block via remove_block API":
+  - requires:
+      test_runner_features: [capabilities]
+      reason:  "index block APIs have only been made available in 9.1.0"
+      capabilities:
+        - method: DELETE
+          path: /{index}/_block/{block}
+  - do:
+      indices.create:
+        index: test_index_2
+
+  - do:
+      indices.add_block:
+        index: test_index_2
+        block: write
+  - is_true: acknowledged
+
+  - do:
+      catch: /cluster_block_exception/
+      index:
+        index:  test_index_2
+        body:   { foo: bar }
+
+  - do:
+      search:
+        index: test_index_2
+
+  - do:
+      indices.remove_block:
+        index: test_index_2
+        block: write
+  - is_true: acknowledged
+
+  - do:
+      index:
+        index: test_index_2
+        body: { foo: bar }
+
+  - do:
+      search:
+        index: test_index_2

server/src/internalClusterTest/java/org/elasticsearch/blocks/SimpleBlocksIT.java

@@ -14,6 +14,7 @@ import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
 import org.elasticsearch.action.admin.indices.readonly.AddIndexBlockResponse;
+import org.elasticsearch.action.admin.indices.readonly.RemoveIndexBlockResponse;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.action.support.PlainActionFuture;
@@ -549,6 +550,198 @@ public class SimpleBlocksIT extends ESIntegTestCase {
         disableIndexBlock(index, block.settingName());
     }
 
+    public void testRemoveBlockToMissingIndex() {
+        IndexNotFoundException e = expectThrows(
+            IndexNotFoundException.class,
+            indicesAdmin().prepareRemoveBlock(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, randomAddableBlock(), "test")
+        );
+        assertThat(e.getMessage(), is("no such index [test]"));
+    }
+
+    public void testRemoveBlockToOneMissingIndex() {
+        createIndex("test1");
+        final IndexNotFoundException e = expectThrows(
+            IndexNotFoundException.class,
+            indicesAdmin().prepareRemoveBlock(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, randomAddableBlock(), "test1", "test2")
+        );
+        assertThat(e.getMessage(), is("no such index [test2]"));
+    }
+
+    public void testRemoveBlockNoIndex() {
+        final ActionRequestValidationException e = expectThrows(
+            ActionRequestValidationException.class,
+            indicesAdmin().prepareRemoveBlock(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, randomAddableBlock())
+        );
+        assertThat(e.getMessage(), containsString("index is missing"));
+    }
+
+    public void testRemoveBlockNullIndex() {
+        expectThrows(
+            NullPointerException.class,
+            () -> indicesAdmin().prepareRemoveBlock(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, randomAddableBlock(), (String[]) null)
+        );
+    }
+
+    public void testCannotRemoveReadOnlyAllowDeleteBlock() {
+        createIndex("test1");
+        final ActionRequestValidationException e = expectThrows(
+            ActionRequestValidationException.class,
+            indicesAdmin().prepareRemoveBlock(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, APIBlock.READ_ONLY_ALLOW_DELETE, "test1")
+        );
+        assertThat(e.getMessage(), containsString("read_only_allow_delete block is for internal use only"));
+    }
+
+    public void testRemoveIndexBlock() throws Exception {
+        final String indexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        createIndex(indexName);
+        ensureGreen(indexName);
+
+        final int nbDocs = randomIntBetween(0, 50);
+        indexRandom(
+            randomBoolean(),
+            false,
+            randomBoolean(),
+            IntStream.range(0, nbDocs).mapToObj(i -> prepareIndex(indexName).setId(String.valueOf(i)).setSource("num", i)).collect(toList())
+        );
+
+        final APIBlock block = randomAddableBlock();
+        try {
+            // First add the block
+            AddIndexBlockResponse addResponse = indicesAdmin().prepareAddBlock(block, indexName).get();
+            assertTrue(
+                "Add block [" + block + "] to index [" + indexName + "] not acknowledged: " + addResponse,
+                addResponse.isAcknowledged()
+            );
+            assertIndexHasBlock(block, indexName);
+
+            // Then remove the block
+            RemoveIndexBlockResponse removeResponse = indicesAdmin().prepareRemoveBlock(
+                TEST_REQUEST_TIMEOUT,
+                TEST_REQUEST_TIMEOUT,
+                block,
+                indexName
+            ).get();
+            assertTrue(
+                "Remove block [" + block + "] from index [" + indexName + "] not acknowledged: " + removeResponse,
+                removeResponse.isAcknowledged()
+            );
+            assertIndexDoesNotHaveBlock(block, indexName);
+        } finally {
+            // Ensure cleanup
+            disableIndexBlock(indexName, block);
+        }
+
+        indicesAdmin().prepareRefresh(indexName).get();
+        assertHitCount(prepareSearch(indexName).setSize(0), nbDocs);
+    }
+
+    public void testRemoveBlockIdempotent() throws Exception {
+        final String indexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        createIndex(indexName);
+        ensureGreen(indexName);
+
+        final APIBlock block = randomAddableBlock();
+        try {
+            // First add the block
+            assertAcked(indicesAdmin().prepareAddBlock(block, indexName));
+            assertIndexHasBlock(block, indexName);
+
+            // Remove the block
+            assertAcked(indicesAdmin().prepareRemoveBlock(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, block, indexName));
+            assertIndexDoesNotHaveBlock(block, indexName);
+
+            // Second remove should be acked too (idempotent behavior)
+            assertAcked(indicesAdmin().prepareRemoveBlock(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, block, indexName));
+            assertIndexDoesNotHaveBlock(block, indexName);
+        } finally {
+            disableIndexBlock(indexName, block);
+        }
+    }
+
+    public void testRemoveBlockOneMissingIndexIgnoreMissing() throws Exception {
+        createIndex("test1");
+        final APIBlock block = randomAddableBlock();
+        try {
+            // First add the block to test1
+            assertAcked(indicesAdmin().prepareAddBlock(block, "test1"));
+            assertIndexHasBlock(block, "test1");
+
+            // Remove from both test1 and test2 (missing), with lenient options
+            assertBusy(
+                () -> assertAcked(
+                    indicesAdmin().prepareRemoveBlock(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, block, "test1", "test2")
+                        .setIndicesOptions(lenientExpandOpen())
+                )
+            );
+            assertIndexDoesNotHaveBlock(block, "test1");
+        } finally {
+            disableIndexBlock("test1", block);
+        }
+    }
+
+    static void assertIndexDoesNotHaveBlock(APIBlock block, final String... indices) {
+        final ClusterState clusterState = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState();
+        final ProjectId projectId = Metadata.DEFAULT_PROJECT_ID;
+        for (String index : indices) {
+            final IndexMetadata indexMetadata = clusterState.metadata().getProject(projectId).indices().get(index);
+            final Settings indexSettings = indexMetadata.getSettings();
+            assertThat(
+                "Index " + index + " should not have block setting [" + block.settingName() + "]",
+                indexSettings.getAsBoolean(block.settingName(), false),
+                is(false)
+            );
+            assertThat(
+                "Index " + index + " should not have block [" + block.getBlock() + "]",
+                clusterState.blocks().hasIndexBlock(projectId, index, block.getBlock()),
+                is(false)
+            );
+        }
+    }
+
+    public void testRemoveWriteBlockAlsoRemovesVerifiedReadOnlySetting() throws Exception {
+        final String indexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        createIndex(indexName);
+        ensureGreen(indexName);
+
+        try {
+            // Add write block
+            AddIndexBlockResponse addResponse = indicesAdmin().prepareAddBlock(APIBlock.WRITE, indexName).get();
+            assertTrue("Add write block not acknowledged: " + addResponse, addResponse.isAcknowledged());
+            assertIndexHasBlock(APIBlock.WRITE, indexName);
+
+            // Verify VERIFIED_READ_ONLY_SETTING is set
+            ClusterState state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState();
+            IndexMetadata indexMetadata = state.metadata().getProject(Metadata.DEFAULT_PROJECT_ID).index(indexName);
+            assertThat(
+                "VERIFIED_READ_ONLY_SETTING should be true",
+                MetadataIndexStateService.VERIFIED_READ_ONLY_SETTING.get(indexMetadata.getSettings()),
+                is(true)
+            );
+
+            // Remove write block
+            RemoveIndexBlockResponse removeResponse = indicesAdmin().prepareRemoveBlock(
+                TEST_REQUEST_TIMEOUT,
+                TEST_REQUEST_TIMEOUT,
+                APIBlock.WRITE,
+                indexName
+            ).get();
+            assertTrue("Remove write block not acknowledged: " + removeResponse, removeResponse.isAcknowledged());
+            assertIndexDoesNotHaveBlock(APIBlock.WRITE, indexName);
+
+            // Verify VERIFIED_READ_ONLY_SETTING is also removed
+            state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState();
+            indexMetadata = state.metadata().getProject(Metadata.DEFAULT_PROJECT_ID).index(indexName);
+            assertThat(
+                "VERIFIED_READ_ONLY_SETTING should be false after removing write block",
+                MetadataIndexStateService.VERIFIED_READ_ONLY_SETTING.get(indexMetadata.getSettings()),
+                is(false)
+            );
+
+        } finally {
+            disableIndexBlock(indexName, APIBlock.WRITE);
+        }
+    }
+
     /**
      * The read-only-allow-delete block cannot be added via the add index block API; this method chooses randomly from the values that
      * the add index block API does support.

server/src/internalClusterTest/java/org/elasticsearch/operateAllIndices/DestructiveOperationsIT.java

@@ -155,4 +155,46 @@ public class DestructiveOperationsIT extends ESIntegTestCase {
         assertTrue("write block is set on index1", state.getBlocks().hasIndexBlock("index1", IndexMetadata.INDEX_WRITE_BLOCK));
         assertTrue("write block is set on 1index", state.getBlocks().hasIndexBlock("1index", IndexMetadata.INDEX_WRITE_BLOCK));
     }
+
+    public void testRemoveIndexBlockIsRejected() throws Exception {
+        updateClusterSettings(Settings.builder().put(DestructiveOperations.REQUIRES_NAME_SETTING.getKey(), true));
+
+        createIndex("index1", "1index");
+        // Add blocks first
+        assertAcked(indicesAdmin().prepareAddBlock(WRITE, "index1", "1index").get());
+
+        // Test rejected wildcard patterns (while blocks still exist)
+        expectThrows(
+            IllegalArgumentException.class,
+            indicesAdmin().prepareRemoveBlock(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, WRITE, "i*")
+        );
+        expectThrows(
+            IllegalArgumentException.class,
+            indicesAdmin().prepareRemoveBlock(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, WRITE, "_all")
+        );
+
+        // Test successful requests (exact names and non-destructive patterns)
+        assertAcked(indicesAdmin().prepareRemoveBlock(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, WRITE, "1index").get());
+        assertAcked(indicesAdmin().prepareRemoveBlock(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, WRITE, "*", "-*").get());
+    }
+
+    public void testRemoveIndexBlockDefaultBehaviour() throws Exception {
+        if (randomBoolean()) {
+            updateClusterSettings(Settings.builder().put(DestructiveOperations.REQUIRES_NAME_SETTING.getKey(), false));
+        }
+
+        createIndex("index1", "1index");
+        // Add blocks first
+        assertAcked(indicesAdmin().prepareAddBlock(WRITE, "index1", "1index").get());
+
+        if (randomBoolean()) {
+            assertAcked(indicesAdmin().prepareRemoveBlock(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, WRITE, "_all").get());
+        } else {
+            assertAcked(indicesAdmin().prepareRemoveBlock(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, WRITE, "*").get());
+        }
+
+        ClusterState state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState();
+        assertFalse("write block is removed from index1", state.getBlocks().hasIndexBlock("index1", IndexMetadata.INDEX_WRITE_BLOCK));
+        assertFalse("write block is removed from 1index", state.getBlocks().hasIndexBlock("1index", IndexMetadata.INDEX_WRITE_BLOCK));
+    }
 }

server/src/main/java/org/elasticsearch/action/ActionModule.java

@@ -114,6 +114,7 @@ import org.elasticsearch.action.admin.indices.mapping.put.TransportPutMappingAct
 import org.elasticsearch.action.admin.indices.open.OpenIndexAction;
 import org.elasticsearch.action.admin.indices.open.TransportOpenIndexAction;
 import org.elasticsearch.action.admin.indices.readonly.TransportAddIndexBlockAction;
+import org.elasticsearch.action.admin.indices.readonly.TransportRemoveIndexBlockAction;
 import org.elasticsearch.action.admin.indices.readonly.TransportVerifyShardIndexBlockAction;
 import org.elasticsearch.action.admin.indices.recovery.RecoveryAction;
 import org.elasticsearch.action.admin.indices.recovery.TransportRecoveryAction;
@@ -343,6 +344,7 @@ import org.elasticsearch.rest.action.admin.indices.RestPutMappingAction;
 import org.elasticsearch.rest.action.admin.indices.RestRecoveryAction;
 import org.elasticsearch.rest.action.admin.indices.RestRefreshAction;
 import org.elasticsearch.rest.action.admin.indices.RestReloadAnalyzersAction;
+import org.elasticsearch.rest.action.admin.indices.RestRemoveIndexBlockAction;
 import org.elasticsearch.rest.action.admin.indices.RestResizeHandler;
 import org.elasticsearch.rest.action.admin.indices.RestResolveClusterAction;
 import org.elasticsearch.rest.action.admin.indices.RestResolveIndexAction;
@@ -687,6 +689,7 @@ public class ActionModule extends AbstractModule {
         actions.register(OpenIndexAction.INSTANCE, TransportOpenIndexAction.class);
         actions.register(TransportCloseIndexAction.TYPE, TransportCloseIndexAction.class);
         actions.register(TransportAddIndexBlockAction.TYPE, TransportAddIndexBlockAction.class);
+        actions.register(TransportRemoveIndexBlockAction.TYPE, TransportRemoveIndexBlockAction.class);
         actions.register(GetMappingsAction.INSTANCE, TransportGetMappingsAction.class);
         actions.register(GetFieldMappingsAction.INSTANCE, TransportGetFieldMappingsAction.class);
         actions.register(TransportGetFieldMappingsIndexAction.TYPE, TransportGetFieldMappingsIndexAction.class);
@@ -889,6 +892,7 @@ public class ActionModule extends AbstractModule {
         registerHandler.accept(new RestCloseIndexAction());
         registerHandler.accept(new RestOpenIndexAction());
         registerHandler.accept(new RestAddIndexBlockAction());
+        registerHandler.accept(new RestRemoveIndexBlockAction());
         registerHandler.accept(new RestGetHealthAction());
         registerHandler.accept(new RestPrevalidateNodeRemovalAction());
 

server/src/main/java/org/elasticsearch/action/admin/indices/readonly/RemoveIndexBlockRequest.java

@@ -0,0 +1,150 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.action.admin.indices.readonly;
+
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.IndicesRequest;
+import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.action.support.master.AcknowledgedRequest;
+import org.elasticsearch.cluster.metadata.IndexMetadata.APIBlock;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.util.CollectionUtils;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.tasks.CancellableTask;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.elasticsearch.action.ValidateActions.addValidationError;
+
+/**
+ * A request to remove a block from an index.
+ */
+public class RemoveIndexBlockRequest extends AcknowledgedRequest<RemoveIndexBlockRequest> implements IndicesRequest.Replaceable {
+
+    private final APIBlock block;
+    private String[] indices;
+    private IndicesOptions indicesOptions = IndicesOptions.strictExpandOpen();
+
+    public RemoveIndexBlockRequest(StreamInput in) throws IOException {
+        super(in);
+        indices = in.readStringArray();
+        indicesOptions = IndicesOptions.readIndicesOptions(in);
+        block = APIBlock.readFrom(in);
+    }
+
+    /**
+     * Constructs a new request for the specified block and indices
+     */
+    public RemoveIndexBlockRequest(TimeValue masterTimeout, TimeValue ackTimeout, APIBlock block, String... indices) {
+        super(masterTimeout, ackTimeout);
+        this.block = Objects.requireNonNull(block);
+        this.indices = Objects.requireNonNull(indices);
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        ActionRequestValidationException validationException = null;
+        if (CollectionUtils.isEmpty(indices)) {
+            validationException = addValidationError("index is missing", validationException);
+        }
+        if (block == APIBlock.READ_ONLY_ALLOW_DELETE) {
+            validationException = addValidationError("read_only_allow_delete block is for internal use only", validationException);
+        }
+        return validationException;
+    }
+
+    /**
+     * Returns the indices to have blocks removed
+     */
+    @Override
+    public String[] indices() {
+        return indices;
+    }
+
+    /**
+     * Sets the indices to have blocks removed
+     * @param indices the indices to have blocks removed
+     * @return the request itself
+     */
+    @Override
+    public RemoveIndexBlockRequest indices(String... indices) {
+        this.indices = Objects.requireNonNull(indices);
+        return this;
+    }
+
+    /**
+     * Specifies what type of requested indices to ignore and how to deal with wildcard expressions.
+     * For example indices that don't exist.
+     *
+     * @return the desired behaviour regarding indices to ignore and wildcard indices expressions
+     */
+    @Override
+    public IndicesOptions indicesOptions() {
+        return indicesOptions;
+    }
+
+    /**
+     * Specifies what type of requested indices to ignore and how to deal wild wildcard expressions.
+     * For example indices that don't exist.
+     *
+     * @param indicesOptions the desired behaviour regarding indices to ignore and wildcard indices expressions
+     * @return the request itself
+     */
+    public RemoveIndexBlockRequest indicesOptions(IndicesOptions indicesOptions) {
+        this.indicesOptions = indicesOptions;
+        return this;
+    }
+
+    /**
+     * Returns the block to be removed
+     */
+    public APIBlock getBlock() {
+        return block;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeStringArray(indices);
+        indicesOptions.writeIndicesOptions(out);
+        block.writeTo(out);
+    }
+
+    @Override
+    public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+        return new CancellableTask(id, type, action, "", parentTaskId, headers);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        RemoveIndexBlockRequest that = (RemoveIndexBlockRequest) o;
+        return block == that.block && Arrays.equals(indices, that.indices) && Objects.equals(indicesOptions, that.indicesOptions);
+    }
+
+    @Override
+    public int hashCode() {
+        int result = Objects.hash(block, indicesOptions);
+        result = 31 * result + Arrays.hashCode(indices);
+        return result;
+    }
+
+}

server/src/main/java/org/elasticsearch/action/admin/indices/readonly/RemoveIndexBlockRequestBuilder.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.action.admin.indices.readonly;
+
+import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.action.support.master.AcknowledgedRequestBuilder;
+import org.elasticsearch.client.internal.ElasticsearchClient;
+import org.elasticsearch.cluster.metadata.IndexMetadata.APIBlock;
+import org.elasticsearch.core.TimeValue;
+
+/**
+ * Builder for remove index block request
+ */
+public class RemoveIndexBlockRequestBuilder extends AcknowledgedRequestBuilder<
+    RemoveIndexBlockRequest,
+    RemoveIndexBlockResponse,
+    RemoveIndexBlockRequestBuilder> {
+
+    public RemoveIndexBlockRequestBuilder(
+        ElasticsearchClient client,
+        TimeValue masterTimeout,
+        TimeValue ackTimeout,
+        APIBlock block,
+        String... indices
+    ) {
+        super(client, TransportRemoveIndexBlockAction.TYPE, new RemoveIndexBlockRequest(masterTimeout, ackTimeout, block, indices));
+    }
+
+    /**
+     * Sets the indices to be unblocked
+     *
+     * @param indices the indices to be unblocked
+     * @return the request itself
+     */
+    public RemoveIndexBlockRequestBuilder setIndices(String... indices) {
+        request.indices(indices);
+        return this;
+    }
+
+    /**
+     * Specifies what type of requested indices to ignore and how to deal with wildcard expressions.
+     * For example indices that don't exist.
+     *
+     * @param indicesOptions the desired behaviour regarding indices to ignore and wildcard indices expressions
+     * @return the request itself
+     */
+    public RemoveIndexBlockRequestBuilder setIndicesOptions(IndicesOptions indicesOptions) {
+        request.indicesOptions(indicesOptions);
+        return this;
+    }
+}

server/src/main/java/org/elasticsearch/action/admin/indices/readonly/RemoveIndexBlockResponse.java

@@ -0,0 +1,132 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+package org.elasticsearch.action.admin.indices.readonly;
+
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.xcontent.ToXContentObject;
+import org.elasticsearch.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * A response for a remove index block action.
+ */
+public class RemoveIndexBlockResponse extends AcknowledgedResponse {
+
+    public static final RemoveIndexBlockResponse EMPTY = new RemoveIndexBlockResponse(true, List.of());
+
+    private final List<RemoveBlockResult> results;
+
+    public RemoveIndexBlockResponse(StreamInput in) throws IOException {
+        super(in);
+        results = in.readCollectionAsImmutableList(RemoveBlockResult::new);
+    }
+
+    public RemoveIndexBlockResponse(boolean acknowledged, List<RemoveBlockResult> results) {
+        super(acknowledged);
+        this.results = List.copyOf(Objects.requireNonNull(results, "results must not be null"));
+    }
+
+    /**
+     * Returns the list of {@link RemoveBlockResult}.
+     */
+    public List<RemoveBlockResult> getResults() {
+        return results;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeCollection(results);
+    }
+
+    @Override
+    protected void addCustomFields(XContentBuilder builder, Params params) throws IOException {
+        builder.startArray("indices");
+        for (RemoveBlockResult result : results) {
+            result.toXContent(builder, params);
+        }
+        builder.endArray();
+    }
+
+    @Override
+    public String toString() {
+        return Strings.toString(this);
+    }
+
+    public static class RemoveBlockResult implements Writeable, ToXContentObject {
+
+        private final Index index;
+        private final @Nullable Exception exception;
+
+        public RemoveBlockResult(final Index index) {
+            this.index = Objects.requireNonNull(index);
+            this.exception = null;
+        }
+
+        public RemoveBlockResult(final Index index, final Exception failure) {
+            this.index = Objects.requireNonNull(index);
+            this.exception = Objects.requireNonNull(failure);
+        }
+
+        RemoveBlockResult(final StreamInput in) throws IOException {
+            this.index = new Index(in);
+            this.exception = in.readException();
+        }
+
+        @Override
+        public void writeTo(final StreamOutput out) throws IOException {
+            index.writeTo(out);
+            out.writeException(exception);
+        }
+
+        public Index getIndex() {
+            return index;
+        }
+
+        public Exception getException() {
+            return exception;
+        }
+
+        public boolean hasFailures() {
+            return exception != null;
+        }
+
+        @Override
+        public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
+            builder.startObject();
+            {
+                builder.field("name", index.getName());
+                if (hasFailures()) {
+                    builder.startObject("exception");
+                    ElasticsearchException.generateFailureXContent(builder, params, exception, true);
+                    builder.endObject();
+                } else {
+                    builder.field("unblocked", true);
+                }
+            }
+            return builder.endObject();
+        }
+
+        @Override
+        public String toString() {
+            return Strings.toString(this);
+        }
+    }
+
+}

server/src/main/java/org/elasticsearch/action/admin/indices/readonly/TransportRemoveIndexBlockAction.java

@@ -0,0 +1,148 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.action.admin.indices.readonly;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.DestructiveOperations;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.action.support.master.TransportMasterNodeAction;
+import org.elasticsearch.cluster.AckedClusterStateUpdateTask;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateAckListener;
+import org.elasticsearch.cluster.SimpleBatchedAckListenerTaskExecutor;
+import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.cluster.block.ClusterBlockLevel;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.metadata.MetadataIndexStateService;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
+import org.elasticsearch.cluster.project.ProjectResolver;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
+import org.elasticsearch.common.Priority;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
+import org.elasticsearch.core.Tuple;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.injection.guice.Inject;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Removes a single index level block from a given set of indices. This action removes the block setting
+ * and updates the cluster state to reflect the change.
+ */
+public class TransportRemoveIndexBlockAction extends TransportMasterNodeAction<RemoveIndexBlockRequest, RemoveIndexBlockResponse> {
+
+    public static final ActionType<RemoveIndexBlockResponse> TYPE = new ActionType<>("indices:admin/block/remove");
+
+    private final ProjectResolver projectResolver;
+    private final IndexNameExpressionResolver indexNameExpressionResolver;
+    private final DestructiveOperations destructiveOperations;
+    private final MasterServiceTaskQueue<AckedClusterStateUpdateTask> removeBlocksQueue;
+
+    @Inject
+    public TransportRemoveIndexBlockAction(
+        TransportService transportService,
+        ClusterService clusterService,
+        ThreadPool threadPool,
+        ActionFilters actionFilters,
+        ProjectResolver projectResolver,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        DestructiveOperations destructiveOperations
+    ) {
+        super(
+            TYPE.name(),
+            transportService,
+            clusterService,
+            threadPool,
+            actionFilters,
+            RemoveIndexBlockRequest::new,
+            RemoveIndexBlockResponse::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
+        this.projectResolver = projectResolver;
+        this.indexNameExpressionResolver = indexNameExpressionResolver;
+        this.destructiveOperations = destructiveOperations;
+
+        removeBlocksQueue = clusterService.createTaskQueue("remove-blocks", Priority.URGENT, new SimpleBatchedAckListenerTaskExecutor<>() {
+            @Override
+            public Tuple<ClusterState, ClusterStateAckListener> executeTask(AckedClusterStateUpdateTask task, ClusterState clusterState)
+                throws Exception {
+                return Tuple.tuple(task.execute(clusterState), task);
+            }
+        });
+    }
+
+    @Override
+    protected void doExecute(Task task, RemoveIndexBlockRequest request, ActionListener<RemoveIndexBlockResponse> listener) {
+        destructiveOperations.failDestructive(request.indices());
+        super.doExecute(task, request, listener);
+    }
+
+    @Override
+    protected ClusterBlockException checkBlock(RemoveIndexBlockRequest request, ClusterState state) {
+        if (request.getBlock().getBlock().levels().contains(ClusterBlockLevel.METADATA_WRITE)
+            && state.blocks().global(ClusterBlockLevel.METADATA_WRITE).isEmpty()) {
+            return null;
+        }
+        final ProjectMetadata projectMetadata = projectResolver.getProjectMetadata(state);
+        return state.blocks()
+            .indicesBlockedException(
+                projectMetadata.id(),
+                ClusterBlockLevel.METADATA_WRITE,
+                indexNameExpressionResolver.concreteIndexNames(projectMetadata, request)
+            );
+    }
+
+    @Override
+    protected void masterOperation(
+        final Task task,
+        final RemoveIndexBlockRequest request,
+        final ClusterState state,
+        final ActionListener<RemoveIndexBlockResponse> listener
+    ) throws Exception {
+        final Index[] concreteIndices = indexNameExpressionResolver.concreteIndices(state, request);
+        if (concreteIndices == null || concreteIndices.length == 0) {
+            listener.onResponse(RemoveIndexBlockResponse.EMPTY);
+            return;
+        }
+
+        final var projectId = projectResolver.getProjectId();
+        removeBlocksQueue.submitTask(
+            "remove-index-block-[" + request.getBlock().toString() + "]-" + Arrays.toString(concreteIndices),
+            new AckedClusterStateUpdateTask(request, listener) {
+                private List<RemoveIndexBlockResponse.RemoveBlockResult> results;
+
+                @Override
+                public ClusterState execute(ClusterState currentState) {
+                    final var tuple = MetadataIndexStateService.removeIndexBlock(
+                        currentState.projectState(projectId),
+                        concreteIndices,
+                        request.getBlock()
+                    );
+                    results = tuple.v2();
+                    return tuple.v1();
+                }
+
+                @Override
+                protected AcknowledgedResponse newResponse(boolean acknowledged) {
+                    return new RemoveIndexBlockResponse(acknowledged, results);
+                }
+            },
+            request.masterNodeTimeout()
+        );
+    }
+
+}

server/src/main/java/org/elasticsearch/client/internal/IndicesAdminClient.java

@@ -66,7 +66,11 @@ import org.elasticsearch.action.admin.indices.open.OpenIndexResponse;
 import org.elasticsearch.action.admin.indices.readonly.AddIndexBlockRequest;
 import org.elasticsearch.action.admin.indices.readonly.AddIndexBlockRequestBuilder;
 import org.elasticsearch.action.admin.indices.readonly.AddIndexBlockResponse;
+import org.elasticsearch.action.admin.indices.readonly.RemoveIndexBlockRequest;
+import org.elasticsearch.action.admin.indices.readonly.RemoveIndexBlockRequestBuilder;
+import org.elasticsearch.action.admin.indices.readonly.RemoveIndexBlockResponse;
 import org.elasticsearch.action.admin.indices.readonly.TransportAddIndexBlockAction;
+import org.elasticsearch.action.admin.indices.readonly.TransportRemoveIndexBlockAction;
 import org.elasticsearch.action.admin.indices.recovery.RecoveryAction;
 import org.elasticsearch.action.admin.indices.recovery.RecoveryRequest;
 import org.elasticsearch.action.admin.indices.recovery.RecoveryRequestBuilder;
@@ -219,6 +223,19 @@ public class IndicesAdminClient implements ElasticsearchClient {
         execute(TransportAddIndexBlockAction.TYPE, request, listener);
     }
 
+    public RemoveIndexBlockRequestBuilder prepareRemoveBlock(
+        TimeValue masterTimeout,
+        TimeValue ackTimeout,
+        APIBlock block,
+        String... indices
+    ) {
+        return new RemoveIndexBlockRequestBuilder(this, masterTimeout, ackTimeout, block, indices);
+    }
+
+    public void removeBlock(RemoveIndexBlockRequest request, ActionListener<RemoveIndexBlockResponse> listener) {
+        execute(TransportRemoveIndexBlockAction.TYPE, request, listener);
+    }
+
     public OpenIndexRequestBuilder prepareOpen(String... indices) {
         return new OpenIndexRequestBuilder(this, indices);
     }

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

@@ -316,6 +316,11 @@ public class IndexMetadata implements Diffable<IndexMetadata>, ToXContentFragmen
             return block;
         }
 
+        @Override
+        public String toString() {
+            return name;
+        }
+
         public static APIBlock fromName(String name) {
             for (APIBlock block : APIBlock.values()) {
                 if (block.name.equals(name)) {

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

@@ -25,6 +25,7 @@ import org.elasticsearch.action.admin.indices.readonly.AddIndexBlockClusterState
 import org.elasticsearch.action.admin.indices.readonly.AddIndexBlockResponse;
 import org.elasticsearch.action.admin.indices.readonly.AddIndexBlockResponse.AddBlockResult;
 import org.elasticsearch.action.admin.indices.readonly.AddIndexBlockResponse.AddBlockShardResult;
+import org.elasticsearch.action.admin.indices.readonly.RemoveIndexBlockResponse.RemoveBlockResult;
 import org.elasticsearch.action.admin.indices.readonly.TransportVerifyShardIndexBlockAction;
 import org.elasticsearch.action.support.ActiveShardsObserver;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
@@ -69,6 +70,7 @@ import org.elasticsearch.index.IndexReshardService;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.shard.IndexLongFieldRange;
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.indices.IndexClosedException;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.indices.ShardLimitValidator;
 import org.elasticsearch.injection.guice.Inject;
@@ -1157,6 +1159,72 @@ public class MetadataIndexStateService {
         );
     }
 
+    public static Tuple<ClusterState, List<RemoveBlockResult>> removeIndexBlock(
+        ProjectState projectState,
+        final Index[] indices,
+        final APIBlock block
+    ) {
+        final ProjectMetadata.Builder projectBuilder = ProjectMetadata.builder(projectState.metadata());
+        final ClusterBlocks.Builder blocks = ClusterBlocks.builder(projectState.blocks());
+        final List<String> effectivelyUnblockedIndices = new ArrayList<>();
+        final Map<String, RemoveBlockResult> results = new HashMap<>();
+
+        for (Index index : indices) {
+            try {
+                final IndexMetadata indexMetadata = projectState.metadata().getIndexSafe(index);
+                if (indexMetadata.getState() == IndexMetadata.State.CLOSE) {
+                    results.put(index.getName(), new RemoveBlockResult(index, new IndexClosedException(index)));
+                    continue;
+                }
+
+                final Settings indexSettings = indexMetadata.getSettings();
+                final boolean hasBlockSetting = block.setting().get(indexSettings);
+
+                final boolean hasBlock = projectState.blocks().hasIndexBlock(projectState.projectId(), index.getName(), block.block);
+                if (hasBlockSetting == false && hasBlock == false) {
+                    results.put(index.getName(), new RemoveBlockResult(index));
+                    continue;
+                }
+
+                // Remove all blocks with the same ID
+                blocks.removeIndexBlock(projectState.projectId(), index.getName(), block.block);
+
+                // Remove the block setting if it exists
+                if (hasBlockSetting) {
+                    final Settings.Builder updatedSettings = Settings.builder().put(indexSettings);
+                    updatedSettings.remove(block.settingName());
+
+                    if (block.block.contains(ClusterBlockLevel.WRITE)) {
+                        if (blocks.hasIndexBlockLevel(projectState.projectId(), index.getName(), ClusterBlockLevel.WRITE) == false) {
+                            updatedSettings.remove(VERIFIED_READ_ONLY_SETTING.getKey());
+                        }
+                    }
+
+                    final IndexMetadata updatedMetadata = IndexMetadata.builder(indexMetadata)
+                        .settings(updatedSettings)
+                        .settingsVersion(indexMetadata.getSettingsVersion() + 1)
+                        .build();
+
+                    projectBuilder.put(updatedMetadata, true);
+                }
+
+                effectivelyUnblockedIndices.add(index.getName());
+                results.put(index.getName(), new RemoveBlockResult(index));
+
+                logger.debug("remove block {} from index {} succeeded", block.block, index);
+            } catch (final IndexNotFoundException e) {
+                logger.debug("index {} has been deleted since removing block started, ignoring", index);
+                results.put(index.getName(), new RemoveBlockResult(index, e));
+            }
+        }
+
+        logger.info("completed removing [index.blocks.{}] block from indices {}", block.name, effectivelyUnblockedIndices);
+        return Tuple.tuple(
+            ClusterState.builder(projectState.cluster()).putProjectMetadata(projectBuilder).blocks(blocks).build(),
+            List.copyOf(results.values())
+        );
+    }
+
     private class OpenIndicesExecutor implements ClusterStateTaskExecutor<OpenIndicesTask> {
 
         @Override

server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestRemoveIndexBlockAction.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.rest.action.admin.indices;
+
+import org.elasticsearch.action.admin.indices.readonly.RemoveIndexBlockRequest;
+import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.rest.BaseRestHandler;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.ServerlessScope;
+import org.elasticsearch.rest.action.RestCancellableNodeClient;
+import org.elasticsearch.rest.action.RestToXContentListener;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.elasticsearch.rest.RestRequest.Method.DELETE;
+import static org.elasticsearch.rest.RestUtils.getAckTimeout;
+import static org.elasticsearch.rest.RestUtils.getMasterNodeTimeout;
+import static org.elasticsearch.rest.Scope.PUBLIC;
+
+@ServerlessScope(PUBLIC)
+public class RestRemoveIndexBlockAction extends BaseRestHandler {
+
+    @Override
+    public List<Route> routes() {
+        return List.of(new Route(DELETE, "/{index}/_block/{block}"));
+    }
+
+    @Override
+    public String getName() {
+        return "remove_index_block_action";
+    }
+
+    @Override
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
+        RemoveIndexBlockRequest removeIndexBlockRequest = new RemoveIndexBlockRequest(
+            getMasterNodeTimeout(request),
+            getAckTimeout(request),
+            IndexMetadata.APIBlock.fromName(request.param("block")),
+            Strings.splitStringByCommaToArray(request.param("index"))
+        );
+        removeIndexBlockRequest.indicesOptions(IndicesOptions.fromRequest(request, removeIndexBlockRequest.indicesOptions()));
+        return channel -> new RestCancellableNodeClient(client, request.getHttpChannel()).admin()
+            .indices()
+            .removeBlock(removeIndexBlockRequest, new RestToXContentListener<>(channel));
+    }
+}

server/src/test/java/org/elasticsearch/cluster/metadata/MetadataIndexStateServiceBatchingTests.java

@@ -208,6 +208,59 @@ public class MetadataIndexStateServiceBatchingTests extends ESSingleNodeTestCase
         clusterService.removeListener(assertingListener);
     }
 
+    public void testBatchRemoveBlocks() throws Exception {
+        final var clusterService = getInstanceFromNode(ClusterService.class);
+        final var masterService = clusterService.getMasterService();
+
+        // create some indices and add blocks
+        createIndex("test-1", indicesAdmin().prepareCreate("test-1"));
+        createIndex("test-2", indicesAdmin().prepareCreate("test-2"));
+        createIndex("test-3", indicesAdmin().prepareCreate("test-3"));
+        assertAcked(indicesAdmin().prepareAddBlock(APIBlock.WRITE, "test-1", "test-2", "test-3"));
+        ensureGreen("test-1", "test-2", "test-3");
+
+        final var assertingListener = unblockedIndexCountListener();
+        clusterService.addListener(assertingListener);
+
+        final var block1 = blockMasterService(masterService);
+        block1.run(); // wait for block
+
+        // fire off some remove blocks
+        final var future1 = indicesAdmin().prepareRemoveBlock(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, APIBlock.WRITE, "test-1")
+            .execute();
+        final var future2 = indicesAdmin().prepareRemoveBlock(
+            TEST_REQUEST_TIMEOUT,
+            TEST_REQUEST_TIMEOUT,
+            APIBlock.WRITE,
+            "test-2",
+            "test-3"
+        ).execute();
+
+        // check the queue for the remove-block tasks
+        assertThat(findPendingTasks(masterService, "remove-index-block-[write]"), hasSize(2));
+
+        block1.run(); // release block
+
+        // assert that the requests were acknowledged
+        final var resp1 = future1.get();
+        assertAcked(resp1);
+        assertThat(resp1.getResults(), hasSize(1));
+        assertThat(resp1.getResults().get(0).getIndex().getName(), is("test-1"));
+
+        final var resp2 = future2.get();
+        assertAcked(resp2);
+        assertThat(resp2.getResults(), hasSize(2));
+        assertThat(resp2.getResults().stream().map(r -> r.getIndex().getName()).toList(), containsInAnyOrder("test-2", "test-3"));
+
+        // and assert that all the blocks are removed
+        for (String index : List.of("test-1", "test-2", "test-3")) {
+            final var indexMetadata = clusterService.state().metadata().getProject().indices().get(index);
+            assertThat(INDEX_BLOCKS_WRITE_SETTING.get(indexMetadata.getSettings()), is(false));
+        }
+
+        clusterService.removeListener(assertingListener);
+    }
+
     private static CheckedRunnable<Exception> blockMasterService(MasterService masterService) {
         final var executionBarrier = new CyclicBarrier(2);
         masterService.createTaskQueue("block", Priority.URGENT, batchExecutionContext -> {
@@ -237,6 +290,18 @@ public class MetadataIndexStateServiceBatchingTests extends ESSingleNodeTestCase
         );
     }
 
+    private static ClusterStateListener unblockedIndexCountListener() {
+        return event -> assertThat(
+            event.state()
+                .metadata()
+                .getProject()
+                .stream()
+                .filter(indexMetadata -> INDEX_BLOCKS_WRITE_SETTING.get(indexMetadata.getSettings()))
+                .count(),
+            oneOf(0L, 1L, 2L, 3L)  // Allow intermediate states during batched processing
+        );
+    }
+
     private static List<PendingClusterTask> findPendingTasks(MasterService masterService, String taskSourcePrefix) {
         return masterService.pendingTasks().stream().filter(task -> task.getSource().string().startsWith(taskSourcePrefix)).toList();
     }

server/src/test/java/org/elasticsearch/cluster/metadata/MetadataIndexStateServiceTests.java

@@ -11,6 +11,7 @@ package org.elasticsearch.cluster.metadata;
 
 import org.elasticsearch.action.admin.indices.close.CloseIndexResponse;
 import org.elasticsearch.action.admin.indices.close.CloseIndexResponse.IndexResult;
+import org.elasticsearch.action.admin.indices.readonly.RemoveIndexBlockResponse;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.RestoreInProgress;
@@ -585,4 +586,111 @@ public class MetadataIndexStateServiceTests extends ESTestCase {
         }
         return ClusterState.builder(state).putProjectMetadata(ProjectMetadata.builder(projectId)).build();
     }
+
+    public void testRemoveWriteBlockRemovesVerifiedSetting() {
+        ClusterState state = stateWithProject("testRemoveWriteBlockRemovesVerifiedSetting", projectId);
+
+        // Add an index with write block and VERIFIED_READ_ONLY_SETTING
+        final String indexName = "test-index";
+        final Settings.Builder indexSettings = indexSettings(IndexVersion.current(), 1, 0).put(
+            IndexMetadata.APIBlock.WRITE.settingName(),
+            true
+        ).put(MetadataIndexStateService.VERIFIED_READ_ONLY_SETTING.getKey(), true);
+
+        final IndexMetadata indexMetadata = IndexMetadata.builder(indexName)
+            .state(IndexMetadata.State.OPEN)
+            .creationDate(randomNonNegativeLong())
+            .settings(indexSettings)
+            .build();
+
+        state = ClusterState.builder(state)
+            .putProjectMetadata(ProjectMetadata.builder(state.metadata().getProject(projectId)).put(indexMetadata, true))
+            .blocks(ClusterBlocks.builder().blocks(state.blocks()).addIndexBlock(projectId, indexName, IndexMetadata.APIBlock.WRITE.block))
+            .build();
+
+        // Remove the write block
+        final Index index = state.metadata().getProject(projectId).index(indexName).getIndex();
+        final Tuple<ClusterState, List<RemoveIndexBlockResponse.RemoveBlockResult>> result = MetadataIndexStateService.removeIndexBlock(
+            state.projectState(projectId),
+            new Index[] { index },
+            IndexMetadata.APIBlock.WRITE
+        );
+
+        final ClusterState updatedState = result.v1();
+
+        // Verify that both write block setting and VERIFIED_READ_ONLY_SETTING are removed
+        final IndexMetadata updatedIndexMetadata = updatedState.metadata().getProject(projectId).index(indexName);
+        assertThat(
+            "Write block setting should be removed",
+            IndexMetadata.APIBlock.WRITE.setting().get(updatedIndexMetadata.getSettings()),
+            is(false)
+        );
+        assertThat(
+            "VERIFIED_READ_ONLY_SETTING should be removed",
+            MetadataIndexStateService.VERIFIED_READ_ONLY_SETTING.get(updatedIndexMetadata.getSettings()),
+            is(false)
+        );
+        assertThat(
+            "Write block should be removed from cluster state",
+            updatedState.blocks().hasIndexBlock(projectId, indexName, IndexMetadata.APIBlock.WRITE.block),
+            is(false)
+        );
+    }
+
+    public void testRemoveWriteBlockKeepsVerifiedWhenOtherBlocks() {
+        ClusterState state = stateWithProject("testRemoveWriteBlockKeepsVerifiedWhenOtherBlocks", projectId);
+
+        // Add an index with multiple write blocks and VERIFIED_READ_ONLY_SETTING
+        final String indexName = "test-index";
+        final Settings.Builder indexSettings = indexSettings(IndexVersion.current(), 1, 0).put(
+            IndexMetadata.APIBlock.WRITE.settingName(),
+            true
+        )
+            .put(IndexMetadata.APIBlock.READ_ONLY.settingName(), true)  // read_only also blocks writes
+            .put(MetadataIndexStateService.VERIFIED_READ_ONLY_SETTING.getKey(), true);
+
+        final IndexMetadata indexMetadata = IndexMetadata.builder(indexName)
+            .state(IndexMetadata.State.OPEN)
+            .creationDate(randomNonNegativeLong())
+            .settings(indexSettings)
+            .build();
+
+        state = ClusterState.builder(state)
+            .putProjectMetadata(ProjectMetadata.builder(state.metadata().getProject(projectId)).put(indexMetadata, true))
+            .blocks(
+                ClusterBlocks.builder()
+                    .blocks(state.blocks())
+                    .addIndexBlock(projectId, indexName, IndexMetadata.APIBlock.WRITE.block)
+                    .addIndexBlock(projectId, indexName, IndexMetadata.APIBlock.READ_ONLY.block)
+            )
+            .build();
+
+        // Remove only the write block (read_only block still exists and blocks writes)
+        final Index index = state.metadata().getProject(projectId).index(indexName).getIndex();
+        final Tuple<ClusterState, List<RemoveIndexBlockResponse.RemoveBlockResult>> result = MetadataIndexStateService.removeIndexBlock(
+            state.projectState(projectId),
+            new Index[] { index },
+            IndexMetadata.APIBlock.WRITE
+        );
+
+        final ClusterState updatedState = result.v1();
+
+        // Verify that VERIFIED_READ_ONLY_SETTING is kept because read_only block still blocks writes
+        final IndexMetadata updatedIndexMetadata = updatedState.metadata().getProject(projectId).index(indexName);
+        assertThat(
+            "Write block setting should be removed",
+            IndexMetadata.APIBlock.WRITE.setting().get(updatedIndexMetadata.getSettings()),
+            is(false)
+        );
+        assertThat(
+            "VERIFIED_READ_ONLY_SETTING should be kept because read_only block still exists",
+            MetadataIndexStateService.VERIFIED_READ_ONLY_SETTING.get(updatedIndexMetadata.getSettings()),
+            is(true)
+        );
+        assertThat(
+            "Read-only block should still exist",
+            updatedState.blocks().hasIndexBlock(projectId, indexName, IndexMetadata.APIBlock.READ_ONLY.block),
+            is(true)
+        );
+    }
 }

x-pack/plugin/security/qa/operator-privileges-tests/src/javaRestTest/java/org/elasticsearch/xpack/security/operator/Constants.java

@@ -495,6 +495,7 @@ public class Constants {
         "indices:admin/auto_create",
         "indices:admin/block/add",
         "indices:admin/block/add[s]",
+        "indices:admin/block/remove",
         "indices:admin/cache/clear",
         "indices:admin/data_stream/lazy_rollover",
         "indices:internal/admin/ccr/restore/file_chunk/get",