Skip to content

Commit 556ae8f

Browse files
martijnvgmartijnvg
committed
[CCR] Add validation checks that were left out of #30120 (#30463)
1 parent 00fc85f commit 556ae8f

File tree

5 files changed

+130
-44
lines changed

5 files changed

+130
-44
lines changed

‎x-pack/plugin/ccr/qa/multi-cluster-with-security/src/test/java/org/elasticsearch/xpack/ccr/FollowIndexSecurityIT.java

+9
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,11 @@ public void testFollowIndex() throws Exception {
6464
final String indexName2 = "index2";
6565
if (runningAgainstLeaderCluster) {
6666
logger.info("Running against leader cluster");
67+
Settings indexSettings = Settings.builder()
68+
.put("index.soft_deletes.enabled", true)
69+
.build();
70+
createIndex(indexName1, indexSettings);
71+
createIndex(indexName2, indexSettings);
6772
for (int i = 0; i < numDocs; i++) {
6873
logger.info("Indexing doc [{}]", i);
6974
index(indexName1, Integer.toString(i), "field", i);
@@ -169,6 +174,10 @@ private static Map<String, Object> toMap(String response) {
169174
return XContentHelper.convertToMap(JsonXContent.jsonXContent, response, false);
170175
}
171176

177+
protected static void createIndex(String name, Settings settings) throws IOException {
178+
createIndex(name, settings, "");
179+
}
180+
172181
protected static void createIndex(String name, Settings settings, String mapping) throws IOException {
173182
assertOK(adminClient().performRequest(HttpPut.METHOD_NAME, name, Collections.emptyMap(),
174183
new StringEntity("{ \"settings\": " + Strings.toString(settings)

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

+48-43
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;
@@ -224,29 +225,13 @@ protected void doExecute(Request request, ActionListener<Response> listener) {
224225
*/
225226
void start(Request request, String clusterNameAlias, IndexMetaData leaderIndexMetadata, IndexMetaData followIndexMetadata,
226227
ActionListener<Response> handler) {
227-
if (leaderIndexMetadata == null) {
228-
handler.onFailure(new IllegalArgumentException("leader index [" + request.leaderIndex + "] does not exist"));
229-
return;
230-
}
231-
232-
if (followIndexMetadata == null) {
233-
handler.onFailure(new IllegalArgumentException("follow index [" + request.followIndex + "] does not exist"));
234-
return;
235-
}
236-
237-
if (leaderIndexMetadata.getNumberOfShards() != followIndexMetadata.getNumberOfShards()) {
238-
handler.onFailure(new IllegalArgumentException("leader index primary shards [" +
239-
leaderIndexMetadata.getNumberOfShards() + "] does not match with the number of " +
240-
"shards of the follow index [" + followIndexMetadata.getNumberOfShards() + "]"));
241-
// TODO: other validation checks
242-
} else {
228+
validate (leaderIndexMetadata ,followIndexMetadata , request);
243229
final int numShards = followIndexMetadata.getNumberOfShards();
244230
final AtomicInteger counter = new AtomicInteger(numShards);
245231
final AtomicReferenceArray<Object> responses = new AtomicReferenceArray<>(followIndexMetadata.getNumberOfShards());
246232
Map<String, String> filteredHeaders = threadPool.getThreadContext().getHeaders().entrySet().stream()
247233
.filter(e -> ShardFollowTask.HEADER_FILTERS.contains(e.getKey()))
248-
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
249-
for (int i = 0; i < numShards; i++) {
234+
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));for (int i = 0; i < numShards; i++) {
250235
final int shardId = i;
251236
String taskId = followIndexMetadata.getIndexUUID() + "-" + shardId;
252237
ShardFollowTask shardFollowTask = new ShardFollowTask(clusterNameAlias,
@@ -261,39 +246,59 @@ public void onResponse(PersistentTasksCustomMetaData.PersistentTask<ShardFollowT
261246
finalizeResponse();
262247
}
263248

264-
@Override
265-
public void onFailure(Exception e) {
266-
responses.set(shardId, e);
267-
finalizeResponse();
268-
}
249+
@Override
250+
public void onFailure(Exception e) {
251+
responses.set(shardId, e);
252+
finalizeResponse();
253+
}
269254

270-
void finalizeResponse() {
271-
Exception error = null;
272-
if (counter.decrementAndGet() == 0) {
273-
for (int j = 0; j < responses.length(); j++) {
274-
Object response = responses.get(j);
275-
if (response instanceof Exception) {
276-
if (error == null) {
277-
error = (Exception) response;
278-
} else {
279-
error.addSuppressed((Throwable) response);
280-
}
255+
void finalizeResponse() {
256+
Exception error = null;
257+
if (counter.decrementAndGet() == 0) {
258+
for (int j = 0; j < responses.length(); j++) {
259+
Object response = responses.get(j);
260+
if (response instanceof Exception) {
261+
if (error == null) {
262+
error = (Exception) response;
263+
} else {
264+
error.addSuppressed((Throwable) response);
281265
}
282266
}
267+
}
283268

284-
if (error == null) {
285-
// include task ids?
286-
handler.onResponse(new Response(true));
287-
} else {
288-
// TODO: cancel all started tasks
289-
handler.onFailure(error);
290-
}
269+
if (error == null) {
270+
// include task ids?
271+
handler.onResponse(new Response(true));
272+
} else {
273+
// TODO: cancel all started tasks
274+
handler.onFailure(error);
291275
}
292276
}
293277
}
294-
);
295-
}
278+
}
279+
);
296280
}
297281
}
298282
}
283+
284+
285+
static void validate(IndexMetaData leaderIndex, IndexMetaData followIndex, Request request) {
286+
if (leaderIndex == null) {
287+
throw new IllegalArgumentException("leader index [" + request.leaderIndex + "] does not exist");
288+
}
289+
290+
if (followIndex == null) {
291+
throw new IllegalArgumentException("follow index [" + request.followIndex + "] does not exist");
292+
}
293+
if (leaderIndex.getSettings().getAsBoolean(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), false) == false) {
294+
throw new IllegalArgumentException("leader index [" + request.leaderIndex + "] does not have soft deletes enabled");
295+
}
296+
297+
if (leaderIndex.getNumberOfShards() != followIndex.getNumberOfShards()) {
298+
throw new IllegalArgumentException("leader index primary shards [" + leaderIndex.getNumberOfShards() +
299+
"] does not match with the number of shards of the follow index [" + followIndex.getNumberOfShards() + "]");
300+
}
301+
// TODO: other validation checks
302+
}
303+
299304
}

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

+3-1
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;
@@ -142,7 +143,8 @@ public void testGetOperationsBasedOnGlobalSequenceId() throws Exception {
142143
public void testFollowIndex() throws Exception {
143144
final int numberOfPrimaryShards = randomIntBetween(1, 3);
144145

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

148150
final String followerIndexSettings =

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

+66
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.collect.Tuple;
11+
import org.elasticsearch.common.settings.Settings;
12+
import org.elasticsearch.index.IndexSettings;
13+
import org.elasticsearch.test.ESTestCase;
14+
15+
import static org.hamcrest.Matchers.equalTo;
16+
17+
public class FollowExistingIndexActionTests extends ESTestCase {
18+
19+
public void testValidation() {
20+
FollowExistingIndexAction.Request request = new FollowExistingIndexAction.Request();
21+
request.setLeaderIndex("index1");
22+
request.setFollowIndex("index2");
23+
24+
{
25+
Exception e = expectThrows(IllegalArgumentException.class, () -> FollowExistingIndexAction.validate(null, null, request));
26+
assertThat(e.getMessage(), equalTo("leader index [index1] does not exist"));
27+
}
28+
{
29+
IndexMetaData leaderIMD = createIMD("index1", 5);
30+
Exception e = expectThrows(IllegalArgumentException.class, () -> FollowExistingIndexAction.validate(leaderIMD, null, request));
31+
assertThat(e.getMessage(), equalTo("follow index [index2] does not exist"));
32+
}
33+
{
34+
IndexMetaData leaderIMD = createIMD("index1", 5);
35+
IndexMetaData followIMD = createIMD("index2", 5);
36+
Exception e = expectThrows(IllegalArgumentException.class,
37+
() -> FollowExistingIndexAction.validate(leaderIMD, followIMD, request));
38+
assertThat(e.getMessage(), equalTo("leader index [index1] does not have soft deletes enabled"));
39+
}
40+
{
41+
IndexMetaData leaderIMD = createIMD("index1", 5, new Tuple<>(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true"));
42+
IndexMetaData followIMD = createIMD("index2", 4);
43+
Exception e = expectThrows(IllegalArgumentException.class,
44+
() -> FollowExistingIndexAction.validate(leaderIMD, followIMD, request));
45+
assertThat(e.getMessage(),
46+
equalTo("leader index primary shards [5] does not match with the number of shards of the follow index [4]"));
47+
}
48+
{
49+
IndexMetaData leaderIMD = createIMD("index1", 5, new Tuple<>(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true"));
50+
IndexMetaData followIMD = createIMD("index2", 5);
51+
FollowExistingIndexAction.validate(leaderIMD, followIMD, request);
52+
}
53+
}
54+
55+
private static IndexMetaData createIMD(String index, int numShards, Tuple<?, ?>... settings) {
56+
Settings.Builder settingsBuilder = settings(Version.CURRENT);
57+
for (Tuple<?, ?> setting : settings) {
58+
settingsBuilder.put((String) setting.v1(), (String) setting.v2());
59+
}
60+
return IndexMetaData.builder(index).settings(settingsBuilder)
61+
.numberOfShards(numShards)
62+
.numberOfReplicas(0)
63+
.setRoutingNumShards(numShards).build();
64+
}
65+
66+
}

‎x-pack/plugin/src/test/resources/rest-api-spec/test/ccr/follow_and_unfollow.yml

+4
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,10 @@
44
indices.create:
55
index: foo
66
body:
7+
settings:
8+
index:
9+
soft_deletes:
10+
enabled: true
711
mappings:
812
doc:
913
properties:

0 commit comments

Comments
 (0)