From e32e68485eaf0ed9eed7d88478154aff8650da62 Mon Sep 17 00:00:00 2001
From: xutingjun <xutingjun@huawei.com>
Date: Tue, 27 Oct 2015 11:14:56 +0800
Subject: [PATCH] numRunningTasks can't be less than 0, or it will affect
 executor allocation

---
 .../org/apache/spark/ExecutorAllocationManager.scala      | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
index b93536e6536e..e69b22d24a30 100644
--- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
+++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
@@ -575,7 +575,7 @@ private[spark] class ExecutorAllocationManager(
         if (stageIdToNumTasks.isEmpty) {
           allocationManager.onSchedulerQueueEmpty()
           if (numRunningTasks != 0) {
-            logWarning("No stages are running, but numRunningTasks != 0")
+            logWarning(s"No stages are running, but numRunningTasks = $numRunningTasks")
             numRunningTasks = 0
           }
         }
@@ -615,7 +615,11 @@ private[spark] class ExecutorAllocationManager(
       val taskIndex = taskEnd.taskInfo.index
       val stageId = taskEnd.stageId
       allocationManager.synchronized {
-        numRunningTasks -= 1
+        if (numRunningTasks > 0) {
+          numRunningTasks -= 1
+        } else {
+          logWarning("SparkListenerTaskEnd got, but numRunningTasks == 0")
+        }
         // If the executor is no longer running any scheduled tasks, mark it as idle
         if (executorIdToTaskIds.contains(executorId)) {
           executorIdToTaskIds(executorId) -= taskId