[CARMEL-6087] User is continually blocked by circuit breaker (#1009)

Author: allenma

core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala

@@ -29,6 +29,8 @@ import scala.collection.JavaConverters._
 import scala.collection.mutable.{ArrayBuffer, Buffer, HashMap}
 import scala.util.Random
 
+import com.google.common.cache.{Cache, CacheBuilder}
+
 import org.apache.spark._
 import org.apache.spark.TaskState.TaskState
 import org.apache.spark.executor.ExecutorMetrics
@@ -199,6 +201,9 @@ private[spark] class TaskSchedulerImpl(
     getTimeAsSeconds("spark.workload.circuit.breaker.user.duration", "2h")
   private val taskSetHealthChecker =
     ThreadUtils.newDaemonSingleThreadScheduledExecutor("taskset-health-checker")
+  // store all block user queries and blockTime map
+  private val blockUserQueries: Cache[String, java.lang.Long] = CacheBuilder.newBuilder()
+    .maximumSize(10000).build[String, java.lang.Long]()
 
   private val FURTHER_SCHEDULE_LOCK_WAIT_TIME = conf.
     getTimeAsMs("spark.task.furtherScheduledLockWaitTime", "5s")
@@ -1143,14 +1148,16 @@ private[spark] class TaskSchedulerImpl(
             s"exceeded the max value!"
           dagScheduler.cancelStage(taskSet.stageId, Some(reason))
 
-          if (taskSet.userInfo.isDefined) {
+          if (taskSet.userInfo.isDefined && !blockUserQueries.asMap.containsKey(groupId)) {
             val message = Some(s"Heavy query detected from session[${sessionId}], " +
               s"query[$groupId], check more details: " +
               s"http://viewpoint.hermes-prod.svc.25.tess.io/?session=${sessionId}, " +
               s"root cause: ")
             dagScheduler.sc.workloadCircuitBreaker.
               foreach(_.blockUser(sessionType, taskSet.userInfo().get.user,
                 WORKLOAD_CIRCUIT_BREAKER_USER_DURATION_S, Some(message + reason)))
+            // make sure that same query won't block user multiple times
+            blockUserQueries.put(groupId, System.currentTimeMillis())
           }
         }
       }