From 3e2ce9b0b6d3cfc703585742b402f853f75f2771 Mon Sep 17 00:00:00 2001
From: Yannick Welsch <yannick@welsch.lu>
Date: Fri, 21 Apr 2017 11:19:40 +0200
Subject: [PATCH] Don't expose cleaned-up tasks as pending in
 PrioritizedEsThreadPoolExecutor

---
 .../util/concurrent/PrioritizedEsThreadPoolExecutor.java  | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/core/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedEsThreadPoolExecutor.java b/core/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedEsThreadPoolExecutor.java
index 5b3dae7ffae71..ee38637b04c0c 100644
--- a/core/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedEsThreadPoolExecutor.java
+++ b/core/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedEsThreadPoolExecutor.java
@@ -91,7 +91,13 @@ private void addPending(List<Runnable> runnables, List<Pending> pending, boolean
         for (Runnable runnable : runnables) {
             if (runnable instanceof TieBreakingPrioritizedRunnable) {
                 TieBreakingPrioritizedRunnable t = (TieBreakingPrioritizedRunnable) runnable;
-                pending.add(new Pending(unwrap(t.runnable), t.priority(), t.insertionOrder, executing));
+                Runnable innerRunnable = t.runnable;
+                if (innerRunnable != null) {
+                    /** innerRunnable can be null if task is finished but not removed from executor yet,
+                     * see {@link TieBreakingPrioritizedRunnable#run} and {@link TieBreakingPrioritizedRunnable#runAndClean}
+                     */
+                    pending.add(new Pending(unwrap(innerRunnable), t.priority(), t.insertionOrder, executing));
+                }
             } else if (runnable instanceof PrioritizedFutureTask) {
                 PrioritizedFutureTask t = (PrioritizedFutureTask) runnable;
                 Object task = t.task;