Re-enable guarding of commit coordination with spark.speculation sett…

…ing.
apache · Feb 4, 2015 · 3969f5f · 3969f5f
1 parent ede7590
commit 3969f5f
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 7 deletions.
diff --git a/core/src/main/scala/org/apache/spark/SparkHadoopWriter.scala b/core/src/main/scala/org/apache/spark/SparkHadoopWriter.scala
@@ -131,12 +131,7 @@ class SparkHadoopWriter(@transient jobConf: JobConf)
         // attempts, which should only occur if speculation is enabled
         val speculationEnabled = sparkConf.getBoolean("spark.speculation", false)
         // This (undocumented) setting is an escape-hatch in case the commit code introduces bugs
-        // sparkConf.getBoolean("spark.hadoop.outputCommitCoordination.enabled", speculationEnabled)
-
-        // TODO: revert this before merging the PR; this is enabled so this code path is exercised
-        // by more tests (even though it might not be _necessary_, it should be _safe_ to do the
-        // extra coordination)
-        true
+        sparkConf.getBoolean("spark.hadoop.outputCommitCoordination.enabled", speculationEnabled)
       }
       if (shouldCoordinateWithDriver) {
         val outputCommitCoordinator = SparkEnv.get.outputCommitCoordinator

diff --git a/core/src/test/scala/org/apache/spark/scheduler/OutputCommitCoordinatorSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/OutputCommitCoordinatorSuite.scala
@@ -72,7 +72,11 @@ class OutputCommitCoordinatorSuite extends FunSuite with BeforeAndAfter {
 
   before {
     tempDir = Utils.createTempDir()
-    sc = new SparkContext("local[4]", classOf[OutputCommitCoordinatorSuite].getSimpleName) {
+    val conf = new SparkConf()
+      .setMaster("local[4]")
+      .setAppName(classOf[OutputCommitCoordinatorSuite].getSimpleName)
+      .set("spark.speculation", "true")
+    sc = new SparkContext(conf) {
       override private[spark] def createSparkEnv(
         conf: SparkConf,
         isLocal: Boolean,