Fail earlier Put Follow requests for closed leader indices (#47637)

Backport of (#47582)

Today when following a new leader index, we fetch the remote cluster state, 
check the remote cluster license, check the user privileges, retrieve the 
index shard stats before initiating a CCR restore session.

But if the leader index to follow is closed, we're executing a bunch of 
operations that would inevitability fail at some point (on retrieving the
 index shard stats, because this type of request forbid closed indices 
when resolving indices). We could fail a Put Follow request at the first 
step by checking the leader index state directly from the remote cluster 
state.

This also helps the Resume Follow API to fail a bit earlier.

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

@@ -32,6 +32,7 @@ import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.engine.CommitStats;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.indices.IndexClosedException;
 import org.elasticsearch.license.RemoteClusterLicenseChecker;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.rest.RestStatus;
@@ -130,7 +131,10 @@ public class CcrLicenseChecker {
                         onFailure.accept(new IndexNotFoundException(leaderIndex));
                         return;
                     }
-
+                    if (leaderIndexMetaData.getState() == IndexMetaData.State.CLOSE) {
+                        onFailure.accept(new IndexClosedException(leaderIndexMetaData.getIndex()));
+                        return;
+                    }
                     final Client remoteClient = client.getRemoteClusterClient(clusterAlias);
                     hasPrivilegesToFollowIndices(remoteClient, new String[] {leaderIndex}, e -> {
                         if (e == null) {

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

@@ -72,6 +72,7 @@ import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.seqno.ReplicationTracker;
 import org.elasticsearch.index.seqno.RetentionLeaseActions;
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.indices.IndexClosedException;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.rest.RestStatus;
@@ -110,6 +111,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
 import static java.util.Collections.singletonMap;
@@ -744,6 +746,47 @@ public class IndexFollowingIT extends CcrIntegTestCase {
         ensureNoCcrTasks();
     }
 
+    public void testFollowClosedIndex() {
+        final String leaderIndex = "test-index";
+        assertAcked(leaderClient().admin().indices().prepareCreate(leaderIndex)
+            .setSettings(Settings.builder()
+                .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true)
+                .build()));
+        assertAcked(leaderClient().admin().indices().prepareClose(leaderIndex));
+
+        final String followerIndex = "follow-test-index";
+        expectThrows(IndexClosedException.class,
+            () -> followerClient().execute(PutFollowAction.INSTANCE, putFollow(leaderIndex, followerIndex)).actionGet());
+        assertFalse(followerClient().admin().indices().prepareExists(followerIndex).get().isExists());
+    }
+
+    public void testResumeFollowOnClosedIndex() throws Exception {
+        final String leaderIndex = "test-index";
+        assertAcked(leaderClient().admin().indices().prepareCreate(leaderIndex)
+            .setSettings(Settings.builder()
+                .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true)
+                .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
+                .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0)
+                .build()));
+        ensureLeaderGreen(leaderIndex);
+
+        final int nbDocs = randomIntBetween(10, 100);
+        IntStream.of(nbDocs).forEach(i -> leaderClient().prepareIndex().setIndex(leaderIndex).setSource("field", i).get());
+
+        final String followerIndex = "follow-test-index";
+        PutFollowAction.Response response =
+            followerClient().execute(PutFollowAction.INSTANCE, putFollow(leaderIndex, followerIndex)).actionGet();
+        assertTrue(response.isFollowIndexCreated());
+        assertTrue(response.isFollowIndexShardsAcked());
+        assertTrue(response.isIndexFollowingStarted());
+
+        pauseFollow(followerIndex);
+        assertAcked(leaderClient().admin().indices().prepareClose(leaderIndex));
+
+        expectThrows(IndexClosedException.class, () ->
+            followerClient().execute(ResumeFollowAction.INSTANCE, resumeFollow(followerIndex)).actionGet());
+    }
+
     public void testDeleteFollowerIndex() throws Exception {
         assertAcked(leaderClient().admin().indices().prepareCreate("index1")
             .setSettings(Settings.builder()