Merge pull request #12599 from jpountz/fix/PrioritizedEsThreadPoolExecutor_concurrency

Fix concurrency issue in PrioritizedEsThreadPoolExecutor.

core/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedEsThreadPoolExecutor.java

@@ -19,6 +19,7 @@
 package org.elasticsearch.common.util.concurrent;
 
 import com.google.common.collect.Lists;
+
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.unit.TimeValue;
 
@@ -161,8 +162,10 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor {
 
         private Runnable runnable;
         private final long insertionOrder;
-        private volatile ScheduledFuture<?> timeoutFuture;
+
-        private volatile boolean started = false;
+        // these two variables are protected by 'this'
+        private ScheduledFuture<?> timeoutFuture;
+        private boolean started = false;
 
         TieBreakingPrioritizedRunnable(PrioritizedRunnable runnable, long insertionOrder) {
             this(runnable, runnable.priority(), insertionOrder);
@@ -176,10 +179,12 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor {
 
         @Override
         public void run() {
+            synchronized (this) {
                 // make the task as stared. This is needed for synchronization with the timeout handling
                 // see  #scheduleTimeout()
                 started = true;
                 FutureUtils.cancel(timeoutFuture);
+            }
             runAndClean(runnable);
         }
 
@@ -193,6 +198,11 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor {
         }
 
         public void scheduleTimeout(ScheduledExecutorService timer, final Runnable timeoutCallback, TimeValue timeValue) {
+            synchronized (this) {
+                if (timeoutFuture != null) {
+                    throw new IllegalStateException("scheduleTimeout may only be called once");
+                }
+                if (started == false) {
                     timeoutFuture = timer.schedule(new Runnable() {
                         @Override
                         public void run() {
@@ -201,9 +211,7 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor {
                             }
                         }
                     }, timeValue.nanos(), TimeUnit.NANOSECONDS);
-            if (started) {
+                }
-                // if the actual action already it might have missed the setting of the future. Clean it ourselves.
-                FutureUtils.cancel(timeoutFuture);
             }
         }