[CCR] Fail with a descriptive error if leader index does not exist (#33797)

Closes #33737

Author: martijnvg

x-pack/plugin/ccr/qa/multi-cluster/src/test/java/org/elasticsearch/xpack/ccr/FollowIndexIT.java

@@ -25,6 +25,7 @@
 import java.util.Map;
 
 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.endsWith;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
@@ -82,6 +83,18 @@ public void testFollowIndex() throws Exception {
         }
     }
 
+    public void testFollowNonExistingLeaderIndex() throws Exception {
+        assumeFalse("Test should only run when both clusters are running", runningAgainstLeaderCluster);
+        ResponseException e = expectThrows(ResponseException.class,
+            () -> followIndex("leader_cluster:non-existing-index", "non-existing-index"));
+        assertThat(e.getMessage(), containsString("no such index"));
+        assertThat(e.getResponse().getStatusLine().getStatusCode(), equalTo(404));
+
+        e = expectThrows(ResponseException.class, () -> createAndFollowIndex("leader_cluster:non-existing-index", "non-existing-index"));
+        assertThat(e.getMessage(), containsString("no such index"));
+        assertThat(e.getResponse().getStatusLine().getStatusCode(), equalTo(404));
+    }
+
     public void testAutoFollowPatterns() throws Exception {
         assumeFalse("Test should only run when both clusters are running", runningAgainstLeaderCluster);
 

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

@@ -26,6 +26,7 @@
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.CheckedConsumer;
+import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.engine.CommitStats;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.shard.ShardId;
@@ -110,6 +111,11 @@ public <T> void checkRemoteClusterLicenseAndFetchLeaderIndexMetadataAndHistoryUU
                 onFailure,
                 leaderClusterState -> {
                     IndexMetaData leaderIndexMetaData = leaderClusterState.getMetaData().index(leaderIndex);
+                    if (leaderIndexMetaData == null) {
+                        onFailure.accept(new IndexNotFoundException(leaderIndex));
+                        return;
+                    }
+
                     final Client leaderClient = client.getRemoteClusterClient(clusterAlias);
                     fetchLeaderHistoryUUIDs(leaderClient, leaderIndexMetaData, onFailure, historyUUIDs -> {
                         consumer.accept(historyUUIDs, leaderIndexMetaData);

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

@@ -28,6 +28,7 @@
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.license.LicenseUtils;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.RemoteClusterAware;
@@ -121,6 +122,11 @@ private void createFollowerIndexAndFollowLocalIndex(
         // following an index in local cluster, so use local cluster state to fetch leader index metadata
         final String leaderIndex = request.getFollowRequest().getLeaderIndex();
         final IndexMetaData leaderIndexMetadata = state.getMetaData().index(leaderIndex);
+        if (leaderIndexMetadata == null) {
+            listener.onFailure(new IndexNotFoundException(leaderIndex));
+            return;
+        }
+
         Consumer<String[]> handler = historyUUIDs -> {
             createFollowerIndex(leaderIndexMetadata, historyUUIDs, request, listener);
         };

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

@@ -429,12 +429,21 @@ public void testFollowNonExistentIndex() throws Exception {
         // Leader index does not exist.
         FollowIndexAction.Request followRequest1 = createFollowRequest("non-existent-leader", "test-follower");
         expectThrows(IndexNotFoundException.class, () -> client().execute(FollowIndexAction.INSTANCE, followRequest1).actionGet());
+        expectThrows(IndexNotFoundException.class,
+            () -> client().execute(CreateAndFollowIndexAction.INSTANCE, new CreateAndFollowIndexAction.Request(followRequest1))
+                .actionGet());
         // Follower index does not exist.
         FollowIndexAction.Request followRequest2 = createFollowRequest("non-test-leader", "non-existent-follower");
         expectThrows(IndexNotFoundException.class, () -> client().execute(FollowIndexAction.INSTANCE, followRequest2).actionGet());
+        expectThrows(IndexNotFoundException.class,
+            () -> client().execute(CreateAndFollowIndexAction.INSTANCE, new CreateAndFollowIndexAction.Request(followRequest2))
+                .actionGet());
         // Both indices do not exist.
         FollowIndexAction.Request followRequest3 = createFollowRequest("non-existent-leader", "non-existent-follower");
         expectThrows(IndexNotFoundException.class, () -> client().execute(FollowIndexAction.INSTANCE, followRequest3).actionGet());
+        expectThrows(IndexNotFoundException.class,
+            () -> client().execute(CreateAndFollowIndexAction.INSTANCE, new CreateAndFollowIndexAction.Request(followRequest3))
+                .actionGet());
     }
 
     public void testFollowIndex_lowMaxTranslogBytes() throws Exception {