problems with my code that stimulate task failure - task is not resubmitted

[Gil Vernik <GI...@il.ibm.com>]

I have some code that stimulates task failure in the speculative mode.
The code i compile to jar and execute with 

./bin/spark-submit --class com.test.SparkTest --jars --driver-memory 2g 
--executor-memory 1g --master local[4] --conf spark.speculation=true 
--conf spark.task.maxFailures=4 SparkTest.jar 

In addition /conf/spark-defaults.sh contains

spark.speculation = true
spark.task.maxFailures = 4

However when i run my code, i don't see any speculate mode working and 
even worst - the failed task is not resubmitted at all and job fails only 
after 1 unsuccessful task.
Attached both the exception and the code. Can someone explain me what i do 
wrong? 

I need help to make
- Job to fail only after 4 attempts to resubmit the task and not after 1 
failed task.
- Speculative mode to function

Here is the exception:
********************
16/03/05 19:32:51 ERROR Executor: Exception in task 0.0 in stage 0.0 (TID 
0)
org.apache.spark.SparkException: Task failed while writing rows.
        at 
org.apache.spark.sql.execution.datasources.DefaultWriterContainer.writeRows(WriterContainer.scala:266)
        at 
org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1$$anonfun$apply$mcV$sp$3.apply(InsertIntoHadoopFsRelation.scala:151)
        at 
org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1$$anonfun$apply$mcV$sp$3.apply(InsertIntoHadoopFsRelation.scala:151)
        at 
org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:69)
        at org.apache.spark.scheduler.Task.run(Task.scala:81)
        at 
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
        at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.Exception: new exception
        at com.test.SparkTest$$anonfun$1.apply(SparkTest.scala:27)
        at com.test.SparkTest$$anonfun$1.apply(SparkTest.scala:21)
        at 
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown 
Source)
        at 
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
        at 
org.apache.spark.sql.execution.WholeStageCodegen$$anonfun$5$$anon$1.hasNext(WholeStageCodegen.scala:285)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:369)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:369)
        at 
org.apache.spark.sql.execution.datasources.DefaultWriterContainer.writeRows(WriterContainer.scala:256)
        ... 8 more
16/03/05 19:32:51 WARN TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0, 
localhost): org.apache.spark.SparkException: Task failed while writing 
rows.
        at 
org.apache.spark.sql.execution.datasources.DefaultWriterContainer.writeRows(WriterContainer.scala:266)
        at 
org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1$$anonfun$apply$mcV$sp$3.apply(InsertIntoHadoopFsRelation.scala:151)
        at 
org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1$$anonfun$apply$mcV$sp$3.apply(InsertIntoHadoopFsRelation.scala:151)
        at 
org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:69)
        at org.apache.spark.scheduler.Task.run(Task.scala:81)
        at 
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
        at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.Exception: new exception
        at com.test.SparkTest$$anonfun$1.apply(SparkTest.scala:27)
        at com.test.SparkTest$$anonfun$1.apply(SparkTest.scala:21)
        at 
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown 
Source)
        at 
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
        at 
org.apache.spark.sql.execution.WholeStageCodegen$$anonfun$5$$anon$1.hasNext(WholeStageCodegen.scala:285)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:369)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:369)
        at 
org.apache.spark.sql.execution.datasources.DefaultWriterContainer.writeRows(WriterContainer.scala:256)
        ... 8 more

16/03/05 19:32:51 ERROR TaskSetManager: Task 0 in stage 0.0 failed 1 
times; aborting job
16/03/05 19:32:51 ERROR InsertIntoHadoopFsRelation: Aborting job.
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 
in stage 0.0 failed 1 times, most recent failure: Lost task 0.0 in stage 
0.0 (TID 0, localhost): org.apache.spark.SparkException: Task failed while 
writing rows.

Here is the code
****************
  val sparkConf = new SparkConf().setAppName("SparkTest")
  val sc = new SparkContext(sparkConf)
  val sqlContext = new SQLContext(sc)
  import sqlContext.implicits._

  val failSpeculativeTask = sqlContext.udf.register("failSpeculativeTask", 
(i: Int, partitionId: Int, attemptNumber: Int) => {
    if (partitionId == 0 && i == 5) {
      if (attemptNumber == 0) {
        Thread.sleep(15000)
        throw new Exception("new exception")
      } else {
        Thread.sleep(10000)
      }
    }
    i
  })

  val df = sc.parallelize((1 to 100), 20).mapPartitions { iter =>
    val context = org.apache.spark.TaskContext.get()
    val partitionId = context.partitionId
    val attemptNumber = context.attemptNumber
    iter.map(i => (i, partitionId, attemptNumber))
  }.toDF("i", "partitionId", "attemptNumber")

  df.select(failSpeculativeTask($"i", $"partitionId", $"attemptNumber"
).as("i"), $"partitionId", $"attemptNumber")
  .write.mode("overwrite").format("csv").save("/output-directory/")

  }