From 556ae8f3ed0fa0519c55fecc737a872d4aed7e6b Mon Sep 17 00:00:00 2001
From: Martijn van Groningen <martijn.v.groningen@gmail.com>
Date: Wed, 16 May 2018 09:46:03 +0200
Subject: [PATCH] [CCR] Add validation checks that were left out of #30120
 (#30463)

---
 .../xpack/ccr/FollowIndexSecurityIT.java      |  9 ++
 .../ccr/action/FollowExistingIndexAction.java | 91 ++++++++++---------
 .../xpack/ccr/ShardChangesIT.java             |  4 +-
 .../FollowExistingIndexActionTests.java       | 66 ++++++++++++++
 .../test/ccr/follow_and_unfollow.yml          |  4 +
 5 files changed, 130 insertions(+), 44 deletions(-)
 create mode 100644 x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/FollowExistingIndexActionTests.java

diff --git a/x-pack/plugin/ccr/qa/multi-cluster-with-security/src/test/java/org/elasticsearch/xpack/ccr/FollowIndexSecurityIT.java b/x-pack/plugin/ccr/qa/multi-cluster-with-security/src/test/java/org/elasticsearch/xpack/ccr/FollowIndexSecurityIT.java
index 85648a039c2ab..2fcc0890f2bc0 100644
--- a/x-pack/plugin/ccr/qa/multi-cluster-with-security/src/test/java/org/elasticsearch/xpack/ccr/FollowIndexSecurityIT.java
+++ b/x-pack/plugin/ccr/qa/multi-cluster-with-security/src/test/java/org/elasticsearch/xpack/ccr/FollowIndexSecurityIT.java
@@ -64,6 +64,11 @@ public void testFollowIndex() throws Exception {
         final String indexName2 = "index2";
         if (runningAgainstLeaderCluster) {
             logger.info("Running against leader cluster");
+            Settings indexSettings = Settings.builder()
+                .put("index.soft_deletes.enabled", true)
+                .build();
+            createIndex(indexName1, indexSettings);
+            createIndex(indexName2, indexSettings);
             for (int i = 0; i < numDocs; i++) {
                 logger.info("Indexing doc [{}]", i);
                 index(indexName1, Integer.toString(i), "field", i);
@@ -169,6 +174,10 @@ private static Map<String, Object> toMap(String response) {
         return XContentHelper.convertToMap(JsonXContent.jsonXContent, response, false);
     }
 
+    protected static void createIndex(String name, Settings settings) throws IOException {
+        createIndex(name, settings, "");
+    }
+
     protected static void createIndex(String name, Settings settings, String mapping) throws IOException {
         assertOK(adminClient().performRequest(HttpPut.METHOD_NAME, name, Collections.emptyMap(),
             new StringEntity("{ \"settings\": " + Strings.toString(settings)
diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/FollowExistingIndexAction.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/FollowExistingIndexAction.java
index 963f17d975882..2ebf1fc52b424 100644
--- a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/FollowExistingIndexAction.java
+++ b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/FollowExistingIndexAction.java
@@ -24,6 +24,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
 import org.elasticsearch.persistent.PersistentTasksService;
@@ -224,29 +225,13 @@ protected void doExecute(Request request, ActionListener<Response> listener) {
          */
         void start(Request request, String clusterNameAlias, IndexMetaData leaderIndexMetadata, IndexMetaData followIndexMetadata,
                    ActionListener<Response> handler) {
-            if (leaderIndexMetadata == null) {
-                handler.onFailure(new IllegalArgumentException("leader index [" + request.leaderIndex + "] does not exist"));
-                return;
-            }
-
-            if (followIndexMetadata == null) {
-                handler.onFailure(new IllegalArgumentException("follow index [" + request.followIndex + "] does not exist"));
-                return;
-            }
-
-            if (leaderIndexMetadata.getNumberOfShards() != followIndexMetadata.getNumberOfShards()) {
-                handler.onFailure(new IllegalArgumentException("leader index primary shards [" +
-                        leaderIndexMetadata.getNumberOfShards() +  "] does not match with the number of " +
-                        "shards of the follow index [" + followIndexMetadata.getNumberOfShards() + "]"));
-                // TODO: other validation checks
-            } else {
+            validate (leaderIndexMetadata ,followIndexMetadata , request);
                 final int numShards = followIndexMetadata.getNumberOfShards();
                 final AtomicInteger counter = new AtomicInteger(numShards);
                 final AtomicReferenceArray<Object> responses = new AtomicReferenceArray<>(followIndexMetadata.getNumberOfShards());
                 Map<String, String> filteredHeaders = threadPool.getThreadContext().getHeaders().entrySet().stream()
                     .filter(e -> ShardFollowTask.HEADER_FILTERS.contains(e.getKey()))
-                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-                for (int i = 0; i < numShards; i++) {
+                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));for (int i = 0; i < numShards; i++) {
                     final int shardId = i;
                     String taskId = followIndexMetadata.getIndexUUID() + "-" + shardId;
                     ShardFollowTask shardFollowTask = new ShardFollowTask(clusterNameAlias,
@@ -261,39 +246,59 @@ public void onResponse(PersistentTasksCustomMetaData.PersistentTask<ShardFollowT
                                     finalizeResponse();
                                 }
 
-                                @Override
-                                public void onFailure(Exception e) {
-                                    responses.set(shardId, e);
-                                    finalizeResponse();
-                                }
+                            @Override
+                            public void onFailure(Exception e) {
+                                responses.set(shardId, e);
+                                finalizeResponse();
+                            }
 
-                                void finalizeResponse() {
-                                    Exception error = null;
-                                    if (counter.decrementAndGet() == 0) {
-                                        for (int j = 0; j < responses.length(); j++) {
-                                            Object response = responses.get(j);
-                                            if (response instanceof Exception) {
-                                                if (error == null) {
-                                                    error = (Exception) response;
-                                                } else {
-                                                    error.addSuppressed((Throwable) response);
-                                                }
+                            void finalizeResponse() {
+                                Exception error = null;
+                                if (counter.decrementAndGet() == 0) {
+                                    for (int j = 0; j < responses.length(); j++) {
+                                        Object response = responses.get(j);
+                                        if (response instanceof Exception) {
+                                            if (error == null) {
+                                                error = (Exception) response;
+                                            } else {
+                                                error.addSuppressed((Throwable) response);
                                             }
                                         }
+                                    }
 
-                                        if (error == null) {
-                                            // include task ids?
-                                            handler.onResponse(new Response(true));
-                                        } else {
-                                            // TODO: cancel all started tasks
-                                            handler.onFailure(error);
-                                        }
+                                    if (error == null) {
+                                        // include task ids?
+                                        handler.onResponse(new Response(true));
+                                    } else {
+                                        // TODO: cancel all started tasks
+                                        handler.onFailure(error);
                                     }
                                 }
                             }
-                    );
-                }
+                        }
+                );
             }
         }
     }
+
+
+    static void validate(IndexMetaData leaderIndex, IndexMetaData followIndex, Request request) {
+        if (leaderIndex == null) {
+            throw new IllegalArgumentException("leader index [" + request.leaderIndex + "] does not exist");
+        }
+
+        if (followIndex == null) {
+            throw new IllegalArgumentException("follow index [" + request.followIndex + "] does not exist");
+        }
+        if (leaderIndex.getSettings().getAsBoolean(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), false) == false) {
+            throw new IllegalArgumentException("leader index [" + request.leaderIndex + "] does not have soft deletes enabled");
+        }
+
+        if (leaderIndex.getNumberOfShards() != followIndex.getNumberOfShards()) {
+            throw new IllegalArgumentException("leader index primary shards [" + leaderIndex.getNumberOfShards() +
+                "] does not match with the number of shards of the follow index [" + followIndex.getNumberOfShards() + "]");
+        }
+        // TODO: other validation checks
+    }
+
 }
diff --git a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/ShardChangesIT.java b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/ShardChangesIT.java
index 13b9b4a6ffb76..5c3ec341a316c 100644
--- a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/ShardChangesIT.java
+++ b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/ShardChangesIT.java
@@ -19,6 +19,7 @@
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.common.xcontent.support.XContentMapValues;
 import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
@@ -142,7 +143,8 @@ public void testGetOperationsBasedOnGlobalSequenceId() throws Exception {
     public void testFollowIndex() throws Exception {
         final int numberOfPrimaryShards = randomIntBetween(1, 3);
 
-        final String leaderIndexSettings = getIndexSettings(numberOfPrimaryShards, Collections.emptyMap());
+        final String leaderIndexSettings = getIndexSettings(numberOfPrimaryShards,
+            Collections.singletonMap(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true"));
         assertAcked(client().admin().indices().prepareCreate("index1").setSource(leaderIndexSettings, XContentType.JSON));
 
         final String followerIndexSettings =
diff --git a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/FollowExistingIndexActionTests.java b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/FollowExistingIndexActionTests.java
new file mode 100644
index 0000000000000..5c6e3e3f9543d
--- /dev/null
+++ b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/FollowExistingIndexActionTests.java
@@ -0,0 +1,66 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.ccr.action;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.test.ESTestCase;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class FollowExistingIndexActionTests extends ESTestCase {
+
+    public void testValidation() {
+        FollowExistingIndexAction.Request request = new FollowExistingIndexAction.Request();
+        request.setLeaderIndex("index1");
+        request.setFollowIndex("index2");
+
+        {
+            Exception e = expectThrows(IllegalArgumentException.class, () -> FollowExistingIndexAction.validate(null, null, request));
+            assertThat(e.getMessage(), equalTo("leader index [index1] does not exist"));
+        }
+        {
+            IndexMetaData leaderIMD = createIMD("index1", 5);
+            Exception e = expectThrows(IllegalArgumentException.class, () -> FollowExistingIndexAction.validate(leaderIMD, null, request));
+            assertThat(e.getMessage(), equalTo("follow index [index2] does not exist"));
+        }
+        {
+            IndexMetaData leaderIMD = createIMD("index1", 5);
+            IndexMetaData followIMD = createIMD("index2", 5);
+            Exception e = expectThrows(IllegalArgumentException.class,
+                () -> FollowExistingIndexAction.validate(leaderIMD, followIMD, request));
+            assertThat(e.getMessage(), equalTo("leader index [index1] does not have soft deletes enabled"));
+        }
+        {
+            IndexMetaData leaderIMD = createIMD("index1", 5, new Tuple<>(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true"));
+            IndexMetaData followIMD = createIMD("index2", 4);
+            Exception e = expectThrows(IllegalArgumentException.class,
+                () -> FollowExistingIndexAction.validate(leaderIMD, followIMD, request));
+            assertThat(e.getMessage(),
+                equalTo("leader index primary shards [5] does not match with the number of shards of the follow index [4]"));
+        }
+        {
+            IndexMetaData leaderIMD = createIMD("index1", 5, new Tuple<>(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true"));
+            IndexMetaData followIMD = createIMD("index2", 5);
+            FollowExistingIndexAction.validate(leaderIMD, followIMD, request);
+        }
+    }
+
+    private static IndexMetaData createIMD(String index, int numShards, Tuple<?, ?>... settings) {
+        Settings.Builder settingsBuilder = settings(Version.CURRENT);
+        for (Tuple<?, ?> setting : settings) {
+            settingsBuilder.put((String) setting.v1(), (String) setting.v2());
+        }
+        return IndexMetaData.builder(index).settings(settingsBuilder)
+            .numberOfShards(numShards)
+            .numberOfReplicas(0)
+            .setRoutingNumShards(numShards).build();
+    }
+
+}
diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/test/ccr/follow_and_unfollow.yml b/x-pack/plugin/src/test/resources/rest-api-spec/test/ccr/follow_and_unfollow.yml
index 6ab9b31fd6184..2d0a361f2d492 100644
--- a/x-pack/plugin/src/test/resources/rest-api-spec/test/ccr/follow_and_unfollow.yml
+++ b/x-pack/plugin/src/test/resources/rest-api-spec/test/ccr/follow_and_unfollow.yml
@@ -4,6 +4,10 @@
       indices.create:
         index: foo
         body:
+          settings:
+            index:
+              soft_deletes:
+                enabled: true
           mappings:
             doc:
               properties: