Handle no such remote cluster exception in ccr (#53415)

A remote client can throw a NoSuchRemoteClusterException while fetching 
the cluster state from the leader cluster. We also need to handle that
exception when retrying to add a retention lease to the leader shard.

Closes #53225

Author: dnhatn

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

@@ -136,16 +136,7 @@ protected void innerUpdateMapping(long minRequiredMappingVersion, LongConsumer h
                 final Index followerIndex = params.getFollowShardId().getIndex();
                 final Index leaderIndex = params.getLeaderShardId().getIndex();
                 final Supplier<TimeValue> timeout = () -> isStopped() ? TimeValue.MINUS_ONE : waitForMetadataTimeOut;
-
-                final Client remoteClient;
-                try {
-                    remoteClient = remoteClient(params);
-                } catch (NoSuchRemoteClusterException e) {
-                    errorHandler.accept(e);
-                    return;
-                }
-
-                CcrRequests.getIndexMetadata(remoteClient, leaderIndex, minRequiredMappingVersion, 0L, timeout, ActionListener.wrap(
+                final ActionListener<IndexMetaData> listener = ActionListener.wrap(
                     indexMetaData -> {
                         if (indexMetaData.getMappings().isEmpty()) {
                             assert indexMetaData.getMappingVersion() == 1;
@@ -161,7 +152,12 @@ protected void innerUpdateMapping(long minRequiredMappingVersion, LongConsumer h
                             errorHandler));
                     },
                     errorHandler
-                ));
+                );
+                try {
+                    CcrRequests.getIndexMetadata(remoteClient(params), leaderIndex, minRequiredMappingVersion, 0L, timeout, listener);
+                } catch (NoSuchRemoteClusterException e) {
+                    errorHandler.accept(e);
+                }
             }
 
             @Override
@@ -311,21 +307,27 @@ protected Scheduler.Cancellable scheduleBackgroundRetentionLeaseRenewal(final Lo
                                         "{} background adding retention lease [{}] while following",
                                         params.getFollowShardId(),
                                         retentionLeaseId);
-                                CcrRetentionLeases.asyncAddRetentionLease(
+                                try {
+                                    final ActionListener<RetentionLeaseActions.Response> wrappedListener = ActionListener.wrap(
+                                        r -> {},
+                                        inner -> {
+                                            /*
+                                             * If this fails that the retention lease already exists, something highly unusual is
+                                             * going on. Log it, and renew again after another renew interval has passed.
+                                             */
+                                            final Throwable innerCause = ExceptionsHelper.unwrapCause(inner);
+                                            logRetentionLeaseFailure(retentionLeaseId, innerCause);
+                                        });
+                                    CcrRetentionLeases.asyncAddRetentionLease(
                                         params.getLeaderShardId(),
                                         retentionLeaseId,
                                         followerGlobalCheckpoint.getAsLong(),
                                         remoteClient(params),
-                                        ActionListener.wrap(
-                                                r -> {},
-                                                inner -> {
-                                                    /*
-                                                     * If this fails that the retention lease already exists, something highly unusual is
-                                                     * going on. Log it, and renew again after another renew interval has passed.
-                                                     */
-                                                    final Throwable innerCause = ExceptionsHelper.unwrapCause(inner);
-                                                    logRetentionLeaseFailure(retentionLeaseId, innerCause);
-                                                }));
+                                        wrappedListener);
+                                } catch (NoSuchRemoteClusterException rce) {
+                                    // we will attempt to renew again after another renew interval has passed
+                                    logRetentionLeaseFailure(retentionLeaseId, rce);
+                                }
                             } else {
                                  // if something else happened, we will attempt to renew again after another renew interval has passed
                             }