Deprioritize master service (#94318)

The changes introduced in #92021 mean that the master service no longer
needs to use a prioritized executor. Prioritized executors are weird,
for instance they don't propagate rejections to `AbstractRunnable` tasks
properly. This commit moves to using a regular scaling executor and
removes some of the now-unnecessary workarounds for handling the
prioritized executor's weirdness.

server/src/main/java/org/elasticsearch/cluster/service/MasterService.java

@@ -39,7 +39,6 @@ import org.elasticsearch.common.util.concurrent.CountDown;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.common.util.concurrent.FutureUtils;
-import org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Releasable;
@@ -62,6 +61,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -108,7 +108,7 @@ public class MasterService extends AbstractLifecycleComponent {
     protected final ThreadPool threadPool;
     private final TaskManager taskManager;
 
-    private volatile PrioritizedEsThreadPoolExecutor threadPoolExecutor;
+    private volatile ExecutorService threadPoolExecutor;
     private final AtomicInteger totalQueueSize = new AtomicInteger();
     private volatile Batch currentlyExecutingBatch;
     private final Map<Priority, PerPriorityQueue> queuesByPriority;
@@ -155,12 +155,16 @@ public class MasterService extends AbstractLifecycleComponent {
         threadPoolExecutor = createThreadPoolExecutor();
     }
 
-    protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
-        return EsExecutors.newSinglePrioritizing(
+    protected ExecutorService createThreadPoolExecutor() {
+        return EsExecutors.newScaling(
             nodeName + "/" + MASTER_UPDATE_THREAD_NAME,
+            0,
+            1,
+            60,
+            TimeUnit.SECONDS,
+            true,
             daemonThreadFactory(nodeName, MASTER_UPDATE_THREAD_NAME),
-            threadPool.getThreadContext(),
-            threadPool.scheduler()
+            threadPool.getThreadContext()
         );
     }
 
@@ -1137,24 +1141,30 @@ public class MasterService extends AbstractLifecycleComponent {
         return e instanceof NotMasterException || e instanceof FailedToCommitClusterStateException;
     }
 
-    private final Runnable queuesProcessor = new Runnable() {
+    private final Runnable queuesProcessor = new AbstractRunnable() {
         @Override
-        public void run() {
+        public void doRun() {
             assert threadPool.getThreadContext().isSystemContext();
             assert totalQueueSize.get() > 0;
             assert currentlyExecutingBatch == null;
-            try {
-                final var nextBatch = takeNextBatch();
-                assert currentlyExecutingBatch == nextBatch;
-                if (lifecycle.started()) {
-                    nextBatch.run();
-                } else {
-                    nextBatch.onRejection(new FailedToCommitClusterStateException("node closed", getRejectionException()));
-                }
-            } catch (Exception e) {
-                logger.error("unexpected exception executing queue entry", e);
-                assert false : e;
-            } finally {
+            final var nextBatch = takeNextBatch();
+            assert currentlyExecutingBatch == nextBatch;
+            if (lifecycle.started()) {
+                nextBatch.run();
+            } else {
+                nextBatch.onRejection(new FailedToCommitClusterStateException("node closed", getRejectionException()));
+            }
+        }
+
+        @Override
+        public void onFailure(Exception e) {
+            logger.error("unexpected exception executing queue entry", e);
+            assert false : e;
+        }
+
+        @Override
+        public void onAfter() {
+            if (currentlyExecutingBatch != null) {
                 currentlyExecutingBatch = null;
                 if (totalQueueSize.decrementAndGet() > 0) {
                     starvationWatcher.onNonemptyQueue();
@@ -1165,6 +1175,12 @@ public class MasterService extends AbstractLifecycleComponent {
             }
         }
 
+        @Override
+        public void onRejection(Exception e) {
+            assert e instanceof EsRejectedExecutionException esre && esre.isExecutorShutdown() : e;
+            drainQueueOnRejection(new FailedToCommitClusterStateException("node closed", e));
+        }
+
         @Override
         public String toString() {
             return "master service queue processor";
@@ -1193,16 +1209,11 @@ public class MasterService extends AbstractLifecycleComponent {
             return;
         }
 
-        try {
-            assert totalQueueSize.get() > 0;
-            final var threadContext = threadPool.getThreadContext();
-            try (var ignored = threadContext.stashContext()) {
-                threadContext.markAsSystemContext();
-                threadPoolExecutor.execute(queuesProcessor);
-            }
-        } catch (Exception e) {
-            assert e instanceof EsRejectedExecutionException esre && esre.isExecutorShutdown() : e;
-            drainQueueOnRejection(new FailedToCommitClusterStateException("node closed", e));
+        assert totalQueueSize.get() > 0;
+        final var threadContext = threadPool.getThreadContext();
+        try (var ignored = threadContext.stashContext()) {
+            threadContext.markAsSystemContext();
+            threadPoolExecutor.execute(queuesProcessor);
         }
     }
 

server/src/test/java/org/elasticsearch/cluster/routing/allocation/allocator/ClusterAllocationSimulationTests.java

@@ -37,7 +37,9 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.RatioValue;
 import org.elasticsearch.common.util.concurrent.DeterministicTaskQueue;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor;
+import org.elasticsearch.common.util.concurrent.StoppableExecutorServiceWrapper;
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.snapshots.SnapshotShardSizeInfo;
@@ -54,6 +56,7 @@ import java.util.Comparator;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -216,28 +219,6 @@ public class ClusterAllocationSimulationTests extends ESAllocationTestCase {
         final var settings = Settings.EMPTY;
         final var clusterSettings = new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
 
-        final var directExecutor = new PrioritizedEsThreadPoolExecutor(
-            "master-service",
-            1,
-            1,
-            1,
-            TimeUnit.SECONDS,
-            r -> { throw new AssertionError("should not create new threads"); },
-            null,
-            null
-        ) {
-
-            @Override
-            public void execute(Runnable command, final TimeValue timeout, final Runnable timeoutCallback) {
-                execute(command);
-            }
-
-            @Override
-            public void execute(Runnable command) {
-                command.run();
-            }
-        };
-
         final var masterService = new MasterService(
             settings,
             clusterSettings,
@@ -245,12 +226,34 @@ public class ClusterAllocationSimulationTests extends ESAllocationTestCase {
             new TaskManager(settings, threadPool, Set.of())
         ) {
             @Override
-            protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
-                return directExecutor;
+            protected ExecutorService createThreadPoolExecutor() {
+                return new StoppableExecutorServiceWrapper(EsExecutors.DIRECT_EXECUTOR_SERVICE);
             }
         };
 
         final var applierService = new ClusterApplierService("master", settings, clusterSettings, threadPool) {
+            private final PrioritizedEsThreadPoolExecutor directExecutor = new PrioritizedEsThreadPoolExecutor(
+                "master-service",
+                1,
+                1,
+                1,
+                TimeUnit.SECONDS,
+                r -> { throw new AssertionError("should not create new threads"); },
+                null,
+                null
+            ) {
+
+                @Override
+                public void execute(Runnable command, final TimeValue timeout, final Runnable timeoutCallback) {
+                    execute(command);
+                }
+
+                @Override
+                public void execute(Runnable command) {
+                    command.run();
+                }
+            };
+
             @Override
             protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
                 return directExecutor;

server/src/test/java/org/elasticsearch/cluster/service/MasterServiceTests.java

@@ -25,6 +25,7 @@ import org.elasticsearch.cluster.ClusterStateTaskListener;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.LocalMasterServiceTask;
 import org.elasticsearch.cluster.NotMasterException;
+import org.elasticsearch.cluster.SimpleBatchedExecutor;
 import org.elasticsearch.cluster.ack.AckedRequest;
 import org.elasticsearch.cluster.block.ClusterBlocks;
 import org.elasticsearch.cluster.coordination.ClusterStatePublisher;
@@ -33,13 +34,14 @@ import org.elasticsearch.cluster.metadata.ProcessClusterEventTimeoutException;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.common.Priority;
+import org.elasticsearch.common.Randomness;
 import org.elasticsearch.common.component.Lifecycle;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.concurrent.DeterministicTaskQueue;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
-import org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor;
+import org.elasticsearch.common.util.concurrent.StoppableExecutorServiceWrapper;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.core.SuppressForbidden;
 import org.elasticsearch.core.TimeValue;
@@ -59,13 +61,16 @@ import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
 
+import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.EnumMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.BrokenBarrierException;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -80,6 +85,7 @@ import static org.elasticsearch.cluster.service.MasterService.MAX_TASK_DESCRIPTI
 import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
@@ -133,7 +139,7 @@ public class MasterServiceTests extends ESTestCase {
         boolean makeMaster,
         TaskManager taskManager,
         ThreadPool threadPool,
-        PrioritizedEsThreadPoolExecutor threadPoolExecutor
+        ExecutorService threadPoolExecutor
     ) {
         final DiscoveryNode localNode = new DiscoveryNode("node1", buildNewFakeTransportAddress(), emptyMap(), emptySet(), Version.CURRENT);
         final Settings settings = Settings.builder()
@@ -152,7 +158,7 @@ public class MasterServiceTests extends ESTestCase {
             taskManager
         ) {
             @Override
-            protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
+            protected ExecutorService createThreadPoolExecutor() {
                 if (threadPoolExecutor == null) {
                     return super.createThreadPoolExecutor();
                 } else {
@@ -1119,7 +1125,7 @@ public class MasterServiceTests extends ESTestCase {
                 return ClusterState.builder(batchExecutionContext.initialState()).build();
             }).submitTask("testBlockingCallInClusterStateTaskListenerFails", new ExpectSuccessTask(), null);
 
-            latch.await();
+            assertTrue(latch.await(10, TimeUnit.SECONDS));
             assertNotNull(assertionRef.get());
             assertThat(assertionRef.get().getMessage(), containsString("Reason: [Blocking operation]"));
         }
@@ -1923,9 +1929,7 @@ public class MasterServiceTests extends ESTestCase {
         final var deterministicTaskQueue = new DeterministicTaskQueue();
 
         final var threadPool = deterministicTaskQueue.getThreadPool();
-        final var threadPoolExecutor = deterministicTaskQueue.getPrioritizedEsThreadPoolExecutor();
-
-        try (var masterService = createMasterService(true, null, threadPool, threadPoolExecutor)) {
+        try (var masterService = createMasterService(true, null, threadPool, new StoppableExecutorServiceWrapper(threadPool.generic()))) {
 
             final var actionCount = new AtomicInteger();
 
@@ -2101,32 +2105,18 @@ public class MasterServiceTests extends ESTestCase {
     public void testRejectionBehaviour() {
 
         final var deterministicTaskQueue = new DeterministicTaskQueue();
-
         final var threadPool = deterministicTaskQueue.getThreadPool();
-        final var threadPoolExecutor = new PrioritizedEsThreadPoolExecutor(
+        final var threadPoolExecutor = EsExecutors.newScaling(
             "Rejecting",
             1,
             1,
             1,
             TimeUnit.SECONDS,
+            true,
             r -> { throw new AssertionError("should not create new threads"); },
-            null,
-            null
-        ) {
-            @Override
-            public void execute(Runnable command, final TimeValue timeout, final Runnable timeoutCallback) {
-                throw new AssertionError("not implemented");
-            }
-
-            @Override
-            public void execute(Runnable command) {
-                if (command instanceof AbstractRunnable) {
-                    throw new AssertionError("unexpected abstract runnable: " + command);
-                } else {
-                    throw new EsRejectedExecutionException("test", true);
-                }
-            }
-        };
+            threadPool.getThreadContext()
+        );
+        threadPoolExecutor.shutdown();
 
         try (var masterService = createMasterService(true, null, threadPool, threadPoolExecutor)) {
 
@@ -2174,9 +2164,8 @@ public class MasterServiceTests extends ESTestCase {
                     }
                 });
             }
-            threadPool.getThreadContext().markAsSystemContext();
-            deterministicTaskQueue.runAllTasks();
-
+            assertFalse(deterministicTaskQueue.hasRunnableTasks());
+            assertFalse(deterministicTaskQueue.hasDeferredTasks());
             assertEquals(2, actionCount.get());
         }
     }
@@ -2186,9 +2175,7 @@ public class MasterServiceTests extends ESTestCase {
         final var deterministicTaskQueue = new DeterministicTaskQueue();
 
         final var threadPool = deterministicTaskQueue.getThreadPool();
-        final var threadPoolExecutor = deterministicTaskQueue.getPrioritizedEsThreadPoolExecutor();
-
-        try (var masterService = createMasterService(true, null, threadPool, threadPoolExecutor)) {
+        try (var masterService = createMasterService(true, null, threadPool, new StoppableExecutorServiceWrapper(threadPool.generic()))) {
 
             final var actionCount = new AtomicInteger();
             final var testHeader = "test-header";
@@ -2254,9 +2241,7 @@ public class MasterServiceTests extends ESTestCase {
         final var deterministicTaskQueue = new DeterministicTaskQueue();
 
         final var threadPool = deterministicTaskQueue.getThreadPool();
-        final var threadPoolExecutor = deterministicTaskQueue.getPrioritizedEsThreadPoolExecutor();
-
-        try (var masterService = createMasterService(true, null, threadPool, threadPoolExecutor)) {
+        try (var masterService = createMasterService(true, null, threadPool, new StoppableExecutorServiceWrapper(threadPool.generic()))) {
 
             final var actionCount = new AtomicInteger();
             final var testHeader = "test-header";
@@ -2349,6 +2334,71 @@ public class MasterServiceTests extends ESTestCase {
         }
     }
 
+    public void testPrioritization() {
+        final var deterministicTaskQueue = new DeterministicTaskQueue();
+        final var threadPool = deterministicTaskQueue.getThreadPool();
+        try (var masterService = createMasterService(true, null, threadPool, new StoppableExecutorServiceWrapper(threadPool.generic()))) {
+
+            // specify the order in which the priorities should run, rather than relying on their enum order which would be easy to reverse
+            final var prioritiesOrder = List.of(
+                Priority.IMMEDIATE,
+                Priority.URGENT,
+                Priority.HIGH,
+                Priority.NORMAL,
+                Priority.LOW,
+                Priority.LANGUID
+            );
+            final var prioritiesQueue = new ArrayDeque<>(prioritiesOrder);
+
+            final var simpleExecutor = new SimpleBatchedExecutor<ClusterStateUpdateTask, Void>() {
+                @Override
+                public Tuple<ClusterState, Void> executeTask(ClusterStateUpdateTask task, ClusterState clusterState) throws Exception {
+                    return Tuple.tuple(task.execute(clusterState), null);
+                }
+
+                @Override
+                public void taskSucceeded(ClusterStateUpdateTask clusterStateTaskListener, Void result) {}
+            };
+
+            final var queues = new EnumMap<Priority, MasterServiceTaskQueue<ClusterStateUpdateTask>>(Priority.class);
+            final var tasks = new ArrayList<ClusterStateUpdateTask>();
+            for (final var priority : Priority.values()) {
+                queues.put(priority, masterService.createTaskQueue(priority.name(), priority, simpleExecutor));
+                tasks.add(new ClusterStateUpdateTask(priority) {
+                    @Override
+                    public ClusterState execute(ClusterState currentState) {
+                        assertEquals(priority, prioritiesQueue.poll());
+                        assertEquals(priority, priority());
+                        return currentState;
+                    }
+
+                    @Override
+                    public void onFailure(Exception e) {
+                        throw new AssertionError("unexpected", e);
+                    }
+                });
+            }
+
+            Randomness.shuffle(tasks);
+            for (final var task : tasks) {
+                if (randomBoolean()) {
+                    queues.get(task.priority()).submitTask("test", task, null);
+                } else {
+                    masterService.submitUnbatchedStateUpdateTask("test", task);
+                }
+            }
+
+            assertEquals(
+                prioritiesOrder,
+                masterService.pendingTasks().stream().map(PendingClusterTask::priority).collect(Collectors.toList())
+            );
+
+            threadPool.getThreadContext().markAsSystemContext();
+            deterministicTaskQueue.runAllTasks();
+            assertThat(prioritiesQueue, empty());
+        }
+    }
+
     /**
      * Returns the cluster state that the master service uses (and that is provided by the discovery layer)
      */

server/src/test/java/org/elasticsearch/indices/cluster/ClusterStateChanges.java

@@ -75,7 +75,8 @@ import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.IndexScopedSettings;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
+import org.elasticsearch.common.util.concurrent.StoppableExecutorServiceWrapper;
 import org.elasticsearch.core.CheckedFunction;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.env.Environment;
@@ -108,7 +109,7 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ExecutorService;
 import java.util.function.Function;
 
 import static com.carrotsearch.randomizedtesting.RandomizedTest.getRandom;
@@ -175,28 +176,9 @@ public class ClusterStateChanges {
             new TaskManager(SETTINGS, threadPool, Collections.emptySet())
         ) {
             @Override
-            protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
+            protected ExecutorService createThreadPoolExecutor() {
                 // run master tasks inline, no need to fork to a separate thread
-                return new PrioritizedEsThreadPoolExecutor(
-                    "fake-master",
-                    1,
-                    1,
-                    1,
-                    TimeUnit.SECONDS,
-                    r -> { throw new AssertionError("should not create new threads"); },
-                    null,
-                    null
-                ) {
-                    @Override
-                    public void execute(Runnable command, final TimeValue timeout, final Runnable timeoutCallback) {
-                        command.run();
-                    }
-
-                    @Override
-                    public void execute(Runnable command) {
-                        command.run();
-                    }
-                };
+                return new StoppableExecutorServiceWrapper(EsExecutors.DIRECT_EXECUTOR_SERVICE);
             }
         };
         // mocks

test/framework/src/main/java/org/elasticsearch/cluster/service/FakeThreadPoolMasterService.java

@@ -13,7 +13,6 @@ import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStatePublicationEvent;
 import org.elasticsearch.cluster.coordination.ClusterStatePublisher.AckListener;
-import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
@@ -27,6 +26,7 @@ import org.elasticsearch.threadpool.ThreadPool;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
@@ -71,7 +71,7 @@ public class FakeThreadPoolMasterService extends MasterService {
     }
 
     @Override
-    protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
+    protected ExecutorService createThreadPoolExecutor() {
         return new PrioritizedEsThreadPoolExecutor(
             name,
             1,
@@ -96,11 +96,6 @@ public class FakeThreadPoolMasterService extends MasterService {
                     scheduleNextTaskIfNecessary();
                 }
             }
-
-            @Override
-            public Pending[] getPending() {
-                return pendingTasks.stream().map(r -> new Pending(r, Priority.NORMAL, 0L, false)).toArray(Pending[]::new);
-            }
         };
     }
 

test/framework/src/main/java/org/elasticsearch/common/util/concurrent/StoppableExecutorServiceWrapper.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.common.util.concurrent;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Wrapper around an {@link ExecutorService} which fakes responses to shutdown-related methods.
+ */
+public class StoppableExecutorServiceWrapper implements ExecutorService {
+
+    private final ExecutorService delegate;
+
+    public StoppableExecutorServiceWrapper(ExecutorService delegate) {
+        this.delegate = delegate;
+    }
+
+    @Override
+    public void shutdown() {}
+
+    @Override
+    public List<Runnable> shutdownNow() {
+        return List.of();
+    }
+
+    @Override
+    public boolean isShutdown() {
+        return false;
+    }
+
+    @Override
+    public boolean isTerminated() {
+        return false;
+    }
+
+    @Override
+    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+        return true;
+    }
+
+    @Override
+    public <T> Future<T> submit(Callable<T> task) {
+        return delegate.submit(task);
+    }
+
+    @Override
+    public <T> Future<T> submit(Runnable task, T result) {
+        return delegate.submit(task, result);
+    }
+
+    @Override
+    public Future<?> submit(Runnable task) {
+        return delegate.submit(task);
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
+        return delegate.invokeAll(tasks);
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
+        return delegate.invokeAll(tasks, timeout, unit);
+    }
+
+    @Override
+    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
+        return delegate.invokeAny(tasks);
+    }
+
+    @Override
+    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException,
+        ExecutionException, TimeoutException {
+        return delegate.invokeAny(tasks, timeout, unit);
+    }
+
+    @Override
+    public void execute(Runnable command) {
+        delegate.execute(command);
+    }
+}