Skip to content

Commit 80f5ccb

Browse files
martijnvgmartijnvg
committed
[CCR] Added validation checks that were left out of #30120 
1 parent a5be414 commit 80f5ccb

File tree

3 files changed

+133
-59
lines changed

3 files changed

+133
-59
lines changed

‎x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/FollowExistingIndexAction.java‎

Lines changed: 64 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.elasticsearch.common.io.stream.StreamInput;
2525
import org.elasticsearch.common.io.stream.StreamOutput;
2626
import org.elasticsearch.common.settings.Settings;
27+
import org.elasticsearch.index.IndexSettings;
2728
import org.elasticsearch.index.shard.ShardId;
2829
import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
2930
import org.elasticsearch.persistent.PersistentTasksService;
@@ -223,73 +224,78 @@ protected void doExecute(Request request, ActionListener<Response> listener) {
223224
*/
224225
void start(Request request, String clusterNameAlias, IndexMetaData leaderIndexMetadata, IndexMetaData followIndexMetadata,
225226
ActionListener<Response> handler) {
226-
if (leaderIndexMetadata == null) {
227-
handler.onFailure(new IllegalArgumentException("leader index [" + request.leaderIndex + "] does not exist"));
228-
return;
229-
}
230-
231-
if (followIndexMetadata == null) {
232-
handler.onFailure(new IllegalArgumentException("follow index [" + request.followIndex + "] does not exist"));
233-
return;
234-
}
235-
236-
if (leaderIndexMetadata.getNumberOfShards() != followIndexMetadata.getNumberOfShards()) {
237-
handler.onFailure(new IllegalArgumentException("leader index primary shards [" +
238-
leaderIndexMetadata.getNumberOfShards() + "] does not match with the number of " +
239-
"shards of the follow index [" + followIndexMetadata.getNumberOfShards() + "]"));
240-
// TODO: other validation checks
241-
} else {
242-
final int numShards = followIndexMetadata.getNumberOfShards();
243-
final AtomicInteger counter = new AtomicInteger(numShards);
244-
final AtomicReferenceArray<Object> responses = new AtomicReferenceArray<>(followIndexMetadata.getNumberOfShards());
245-
for (int i = 0; i < numShards; i++) {
246-
final int shardId = i;
247-
String taskId = followIndexMetadata.getIndexUUID() + "-" + shardId;
248-
ShardFollowTask shardFollowTask = new ShardFollowTask(clusterNameAlias,
249-
new ShardId(followIndexMetadata.getIndex(), shardId),
250-
new ShardId(leaderIndexMetadata.getIndex(), shardId),
251-
request.batchSize, request.concurrentProcessors, request.processorMaxTranslogBytes);
252-
persistentTasksService.startPersistentTask(taskId, ShardFollowTask.NAME, shardFollowTask,
253-
new ActionListener<PersistentTasksCustomMetaData.PersistentTask<ShardFollowTask>>() {
254-
@Override
255-
public void onResponse(PersistentTasksCustomMetaData.PersistentTask<ShardFollowTask> task) {
256-
responses.set(shardId, task);
257-
finalizeResponse();
258-
}
227+
validate(leaderIndexMetadata, followIndexMetadata, request);
228+
final int numShards = followIndexMetadata.getNumberOfShards();
229+
final AtomicInteger counter = new AtomicInteger(numShards);
230+
final AtomicReferenceArray<Object> responses = new AtomicReferenceArray<>(followIndexMetadata.getNumberOfShards());
231+
for (int i = 0; i < numShards; i++) {
232+
final int shardId = i;
233+
String taskId = followIndexMetadata.getIndexUUID() + "-" + shardId;
234+
ShardFollowTask shardFollowTask = new ShardFollowTask(clusterNameAlias,
235+
new ShardId(followIndexMetadata.getIndex(), shardId),
236+
new ShardId(leaderIndexMetadata.getIndex(), shardId),
237+
request.batchSize, request.concurrentProcessors, request.processorMaxTranslogBytes);
238+
persistentTasksService.startPersistentTask(taskId, ShardFollowTask.NAME, shardFollowTask,
239+
new ActionListener<PersistentTasksCustomMetaData.PersistentTask<ShardFollowTask>>() {
240+
@Override
241+
public void onResponse(PersistentTasksCustomMetaData.PersistentTask<ShardFollowTask> task) {
242+
responses.set(shardId, task);
243+
finalizeResponse();
244+
}
259245

260-
@Override
261-
public void onFailure(Exception e) {
262-
responses.set(shardId, e);
263-
finalizeResponse();
264-
}
246+
@Override
247+
public void onFailure(Exception e) {
248+
responses.set(shardId, e);
249+
finalizeResponse();
250+
}
265251

266-
void finalizeResponse() {
267-
Exception error = null;
268-
if (counter.decrementAndGet() == 0) {
269-
for (int j = 0; j < responses.length(); j++) {
270-
Object response = responses.get(j);
271-
if (response instanceof Exception) {
272-
if (error == null) {
273-
error = (Exception) response;
274-
} else {
275-
error.addSuppressed((Throwable) response);
276-
}
252+
void finalizeResponse() {
253+
Exception error = null;
254+
if (counter.decrementAndGet() == 0) {
255+
for (int j = 0; j < responses.length(); j++) {
256+
Object response = responses.get(j);
257+
if (response instanceof Exception) {
258+
if (error == null) {
259+
error = (Exception) response;
260+
} else {
261+
error.addSuppressed((Throwable) response);
277262
}
278263
}
264+
}
279265

280-
if (error == null) {
281-
// include task ids?
282-
handler.onResponse(new Response(true));
283-
} else {
284-
// TODO: cancel all started tasks
285-
handler.onFailure(error);
286-
}
266+
if (error == null) {
267+
// include task ids?
268+
handler.onResponse(new Response(true));
269+
} else {
270+
// TODO: cancel all started tasks
271+
handler.onFailure(error);
287272
}
288273
}
289274
}
290-
);
291-
}
275+
}
276+
);
292277
}
293278
}
294279
}
280+
281+
282+
static void validate(IndexMetaData leaderIndex, IndexMetaData followIndex, Request request) {
283+
if (leaderIndex == null) {
284+
throw new IllegalArgumentException("leader index [" + request.leaderIndex + "] does not exist");
285+
}
286+
287+
if (followIndex == null) {
288+
throw new IllegalArgumentException("follow index [" + request.followIndex + "] does not exist");
289+
}
290+
if (leaderIndex.getSettings().getAsBoolean(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), false) == false) {
291+
throw new IllegalArgumentException("leader index [" + request.leaderIndex + "] does not have soft deletes enabled");
292+
}
293+
294+
if (leaderIndex.getNumberOfShards() != followIndex.getNumberOfShards()) {
295+
throw new IllegalArgumentException("leader index primary shards [" + leaderIndex.getNumberOfShards() +
296+
"] does not match with the number of shards of the follow index [" + followIndex.getNumberOfShards() + "]");
297+
}
298+
// TODO: other validation checks
299+
}
300+
295301
}

‎x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/ShardChangesIT.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import org.elasticsearch.common.xcontent.XContentType;
2020
import org.elasticsearch.common.xcontent.support.XContentMapValues;
2121
import org.elasticsearch.index.IndexSettings;
22+
import org.elasticsearch.index.IndexSettings;
2223
import org.elasticsearch.index.shard.ShardId;
2324
import org.elasticsearch.index.translog.Translog;
2425
import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
@@ -143,7 +144,8 @@ public void testGetOperationsBasedOnGlobalSequenceId() throws Exception {
143144
public void testFollowIndex() throws Exception {
144145
final int numberOfPrimaryShards = randomIntBetween(1, 3);
145146

146-
final String leaderIndexSettings = getIndexSettings(numberOfPrimaryShards, Collections.emptyMap());
147+
final String leaderIndexSettings = getIndexSettings(numberOfPrimaryShards,
148+
Collections.singletonMap(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true"));
147149
assertAcked(client().admin().indices().prepareCreate("index1").setSource(leaderIndexSettings, XContentType.JSON));
148150

149151
final String followerIndexSettings =

‎x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/FollowExistingIndexActionTests.java‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the Elastic License;
4+
* you may not use this file except in compliance with the Elastic License.
5+
*/
6+
package org.elasticsearch.xpack.ccr.action;
7+
8+
import org.elasticsearch.Version;
9+
import org.elasticsearch.cluster.metadata.IndexMetaData;
10+
import org.elasticsearch.common.settings.Settings;
11+
import org.elasticsearch.index.IndexSettings;
12+
import org.elasticsearch.test.ESTestCase;
13+
14+
import static org.hamcrest.Matchers.equalTo;
15+
16+
public class FollowExistingIndexActionTests extends ESTestCase {
17+
18+
public void testValidation() {
19+
FollowExistingIndexAction.Request request = new FollowExistingIndexAction.Request();
20+
request.setLeaderIndex("index1");
21+
request.setFollowIndex("index2");
22+
23+
{
24+
Exception e = expectThrows(IllegalArgumentException.class, () -> FollowExistingIndexAction.validate(null, null, request));
25+
assertThat(e.getMessage(), equalTo("leader index [index1] does not exist"));
26+
}
27+
{
28+
IndexMetaData leaderIMD = createIMD("index1", 5);
29+
Exception e = expectThrows(IllegalArgumentException.class, () -> FollowExistingIndexAction.validate(leaderIMD, null, request));
30+
assertThat(e.getMessage(), equalTo("follow index [index2] does not exist"));
31+
}
32+
{
33+
IndexMetaData leaderIMD = createIMD("index1", 5);
34+
IndexMetaData followIMD = createIMD("index2", 5);
35+
Exception e = expectThrows(IllegalArgumentException.class,
36+
() -> FollowExistingIndexAction.validate(leaderIMD, followIMD, request));
37+
assertThat(e.getMessage(), equalTo("leader index [index1] does not have soft deletes enabled"));
38+
}
39+
{
40+
IndexMetaData leaderIMD = createIMD("index1", 5, IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true");
41+
IndexMetaData followIMD = createIMD("index2", 4);
42+
Exception e = expectThrows(IllegalArgumentException.class,
43+
() -> FollowExistingIndexAction.validate(leaderIMD, followIMD, request));
44+
assertThat(e.getMessage(),
45+
equalTo("leader index primary shards [5] does not match with the number of shards of the follow index [4]"));
46+
}
47+
{
48+
IndexMetaData leaderIMD = createIMD("index1", 5, IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true");
49+
IndexMetaData followIMD = createIMD("index2", 5);
50+
FollowExistingIndexAction.validate(leaderIMD, followIMD, request);
51+
}
52+
}
53+
54+
private static IndexMetaData createIMD(String index, int numShards, String... settings) {
55+
assert settings.length % 2 == 0;
56+
Settings.Builder settingsBuilder = settings(Version.CURRENT);
57+
for (int i = 0; i < settings.length; i += 2) {
58+
settingsBuilder.put(settings[i], settings[i + 1]);
59+
}
60+
return IndexMetaData.builder(index).settings(settingsBuilder)
61+
.numberOfShards(numShards)
62+
.numberOfReplicas(0)
63+
.setRoutingNumShards(numShards).build();
64+
}
65+
66+
}

0 commit comments

Comments
 (0)