[CCR] Check whether the rejected execution exception has the shutdown flag set (#33703)

and if so debug log it and otherwise rethrow.

This should fix a couple of test failures where during test teardown tests
failed due to uncaught exceptions being detected.

Author: martijnvg

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

@@ -24,6 +24,7 @@
 import org.elasticsearch.cluster.routing.IndexRoutingTable;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.Index;
@@ -91,8 +92,17 @@ protected AllocatedPersistentTask createTask(long id, String type, String action
             leaderClient = wrapClient(client, params);
         }
         Client followerClient = wrapClient(client, params);
-        BiConsumer<TimeValue, Runnable> scheduler =
-            (delay, command) -> threadPool.schedule(delay, Ccr.CCR_THREAD_POOL_NAME, command);
+        BiConsumer<TimeValue, Runnable> scheduler = (delay, command) -> {
+            try {
+                threadPool.schedule(delay, Ccr.CCR_THREAD_POOL_NAME, command);
+            } catch (EsRejectedExecutionException e) {
+                if (e.isExecutorShutdown()) {
+                    logger.debug("couldn't schedule command, executor is shutting down", e);
+                } else {
+                    throw e;
+                }
+            }
+        };
         return new ShardFollowNodeTask(
                 id, type, action, getDescription(taskInProgress), parentTaskId, headers, params, scheduler, System::nanoTime) {