Skip to content

Commit e149b08

Browse files
tlrxtlrx
authored
[Close Index API] Add unique UUID to ClusterBlock (#36775)
This commit adds a unique id to cluster blocks, so that they can be uniquely identified if needed. This is important for the Close Index API where multiple concurrent closing requests can be executed at the same time. By adding a UUID to the cluster block, we can generate unique "closing block" that can later be verified on shards and then checked again from the cluster state before closing the index. When the verification on shard is done, the closing block is replaced by the regular INDEX_CLOSED_BLOCK instance. If something goes wrong, calling the Open Index API will remove the block. Related to #33888
1 parent f5af79b commit e149b08

16 files changed

Lines changed: 773 additions & 250 deletions

File tree

‎distribution/archives/integ-test-zip/src/test/java/org/elasticsearch/test/rest/WaitForRefreshAndCloseIT.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@
2020
package org.elasticsearch.test.rest;
2121

2222
import org.apache.http.util.EntityUtils;
23-
import org.apache.lucene.util.LuceneTestCase;
2423
import org.elasticsearch.action.ActionFuture;
2524
import org.elasticsearch.action.support.PlainActionFuture;
2625
import org.elasticsearch.client.Request;
@@ -41,7 +40,6 @@
4140
/**
4241
* Tests that wait for refresh is fired if the index is closed.
4342
*/
44-
@LuceneTestCase.AwaitsFix(bugUrl = "to be created")
4543
public class WaitForRefreshAndCloseIT extends ESRestTestCase {
4644
@Before
4745
public void setupIndex() throws IOException {

‎server/src/main/java/org/elasticsearch/action/admin/indices/close/TransportVerifyShardBeforeCloseAction.java‎

Lines changed: 30 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -29,9 +29,10 @@
2929
import org.elasticsearch.cluster.block.ClusterBlock;
3030
import org.elasticsearch.cluster.block.ClusterBlocks;
3131
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
32-
import org.elasticsearch.cluster.metadata.MetaDataIndexStateService;
3332
import org.elasticsearch.cluster.service.ClusterService;
3433
import org.elasticsearch.common.inject.Inject;
34+
import org.elasticsearch.common.io.stream.StreamInput;
35+
import org.elasticsearch.common.io.stream.StreamOutput;
3536
import org.elasticsearch.common.lease.Releasable;
3637
import org.elasticsearch.common.settings.Settings;
3738
import org.elasticsearch.index.shard.IndexShard;
@@ -41,13 +42,14 @@
4142
import org.elasticsearch.threadpool.ThreadPool;
4243
import org.elasticsearch.transport.TransportService;
4344

45+
import java.io.IOException;
46+
import java.util.Objects;
4447
import java.util.function.Consumer;
4548

4649
public class TransportVerifyShardBeforeCloseAction extends TransportReplicationAction<
4750
TransportVerifyShardBeforeCloseAction.ShardRequest, TransportVerifyShardBeforeCloseAction.ShardRequest, ReplicationResponse> {
4851

4952
public static final String NAME = CloseIndexAction.NAME + "[s]";
50-
public static final ClusterBlock EXPECTED_BLOCK = MetaDataIndexStateService.INDEX_CLOSED_BLOCK;
5153

5254
@Inject
5355
public TransportVerifyShardBeforeCloseAction(final Settings settings, final TransportService transportService,
@@ -83,25 +85,25 @@ protected void acquireReplicaOperationPermit(final IndexShard replica,
8385
@Override
8486
protected PrimaryResult<ShardRequest, ReplicationResponse> shardOperationOnPrimary(final ShardRequest shardRequest,
8587
final IndexShard primary) throws Exception {
86-
executeShardOperation(primary);
88+
executeShardOperation(shardRequest, primary);
8789
return new PrimaryResult<>(shardRequest, new ReplicationResponse());
8890
}
8991

9092
@Override
9193
protected ReplicaResult shardOperationOnReplica(final ShardRequest shardRequest, final IndexShard replica) throws Exception {
92-
executeShardOperation(replica);
94+
executeShardOperation(shardRequest, replica);
9395
return new ReplicaResult();
9496
}
9597

96-
private void executeShardOperation(final IndexShard indexShard) {
98+
private void executeShardOperation(final ShardRequest request, final IndexShard indexShard) {
9799
final ShardId shardId = indexShard.shardId();
98100
if (indexShard.getActiveOperationsCount() != 0) {
99101
throw new IllegalStateException("On-going operations in progress while checking index shard " + shardId + " before closing");
100102
}
101103

102104
final ClusterBlocks clusterBlocks = clusterService.state().blocks();
103-
if (clusterBlocks.hasIndexBlock(shardId.getIndexName(), EXPECTED_BLOCK) == false) {
104-
throw new IllegalStateException("Index shard " + shardId + " must be blocked by " + EXPECTED_BLOCK + " before closing");
105+
if (clusterBlocks.hasIndexBlock(shardId.getIndexName(), request.clusterBlock()) == false) {
106+
throw new IllegalStateException("Index shard " + shardId + " must be blocked by " + request.clusterBlock() + " before closing");
105107
}
106108

107109
final long maxSeqNo = indexShard.seqNoStats().getMaxSeqNo();
@@ -139,17 +141,36 @@ public void markShardCopyAsStaleIfNeeded(final ShardId shardId, final String all
139141

140142
public static class ShardRequest extends ReplicationRequest<ShardRequest> {
141143

144+
private ClusterBlock clusterBlock;
145+
142146
ShardRequest(){
143147
}
144148

145-
public ShardRequest(final ShardId shardId, final TaskId parentTaskId) {
149+
public ShardRequest(final ShardId shardId, final ClusterBlock clusterBlock, final TaskId parentTaskId) {
146150
super(shardId);
151+
this.clusterBlock = Objects.requireNonNull(clusterBlock);
147152
setParentTask(parentTaskId);
148153
}
149154

150155
@Override
151156
public String toString() {
152-
return "verify shard before close {" + shardId + "}";
157+
return "verify shard " + shardId + " before close with block " + clusterBlock;
158+
}
159+
160+
@Override
161+
public void readFrom(final StreamInput in) throws IOException {
162+
super.readFrom(in);
163+
clusterBlock = ClusterBlock.readClusterBlock(in);
164+
}
165+
166+
@Override
167+
public void writeTo(final StreamOutput out) throws IOException {
168+
super.writeTo(out);
169+
clusterBlock.writeTo(out);
170+
}
171+
172+
public ClusterBlock clusterBlock() {
173+
return clusterBlock;
153174
}
154175
}
155176
}

‎server/src/main/java/org/elasticsearch/cluster/block/ClusterBlock.java‎

Lines changed: 42 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@
1919

2020
package org.elasticsearch.cluster.block;
2121

22+
import org.elasticsearch.Version;
23+
import org.elasticsearch.common.Nullable;
2224
import org.elasticsearch.common.io.stream.StreamInput;
2325
import org.elasticsearch.common.io.stream.StreamOutput;
2426
import org.elasticsearch.common.io.stream.Streamable;
@@ -30,29 +32,31 @@
3032
import java.util.ArrayList;
3133
import java.util.EnumSet;
3234
import java.util.Locale;
35+
import java.util.Objects;
3336

3437
public class ClusterBlock implements Streamable, ToXContentFragment {
3538

3639
private int id;
37-
40+
private @Nullable String uuid;
3841
private String description;
39-
4042
private EnumSet<ClusterBlockLevel> levels;
41-
4243
private boolean retryable;
43-
4444
private boolean disableStatePersistence = false;
45-
4645
private boolean allowReleaseResources;
47-
4846
private RestStatus status;
4947

50-
ClusterBlock() {
48+
private ClusterBlock() {
49+
}
50+
51+
public ClusterBlock(int id, String description, boolean retryable, boolean disableStatePersistence,
52+
boolean allowReleaseResources, RestStatus status, EnumSet<ClusterBlockLevel> levels) {
53+
this(id, null, description, retryable, disableStatePersistence, allowReleaseResources, status, levels);
5154
}
5255

53-
public ClusterBlock(int id, String description, boolean retryable, boolean disableStatePersistence, boolean allowReleaseResources,
54-
RestStatus status, EnumSet<ClusterBlockLevel> levels) {
56+
public ClusterBlock(int id, String uuid, String description, boolean retryable, boolean disableStatePersistence,
57+
boolean allowReleaseResources, RestStatus status, EnumSet<ClusterBlockLevel> levels) {
5558
this.id = id;
59+
this.uuid = uuid;
5660
this.description = description;
5761
this.retryable = retryable;
5862
this.disableStatePersistence = disableStatePersistence;
@@ -65,6 +69,10 @@ public int id() {
6569
return this.id;
6670
}
6771

72+
public String uuid() {
73+
return uuid;
74+
}
75+
6876
public String description() {
6977
return this.description;
7078
}
@@ -104,6 +112,9 @@ public boolean disableStatePersistence() {
104112
@Override
105113
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
106114
builder.startObject(Integer.toString(id));
115+
if (uuid != null) {
116+
builder.field("uuid", uuid);
117+
}
107118
builder.field("description", description);
108119
builder.field("retryable", retryable);
109120
if (disableStatePersistence) {
@@ -127,6 +138,11 @@ public static ClusterBlock readClusterBlock(StreamInput in) throws IOException {
127138
@Override
128139
public void readFrom(StreamInput in) throws IOException {
129140
id = in.readVInt();
141+
if (in.getVersion().onOrAfter(Version.V_7_0_0)) {
142+
uuid = in.readOptionalString();
143+
} else {
144+
uuid = null;
145+
}
130146
description = in.readString();
131147
final int len = in.readVInt();
132148
ArrayList<ClusterBlockLevel> levels = new ArrayList<>(len);
@@ -143,6 +159,9 @@ public void readFrom(StreamInput in) throws IOException {
143159
@Override
144160
public void writeTo(StreamOutput out) throws IOException {
145161
out.writeVInt(id);
162+
if (out.getVersion().onOrAfter(Version.V_7_0_0)) {
163+
out.writeOptionalString(uuid);
164+
}
146165
out.writeString(description);
147166
out.writeVInt(levels.size());
148167
for (ClusterBlockLevel level : levels) {
@@ -157,7 +176,11 @@ public void writeTo(StreamOutput out) throws IOException {
157176
@Override
158177
public String toString() {
159178
StringBuilder sb = new StringBuilder();
160-
sb.append(id).append(",").append(description).append(", blocks ");
179+
sb.append(id).append(",");
180+
if (uuid != null) {
181+
sb.append(uuid).append(',');
182+
}
183+
sb.append(description).append(", blocks ");
161184
String delimiter = "";
162185
for (ClusterBlockLevel level : levels) {
163186
sb.append(delimiter).append(level.name());
@@ -168,19 +191,19 @@ public String toString() {
168191

169192
@Override
170193
public boolean equals(Object o) {
171-
if (this == o) return true;
172-
if (o == null || getClass() != o.getClass()) return false;
173-
174-
ClusterBlock that = (ClusterBlock) o;
175-
176-
if (id != that.id) return false;
177-
178-
return true;
194+
if (this == o) {
195+
return true;
196+
}
197+
if (o == null || getClass() != o.getClass()) {
198+
return false;
199+
}
200+
final ClusterBlock that = (ClusterBlock) o;
201+
return id == that.id && Objects.equals(uuid, that.uuid);
179202
}
180203

181204
@Override
182205
public int hashCode() {
183-
return id;
206+
return Objects.hash(id, uuid);
184207
}
185208

186209
public boolean isAllowReleaseResources() {

‎server/src/main/java/org/elasticsearch/cluster/block/ClusterBlocks.java‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.elasticsearch.cluster.Diff;
2525
import org.elasticsearch.cluster.metadata.IndexMetaData;
2626
import org.elasticsearch.cluster.metadata.MetaDataIndexStateService;
27+
import org.elasticsearch.common.Nullable;
2728
import org.elasticsearch.common.collect.ImmutableOpenMap;
2829
import org.elasticsearch.common.io.stream.StreamInput;
2930
import org.elasticsearch.common.io.stream.StreamOutput;
@@ -147,6 +148,31 @@ public boolean hasIndexBlock(String index, ClusterBlock block) {
147148
return indicesBlocks.containsKey(index) && indicesBlocks.get(index).contains(block);
148149
}
149150

151+
public boolean hasIndexBlockWithId(String index, int blockId) {
152+
final Set<ClusterBlock> clusterBlocks = indicesBlocks.get(index);
153+
if (clusterBlocks != null) {
154+
for (ClusterBlock clusterBlock : clusterBlocks) {
155+
if (clusterBlock.id() == blockId) {
156+
return true;
157+
}
158+
}
159+
}
160+
return false;
161+
}
162+
163+
@Nullable
164+
public ClusterBlock getIndexBlockWithId(final String index, final int blockId) {
165+
final Set<ClusterBlock> clusterBlocks = indicesBlocks.get(index);
166+
if (clusterBlocks != null) {
167+
for (ClusterBlock clusterBlock : clusterBlocks) {
168+
if (clusterBlock.id() == blockId) {
169+
return clusterBlock;
170+
}
171+
}
172+
}
173+
return null;
174+
}
175+
150176
public void globalBlockedRaiseException(ClusterBlockLevel level) throws ClusterBlockException {
151177
ClusterBlockException blockException = globalBlockedException(level);
152178
if (blockException != null) {
@@ -403,6 +429,18 @@ public Builder removeIndexBlock(String index, ClusterBlock block) {
403429
return this;
404430
}
405431

432+
public Builder removeIndexBlockWithId(String index, int blockId) {
433+
final Set<ClusterBlock> indexBlocks = indices.get(index);
434+
if (indexBlocks == null) {
435+
return this;
436+
}
437+
indexBlocks.removeIf(block -> block.id() == blockId);
438+
if (indexBlocks.isEmpty()) {
439+
indices.remove(index);
440+
}
441+
return this;
442+
}
443+
406444
public ClusterBlocks build() {
407445
// We copy the block sets here in case of the builder is modified after build is called
408446
ImmutableOpenMap.Builder<String, Set<ClusterBlock>> indicesBuilder = ImmutableOpenMap.builder(indices.size());

0 commit comments

Comments
 (0)