Extract and use scheduleUnlessShuttingDown

Author: DaveCTurner

server/src/main/java/org/elasticsearch/cluster/coordination/ElectionSchedulerFactory.java

@@ -28,7 +28,6 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
-import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.threadpool.ThreadPool.Names;
 
@@ -181,15 +180,7 @@ public String toString() {
             };
 
             logger.debug("scheduling {}", runnable);
-            try {
-                threadPool.schedule(TimeValue.timeValueMillis(delayMillis), Names.GENERIC, runnable);
-            } catch (EsRejectedExecutionException e) {
-                if (e.isExecutorShutdown()) {
-                    logger.debug("couldn't schedule next election, executor is shutting down", e);
-                } else {
-                    throw e;
-                }
-            }
+            threadPool.scheduleUnlessShuttingDown(TimeValue.timeValueMillis(delayMillis), Names.GENERIC, runnable);
         }
 
         @Override

server/src/main/java/org/elasticsearch/discovery/PeerFinder.java

@@ -34,7 +34,6 @@
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
-import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.threadpool.ThreadPool.Names;
 import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportRequestOptions;
@@ -249,41 +248,33 @@ private boolean handleWakeUp() {
             }
         });
 
-        try {
-            transportService.getThreadPool().schedule(findPeersInterval, Names.GENERIC, new AbstractRunnable() {
-                @Override
-                public boolean isForceExecution() {
-                    return true;
-                }
+        transportService.getThreadPool().scheduleUnlessShuttingDown(findPeersInterval, Names.GENERIC, new AbstractRunnable() {
+            @Override
+            public boolean isForceExecution() {
+                return true;
+            }
 
-                @Override
-                public void onFailure(Exception e) {
-                    assert false : e;
-                    logger.debug("unexpected exception in wakeup", e);
-                }
+            @Override
+            public void onFailure(Exception e) {
+                assert false : e;
+                logger.debug("unexpected exception in wakeup", e);
+            }
 
-                @Override
-                protected void doRun() {
-                    synchronized (mutex) {
-                        if (handleWakeUp() == false) {
-                            return;
-                        }
+            @Override
+            protected void doRun() {
+                synchronized (mutex) {
+                    if (handleWakeUp() == false) {
+                        return;
                     }
-                    onFoundPeersUpdated();
                 }
+                onFoundPeersUpdated();
+            }
 
-                @Override
-                public String toString() {
-                    return "PeerFinder handling wakeup";
-                }
-            });
-        } catch (EsRejectedExecutionException e) {
-            if (e.isExecutorShutdown()) {
-                logger.debug("couldn't schedule new execution of peer finder, executor is shutting down", e);
-            } else {
-                throw e;
+            @Override
+            public String toString() {
+                return "PeerFinder handling wakeup";
             }
-        }
+        });
 
         return peersRemoved;
     }

server/src/main/java/org/elasticsearch/threadpool/ThreadPool.java

@@ -32,6 +32,7 @@
 import org.elasticsearch.common.unit.SizeValue;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
+import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.concurrent.XRejectedExecutionHandler;
@@ -350,6 +351,19 @@ public ScheduledFuture<?> schedule(TimeValue delay, String executor, Runnable co
         return scheduler.schedule(new ThreadPool.LoggingRunnable(command), delay.millis(), TimeUnit.MILLISECONDS);
     }
 
+    public void scheduleUnlessShuttingDown(TimeValue delay, String executor, Runnable command) {
+        try {
+            schedule(delay, executor, command);
+        } catch (EsRejectedExecutionException e) {
+            if (e.isExecutorShutdown()) {
+                logger.debug(new ParameterizedMessage("could not schedule execution of [{}] after [{}] on [{}] as executor is shut down",
+                    command, delay, executor), e);
+            } else {
+                throw e;
+            }
+        }
+    }
+
     @Override
     public Cancellable scheduleWithFixedDelay(Runnable command, TimeValue interval, String executor) {
         return new ReschedulingRunnable(command, interval, executor, this,