diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java
index 13b12d4b96f..b1d6467168c 100644
--- a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java
+++ b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java
@@ -29,8 +29,7 @@ import org.elasticsearch.indices.IndexClosedException;
 import org.elasticsearch.node.NodeClosedException;
 import org.elasticsearch.persistent.AllocatedPersistentTask;
 import org.elasticsearch.tasks.TaskId;
-import org.elasticsearch.transport.NodeDisconnectedException;
-import org.elasticsearch.transport.NodeNotConnectedException;
+import org.elasticsearch.transport.ConnectTransportException;
 import org.elasticsearch.xpack.ccr.action.bulk.BulkShardOperationsResponse;
 import org.elasticsearch.xpack.core.ccr.ShardFollowNodeTaskStatus;
 
@@ -448,7 +447,10 @@ public abstract class ShardFollowNodeTask extends AllocatedPersistentTask {
             return true;
         }
 
+        // This is thrown when using a Client and its remote cluster alias went MIA
         String noSuchRemoteClusterMessage = "no such remote cluster: " + remoteCluster;
+        // This is thrown when creating a Client and the remote cluster does not exist:
+        String unknownClusterMessage = "unknown cluster alias [" + remoteCluster + "]";
         final Throwable actual = ExceptionsHelper.unwrapCause(e);
         return actual instanceof ShardNotFoundException ||
             actual instanceof IllegalIndexShardStateException ||
@@ -458,11 +460,11 @@ public abstract class ShardFollowNodeTask extends AllocatedPersistentTask {
             actual instanceof ElasticsearchSecurityException || // If user does not have sufficient privileges
             actual instanceof ClusterBlockException || // If leader index is closed or no elected master
             actual instanceof IndexClosedException || // If follow index is closed
-            actual instanceof NodeDisconnectedException ||
-            actual instanceof NodeNotConnectedException ||
+            actual instanceof ConnectTransportException ||
             actual instanceof NodeClosedException ||
             (actual.getMessage() != null && actual.getMessage().contains("TransportService is closed")) ||
-            (actual instanceof IllegalArgumentException && noSuchRemoteClusterMessage.equals(actual.getMessage()));
+            (actual instanceof IllegalArgumentException && (noSuchRemoteClusterMessage.equals(actual.getMessage()) ||
+                unknownClusterMessage.equals(actual.getMessage())));
     }
 
     // These methods are protected for testing purposes:
diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutor.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutor.java
index 14ec147a536..97308126ffb 100644
--- a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutor.java
+++ b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutor.java
@@ -94,12 +94,6 @@ public class ShardFollowTasksExecutor extends PersistentTasksExecutor<ShardFollo
                                                  PersistentTasksCustomMetaData.PersistentTask<ShardFollowTask> taskInProgress,
                                                  Map<String, String> headers) {
         ShardFollowTask params = taskInProgress.getParams();
-        final Client remoteClient;
-        if (params.getRemoteCluster() != null) {
-            remoteClient = wrapClient(client.getRemoteClusterClient(params.getRemoteCluster()), params.getHeaders());
-        } else {
-            remoteClient = wrapClient(client, params.getHeaders());
-        }
         Client followerClient = wrapClient(client, params.getHeaders());
         BiConsumer<TimeValue, Runnable> scheduler = (delay, command) -> {
             try {
@@ -123,8 +117,7 @@ public class ShardFollowTasksExecutor extends PersistentTasksExecutor<ShardFollo
                 Index followIndex = params.getFollowShardId().getIndex();
 
                 ClusterStateRequest clusterStateRequest = CcrRequests.metaDataRequest(leaderIndex.getName());
-
-                remoteClient.admin().cluster().state(clusterStateRequest, ActionListener.wrap(clusterStateResponse -> {
+                CheckedConsumer<ClusterStateResponse, Exception> onResponse = clusterStateResponse -> {
                     IndexMetaData indexMetaData = clusterStateResponse.getState().metaData().getIndexSafe(leaderIndex);
                     if (indexMetaData.getMappings().isEmpty()) {
                         assert indexMetaData.getMappingVersion() == 1;
@@ -140,7 +133,12 @@ public class ShardFollowTasksExecutor extends PersistentTasksExecutor<ShardFollo
                     followerClient.admin().indices().putMapping(putMappingRequest, ActionListener.wrap(
                         putMappingResponse -> handler.accept(indexMetaData.getMappingVersion()),
                         errorHandler));
-                }, errorHandler));
+                };
+                try {
+                    remoteClient(params).admin().cluster().state(clusterStateRequest, ActionListener.wrap(onResponse, errorHandler));
+                } catch (Exception e) {
+                    errorHandler.accept(e);
+                }
             }
 
             @Override
@@ -181,7 +179,11 @@ public class ShardFollowTasksExecutor extends PersistentTasksExecutor<ShardFollo
                         }
                     }
                 };
-                remoteClient.admin().cluster().state(clusterStateRequest, ActionListener.wrap(onResponse, errorHandler));
+                try {
+                    remoteClient(params).admin().cluster().state(clusterStateRequest, ActionListener.wrap(onResponse, errorHandler));
+                } catch (Exception e) {
+                    errorHandler.accept(e);
+                }
             }
 
             private void closeIndexUpdateSettingsAndOpenIndex(String followIndex,
@@ -236,7 +238,7 @@ public class ShardFollowTasksExecutor extends PersistentTasksExecutor<ShardFollo
                 request.setMaxBatchSize(params.getMaxReadRequestSize());
                 request.setPollTimeout(params.getReadPollTimeout());
                 try {
-                    remoteClient.execute(ShardChangesAction.INSTANCE, request, ActionListener.wrap(handler::accept, errorHandler));
+                    remoteClient(params).execute(ShardChangesAction.INSTANCE, request, ActionListener.wrap(handler::accept, errorHandler));
                 } catch (Exception e) {
                     errorHandler.accept(e);
                 }
@@ -251,6 +253,10 @@ public class ShardFollowTasksExecutor extends PersistentTasksExecutor<ShardFollo
         return recordedLeaderShardHistoryUUIDs[params.getLeaderShardId().id()];
     }
 
+    private Client remoteClient(ShardFollowTask params) {
+        return wrapClient(client.getRemoteClusterClient(params.getRemoteCluster()), params.getHeaders());
+    }
+
     interface FollowerStatsInfoHandler {
         void accept(String followerHistoryUUID, long globalCheckpoint, long maxSeqNo);
     }
diff --git a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/CcrIntegTestCase.java b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/CcrIntegTestCase.java
index 01e51ea94f2..05b20050ee3 100644
--- a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/CcrIntegTestCase.java
+++ b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/CcrIntegTestCase.java
@@ -202,7 +202,7 @@ public abstract class CcrIntegTestCase extends ESTestCase {
         builder.put(LicenseService.SELF_GENERATED_LICENSE_TYPE.getKey(), "trial");
         // Let cluster state api return quickly in order to speed up auto follow tests:
         builder.put(CcrSettings.CCR_AUTO_FOLLOW_WAIT_FOR_METADATA_TIMEOUT.getKey(), TimeValue.timeValueMillis(100));
-        if (leaderSeedAddress != null) {
+        if (configureRemoteClusterViaNodeSettings() && leaderSeedAddress != null) {
             builder.put("cluster.remote.leader_cluster.seeds", leaderSeedAddress);
         }
         return new NodeConfigurationSource() {
@@ -247,6 +247,10 @@ public abstract class CcrIntegTestCase extends ESTestCase {
         return true;
     }
 
+    protected boolean configureRemoteClusterViaNodeSettings() {
+        return true;
+    }
+
     protected final Client leaderClient() {
         return clusterGroup.leaderCluster.client();
     }
diff --git a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/RestartIndexFollowingIT.java b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/RestartIndexFollowingIT.java
index d72eca17fdb..b8649aaa432 100644
--- a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/RestartIndexFollowingIT.java
+++ b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/RestartIndexFollowingIT.java
@@ -6,8 +6,11 @@
 
 package org.elasticsearch.xpack.ccr;
 
+import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.CcrIntegTestCase;
 import org.elasticsearch.xpack.core.ccr.action.PutFollowAction;
 
@@ -24,12 +27,17 @@ public class RestartIndexFollowingIT extends CcrIntegTestCase {
         return 1;
     }
 
+    @Override
+    protected boolean configureRemoteClusterViaNodeSettings() {
+        return false;
+    }
+
     public void testFollowIndex() throws Exception {
         final String leaderIndexSettings = getIndexSettings(1, 0,
             singletonMap(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true"));
-            singletonMap(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true");
         assertAcked(leaderClient().admin().indices().prepareCreate("index1").setSource(leaderIndexSettings, XContentType.JSON));
         ensureLeaderGreen("index1");
+        setupRemoteCluster();
 
         final PutFollowAction.Request followRequest = putFollow("index1", "index2");
         followerClient().execute(PutFollowAction.INSTANCE, followRequest).get();
@@ -57,6 +65,28 @@ public class RestartIndexFollowingIT extends CcrIntegTestCase {
             assertThat(followerClient().prepareSearch("index2").get().getHits().getTotalHits().value,
                 equalTo(firstBatchNumDocs + secondBatchNumDocs));
         });
+
+        getLeaderCluster().fullRestart();
+        ensureLeaderGreen("index1");
+        // Remote connection needs to be re-configured, because all the nodes in leader cluster have been restarted:
+        setupRemoteCluster();
+
+        final long thirdBatchNumDocs = randomIntBetween(2, 64);
+        for (int i = 0; i < thirdBatchNumDocs; i++) {
+            leaderClient().prepareIndex("index1", "doc").setSource("{}", XContentType.JSON).get();
+        }
+
+        assertBusy(() -> {
+            assertThat(followerClient().prepareSearch("index2").get().getHits().getTotalHits().value,
+                equalTo(firstBatchNumDocs + secondBatchNumDocs + thirdBatchNumDocs));
+        });
+    }
+
+    private void setupRemoteCluster() {
+        ClusterUpdateSettingsRequest updateSettingsRequest = new ClusterUpdateSettingsRequest();
+        String address = getLeaderCluster().getMasterNodeInstance(TransportService.class).boundAddress().publishAddress().toString();
+        updateSettingsRequest.persistentSettings(Settings.builder().put("cluster.remote.leader_cluster.seeds", address));
+        assertAcked(followerClient().admin().cluster().updateSettings(updateSettingsRequest).actionGet());
     }
 
 }