CCR: Stop FollowExistingIndexAction after report failure (#4111)

We check for the existence of both leader and follower index, then properly 
report to the caller. However, we do not return after reporting failure. This
causes the caller receive exception twice: IllegalArgumentException then
NullPointerException. This commit makes sure to stop the action after reporting
failure.

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

@@ -156,12 +156,14 @@ public class FollowExistingIndexAction extends Action<FollowExistingIndexAction.
                             .index(request.leaderIndex);
                     if (leaderIndexMetadata == null) {
                         listener.onFailure(new IllegalArgumentException("leader index [" + request.leaderIndex + "] does not exist"));
+                        return;
                     }
 
                     IndexMetaData followIndexMetadata = clusterStateResponse.getState().getMetaData()
                             .index(request.followIndex);
                     if (followIndexMetadata == null) {
                         listener.onFailure(new IllegalArgumentException("follow index [" + request.followIndex + "] does not exist"));
+                        return;
                     }
 
                     if (leaderIndexMetadata.getNumberOfShards() != followIndexMetadata.getNumberOfShards()) {

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

@@ -34,15 +34,15 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.translog.Translog;
+import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.discovery.TestZenDiscovery;
-import org.elasticsearch.xpack.core.XPackSettings;
 import org.elasticsearch.xpack.ccr.action.FollowExistingIndexAction;
 import org.elasticsearch.xpack.ccr.action.ShardChangesAction;
 import org.elasticsearch.xpack.ccr.action.ShardFollowTask;
 import org.elasticsearch.xpack.ccr.action.UnfollowIndexAction;
-import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
+import org.elasticsearch.xpack.core.XPackSettings;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -219,6 +219,24 @@ public class ShardChangesIT extends ESIntegTestCase {
         });
     }
 
+    public void testFollowNonExistentIndex() throws Exception {
+        assertAcked(client().admin().indices().prepareCreate("test-leader").get());
+        assertAcked(client().admin().indices().prepareCreate("test-follower").get());
+        final FollowExistingIndexAction.Request followRequest = new FollowExistingIndexAction.Request();
+        // Leader index does not exist.
+        followRequest.setLeaderIndex("non-existent-leader");
+        followRequest.setFollowIndex("test-follower");
+        expectThrows(IllegalArgumentException.class, () -> client().execute(FollowExistingIndexAction.INSTANCE, followRequest).actionGet());
+        // Follower index does not exist.
+        followRequest.setLeaderIndex("test-leader");
+        followRequest.setFollowIndex("non-existent-follower");
+        expectThrows(IllegalArgumentException.class, () -> client().execute(FollowExistingIndexAction.INSTANCE, followRequest).actionGet());
+        // Both indices do not exist.
+        followRequest.setLeaderIndex("non-existent-leader");
+        followRequest.setFollowIndex("non-existent-follower");
+        expectThrows(IllegalArgumentException.class, () -> client().execute(FollowExistingIndexAction.INSTANCE, followRequest).actionGet());
+    }
+
     private CheckedRunnable<Exception> assertTask(final int numberOfPrimaryShards, final Map<ShardId, Long> numDocsPerShard) {
         return () -> {
             final ClusterState clusterState = client().admin().cluster().prepareState().get().getState();