Automatic staging committer conflict-mode for dynamic partition overwrite

    Following up with previous commit, which disabled hard-coded exception
    in PathOutputCommitProtocol.scala when using dynamicPartitionOverwrite.
    Spark 3.5 upgrade throws that exception in another location too,
    so this commit disables that as well.

    We also update unit test in CommitterBindingSuite.scala to disable
    IOException throwing for unsupported dynamic partitioning.

Author: anuvedverma

hadoop-cloud/src/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala

@@ -130,7 +130,9 @@ class PathOutputCommitProtocol(
           logDebug(
             s"Committer $committer has declared compatibility with dynamic partition overwrite")
         } else {
-          throw new IOException(PathOutputCommitProtocol.UNSUPPORTED + ": " + committer)
+          // throw new IOException(PathOutputCommitProtocol.UNSUPPORTED + ": " + committer)
+          // The above exception is disabled with automatic value of
+          // fs.s3a.committer.staging.conflict-mode in HadoopMapReduceCommitProtocol.
         }
       }
     }

hadoop-cloud/src/test/scala/org/apache/spark/internal/io/cloud/CommitterBindingSuite.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.internal.io.cloud
 
-import java.io.{File, FileInputStream, FileOutputStream, IOException, ObjectInputStream, ObjectOutputStream}
+import java.io.{File, FileInputStream, FileOutputStream, ObjectInputStream, ObjectOutputStream}
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{Path, StreamCapabilities}
@@ -148,6 +148,9 @@ class CommitterBindingSuite extends SparkFunSuite {
    * Bind a job to a committer which doesn't support dynamic partitioning.
    * Job setup must fail, and calling `newTaskTempFileAbsPath()` must
    * raise `UnsupportedOperationException`.
+   * With custom patch https://github.com/lyft/spark/pull/34:
+   *  Job setup should now succeed (due to custom patch), but calling
+   *  `newTaskTempFileAbsPath()` must still raise `UnsupportedOperationException`.
    */
   test("reject dynamic partitioning if not supported") {
     val path = new Path("http://example/data")
@@ -162,14 +165,13 @@ class CommitterBindingSuite extends SparkFunSuite {
       jobId,
       path.toUri.toString,
       true)
-    val ioe = intercept[IOException] {
-      committer.setupJob(tContext)
-    }
-    if (!ioe.getMessage.contains(PathOutputCommitProtocol.UNSUPPORTED)) {
-      throw ioe
-    }
 
-    // calls to newTaskTempFileAbsPath() will be rejected
+    // Job setup should now succeed due to the custom patch that disabled
+    // the IOException throwing for unsupported dynamic partitioning
+    committer.setupJob(tContext)
+    committer.setupTask(tContext)
+
+    // calls to newTaskTempFileAbsPath() will still be rejected
     intercept[UnsupportedOperationException] {
       verifyAbsTempFileWorks(tContext, committer)
     }