[CARMEL-5855]Compact Transparency - Other DDL concurrent conflict (#878)

* [CARMEL-5855]Compact Transparency - Other DDL concurrent conflict

Author: hongyzhang

core/src/main/scala/org/apache/spark/internal/io/SparkHadoopWriterUtils.scala

@@ -27,7 +27,7 @@ import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileSystem, FSDataOutputStream, Path}
 import org.apache.hadoop.mapred.{JobConf, JobID}
 
-import org.apache.spark.{SparkConf, TaskContext}
+import org.apache.spark.{CarmelConcurrentModifiedException, SparkConf, TaskContext}
 import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.executor.OutputMetrics
 import org.apache.spark.internal.Logging
@@ -102,12 +102,29 @@ object SparkHadoopWriterUtils extends Logging {
     new Path(parent, s"_WRITING_$diff")
   }
 
+  def withLock[T](fs: FileSystem, path: Path, waitTime: Int, info: String)(f: => T): T = {
+    var lockFileOut: Option[FSDataOutputStream] = None
+    try {
+      lockFileOut = tryToLockFile(fs,
+        path, true, waitTime, info)
+      if (waitTime < 0 || lockFileOut.nonEmpty) {
+        f
+      } else {
+        throw CarmelConcurrentModifiedException(path)
+      }
+    } finally {
+      if (waitTime >= 0 && lockFileOut.nonEmpty) {
+        releaseLockFile(fs, lockFileOut.get, path)
+      }
+    }
+  }
+
   def tryToLockFile(fs: FileSystem, path: Path, retry: Boolean,
                     waitTime: Int, info: String = ""): Option[FSDataOutputStream] = {
     if (waitTime < 0) return null
     val startTime = System.currentTimeMillis()
     log.info(s"[compact info]$info try to lock path $path")
-    val loc = tryToLockFileInternal(fs, path, retry, waitTime)
+    val loc = tryToLockFileInternal(fs, path, retry, waitTime, info)
     val durationMS = System.currentTimeMillis() - startTime
     if (loc.isDefined) {
       log.info(s"[compact info]$info got lock in $durationMS ms for path $path")
@@ -125,11 +142,15 @@ object SparkHadoopWriterUtils extends Logging {
   }
 
   private def tryToLockFileInternal(fs: FileSystem, path: Path, retry: Boolean,
-                            waitTime: Int): Option[FSDataOutputStream] = {
+                                    waitTime: Int, info: String): Option[FSDataOutputStream] = {
     if (waitTime < 0) return null
     val targetPath = getLockFilePath(path)
     val sleepTime = 1000
-    var retryNum = waitTime / sleepTime + 1
+    var retryNum = if (retry) {
+      waitTime / sleepTime + 1
+    } else {
+      1
+    }
     var ret: Option[FSDataOutputStream] = None
     while (retryNum > 0) {
       try {
@@ -142,11 +163,13 @@ object SparkHadoopWriterUtils extends Logging {
         }
       } catch {
         case ex: Throwable =>
-          logError(s"Failed to lock file $path in $retryNum attempt: ${ex.getMessage}")
+          logError(s"[compact info]$info Failed to lock file $path in $retryNum" +
+            s" attempt: ${ex.getMessage}")
       }
 
       if (retryNum > 1) {
-        logWarning(s"Failed to lock file for $targetPath, ${retryNum - 1} retries left, " +
+        logWarning(s"[compact info]$info Failed to lock file for $targetPath," +
+          s" ${retryNum - 1} retries left, " +
           s"sleep for $sleepTime ms")
         Thread.sleep(sleepTime)
       }