Spark Scheduler encounters two independent Deadlocks when trying to kill executors either due to dynamic allocation or blacklisting

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}

import org.apache.spark._
import org.apache.spark.TaskContext

// Simple example of Word Count in Scala
object ScalaWordCount {
def main(args: Array[String]) {

if (args.length < 2) {
System.err.println("Usage: ScalaWordCount <inputFilesURI> <outputFilesUri>")
System.exit(1)
}

val conf = new SparkConf().setAppName("Scala Word Count")
val sc = new SparkContext(conf)

// get the input file uri
val inputFilesUri = args(0)

// get the output file uri
val outputFilesUri = args(1)

while (true) {
val textFile = sc.textFile(inputFilesUri)
val counts = textFile.flatMap(line => line.split(" "))
.map(word => {if (TaskContext.get.partitionId == 5 && TaskContext.get.attemptNumber == 0) throw new Exception("Fail for blacklisting") else (word, 1)})
.reduceByKey(_ + _)
counts.saveAsTextFile(outputFilesUri)
val conf: Configuration = new Configuration()
val path: Path = new Path(outputFilesUri)
val hdfs: FileSystem = FileSystem.get(conf)
hdfs.delete(path, true)
}

sc.stop()
}
}

executorIdToFailureList.remove(exec)
updateNextExpiryTime()
Thread.sleep(2000)
killBlacklistedExecutor(exec)