Spline oct 29414 qa

bundle-2.2/pom.xml

@@ -21,7 +21,7 @@
     <name>Spline Spark Agent Bundle for Spark 2.2</name>
     <parent>
         <groupId>za.co.absa.spline.agent.spark</groupId>
-        <artifactId>spline-spark-agent_2.12</artifactId>
+        <artifactId>spline-spark-agent_2.11</artifactId>
         <relativePath>../pom.xml</relativePath>
         <version>2.3.0-SNAPSHOT</version>
     </parent>

bundle-2.3/pom.xml

@@ -21,7 +21,7 @@
     <name>Spline Spark Agent Bundle for Spark 2.3</name>
     <parent>
         <groupId>za.co.absa.spline.agent.spark</groupId>
-        <artifactId>spline-spark-agent_2.12</artifactId>
+        <artifactId>spline-spark-agent_2.11</artifactId>
         <relativePath>../pom.xml</relativePath>
         <version>2.3.0-SNAPSHOT</version>
     </parent>

bundle-2.4/pom.xml

@@ -16,12 +16,12 @@
   -->
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
-    <artifactId>spark-2.4-spline-agent-bundle_2.12</artifactId>
+    <artifactId>spark-2.4-spline-agent-bundle_2.11</artifactId>
     <packaging>jar</packaging>
     <name>Spline Spark Agent Bundle for Spark 2.4</name>
     <parent>
         <groupId>za.co.absa.spline.agent.spark</groupId>
-        <artifactId>spline-spark-agent_2.12</artifactId>
+        <artifactId>spline-spark-agent_2.11</artifactId>
         <relativePath>../pom.xml</relativePath>
         <version>2.3.0-SNAPSHOT</version>
     </parent>

commons/pom.xml

@@ -18,14 +18,14 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>agent-commons_2.12</artifactId>
+    <artifactId>agent-commons_2.11</artifactId>
     <packaging>jar</packaging>
 
     <name>Spline Spark Agent Commons</name>
 
     <parent>
         <groupId>za.co.absa.spline.agent.spark</groupId>
-        <artifactId>spline-spark-agent_2.12</artifactId>
+        <artifactId>spline-spark-agent_2.11</artifactId>
         <relativePath>../pom.xml</relativePath>
         <version>2.3.0-SNAPSHOT</version>
     </parent>

core/pom.xml

@@ -18,14 +18,14 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>agent-core_2.12</artifactId>
+    <artifactId>agent-core_2.11</artifactId>
     <packaging>jar</packaging>
 
     <name>Spline Spark Agent Core</name>
 
     <parent>
         <groupId>za.co.absa.spline.agent.spark</groupId>
-        <artifactId>spline-spark-agent_2.12</artifactId>
+        <artifactId>spline-spark-agent_2.11</artifactId>
         <relativePath>../pom.xml</relativePath>
         <version>2.3.0-SNAPSHOT</version>
     </parent>

core/src/main/scala/za/co/absa/spline/agent/SplineAgent.scala

@@ -69,7 +69,7 @@ object SplineAgent extends Logging {
     val readCommandExtractor = new PluggableReadCommandExtractor(pluginRegistry, dataSourceFormatResolver)
 
     new SplineAgent {
-      def handle(funcName: FuncName, qe: QueryExecution, result: Either[Throwable, Duration]): Unit = withErrorHandling {
+      def handle(funcName: FuncName, qe: QueryExecution, result: Either[Throwable, Duration]): Unit = withErrorHandling() {
         val idGenerators = new IdGeneratorsBundle(execPlanUUIDGeneratorFactory)
         val harvestingContext = new HarvestingContext(funcName, qe.analyzed, Some(qe.executedPlan), session, idGenerators)
         val postProcessor = new PostProcessor(filters, harvestingContext)
@@ -91,17 +91,21 @@ object SplineAgent extends Logging {
           .harvest(result)
           .foreach({
             case (plan, event) =>
-              lineageDispatcher.send(plan)
-              lineageDispatcher.send(event)
+              withErrorHandling("execution plan dispatch") {
+                lineageDispatcher.send(plan)
+              }
+              withErrorHandling("execution event dispatch") {
+                lineageDispatcher.send(event)
+              }
           })
       }
 
-      private def withErrorHandling(body: => Unit): Unit = {
+      private def withErrorHandling(stage: String = "lineage processing")(body: => Unit): Unit = {
         try body
         catch {
           case NonFatal(e) =>
             val ctx = session.sparkContext
-            logError(s"Unexpected error occurred during lineage processing for application: ${ctx.appName} #${ctx.applicationId}", e)
+            logError(s"Unexpected error occurred during $stage for application: ${ctx.appName} #${ctx.applicationId}", e)
         }
       }
     }

core/src/main/scala/za/co/absa/spline/harvester/dispatcher/CompositeLineageDispatcher.scala

@@ -44,15 +44,19 @@ class CompositeLineageDispatcher(val delegatees: Seq[LineageDispatcher], failOnE
   }
 
   private def delegate(call: LineageDispatcher => Unit): Unit = {
-    delegatees.foreach(withErrorHandling(call))
-  }
+    val failures = delegatees.flatMap { disp =>
+      try {
+        call(disp)
+        None
+      } catch {
+        case NonFatal(e) =>
+          logWarning(s"Proceeding after an error occurred in an underlying dispatcher: ${disp.getClass.getName}", e)
+          Some(e)
+      }
+    }
 
-  private def withErrorHandling(call: LineageDispatcher => Unit): LineageDispatcher => Unit = { disp =>
-    try call(disp)
-    catch {
-      case NonFatal(e) =>
-        if (failOnErrors) throw e
-        else logWarning(s"Proceeding after an error occurred in an underlying dispatcher: ${disp.getClass.getName}", e)
+    if (failOnErrors) {
+      failures.headOption.foreach(e => throw e)
     }
   }
 }

core/src/main/scala/za/co/absa/spline/harvester/dispatcher/FallbackLineageDispatcher.scala

@@ -40,15 +40,23 @@ class FallbackLineageDispatcher(primaryDispatcher: LineageDispatcher, fallbackDi
     catch {
       case e if NonFatal(e) =>
         logError("Error when sending ExecutionPlan, will try to use fallback dispatcher next", e)
-        fallbackDispatcher.send(plan)
+        try fallbackDispatcher.send(plan)
+        catch {
+          case fallbackError if NonFatal(fallbackError) =>
+            logError("Fallback dispatcher also failed when sending ExecutionPlan. Spark job result is not affected.", fallbackError)
+        }
     }
 
   override def send(event: ExecutionEvent): Unit =
     try primaryDispatcher.send(event)
     catch {
       case e if NonFatal(e) =>
         logError("Error when sending ExecutionEvent, will try to use fallback dispatcher next", e)
-        fallbackDispatcher.send(event)
+        try fallbackDispatcher.send(event)
+        catch {
+          case fallbackError if NonFatal(fallbackError) =>
+            logError("Fallback dispatcher also failed when sending ExecutionEvent. Spark job result is not affected.", fallbackError)
+        }
     }
 }
 

core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala

@@ -30,17 +30,11 @@
 import za.co.absa.spline.producer.model.{ExecutionEvent, ExecutionPlan}
 
 import java.net.URI
+import java.nio.charset.StandardCharsets
+import java.util.UUID
 import scala.concurrent.blocking
+import scala.util.control.NonFatal
 
-/**
- * A port of https://github.com/AbsaOSS/spline/tree/release/0.3.9/persistence/hdfs/src/main/scala/za/co/absa/spline/persistence/hdfs
- *
- * Note:
- * This class is unstable, experimental, is mostly used for debugging, with no guarantee to work properly
- * for every generic use case in a real production application.
- *
- * It is NOT thread-safe, strictly synchronous assuming a predefined order of method calls: `send(plan)` and then `send(event)`
- */
 @Experimental
 class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSize: Int)
   extends LineageDispatcher
@@ -62,39 +56,158 @@
   }
 
   override def send(event: ExecutionEvent): Unit = {
-    // check state
     if (this._lastSeenPlan == null || this._lastSeenPlan.id.get != event.planId)
       throw new IllegalStateException("send(event) must be called strictly after send(plan) method with matching plan ID")
 
     try {
-      val path = s"${this._lastSeenPlan.operations.write.outputSource.stripSuffix("/")}/$filename"
+      val sparkContext = SparkContext.getOrCreate()
+
+      val lineageBaseDir = sparkContext.getConf.get(
+        "spark.spline.lineageDispatcher.hdfs.directory",
+        "file:///C:/tmp" // Default to local storage on Windows
+      )
+
+      val executionPlanID = this._lastSeenPlan.id.getOrElse("unknown_plan")
+      val executionPlanName = this._lastSeenPlan.name.replaceAll("[^a-zA-Z0-9_\\-]", "_")
+      val runID = event.extra.get("appId").map(_.toString).getOrElse("unknown_runID")
+
+      val appDir = s"$lineageBaseDir/$executionPlanName"
+      val runDir = s"$appDir/$runID"
+
+      // Best-effort cleanup of previous runs
+      cleanupOldRunFolders(appDir, runID)
+
+      val fileName = s"lineage_${executionPlanID}.json"
+
+      // Build JSON payload
       val planWithEvent = Map(
         "executionPlan" -> this._lastSeenPlan,
         "executionEvent" -> event
       )
-
       import HarvesterJsonSerDe.impl._
-      persistToHadoopFs(planWithEvent.toJson, path)
+      val content = planWithEvent.toJson
+
+      // Atomically materialize the run folder & file
+      persistRunFolderAtomically(runDir, fileName, content)
     } finally {
       this._lastSeenPlan = null
     }
   }
 
-  private def persistToHadoopFs(content: String, fullLineagePath: String): Unit = blocking {
-    val (fs, path) = pathStringToFsWithPath(fullLineagePath)
-    logDebug(s"Opening HadoopFs output stream to $path")
+  /**
+   * Deletes all runID folders except the current one
+   * @param appDirPath Path to the app directory
+   * @param currentRunID The current runID to keep
+   */
+  private def cleanupOldRunFolders(appDirPath: String, currentRunID: String): Unit = {
+    try {
+      val (fs, appPath) = pathStringToFsWithPath(appDirPath)
+
+      if (fs.exists(appPath) && fs.getFileStatus(appPath).isDirectory) {
+        val statuses = fs.listStatus(appPath)
+        statuses.foreach { status =>
+          if (status.isDirectory && status.getPath.getName != currentRunID) {
+            try {
+              logInfo(s"Deleting old runID folder: ${status.getPath}")
+              fs.delete(status.getPath, true) // recursive delete
+            } catch {
+              case NonFatal(e) =>
+                logWarning(s"Failed to delete old runID folder: ${status.getPath}", e)
+            }
+          }
+        }
+      }
+    } catch {
+      case NonFatal(e) =>
+        logWarning(s"Failed to cleanup old run folders in $appDirPath", e)
+    }
+  }
 
-    val replication = fs.getDefaultReplication(path)
-    val blockSize = fs.getDefaultBlockSize(path)
-    val outputStream = fs.create(path, permission, true, bufferSize, replication, blockSize, null)
+  /**
+   * Atomically create a run directory with its lineage JSON file inside.
+   * Strategy:
+   *  1) Ensure parent exists (mkdirs is idempotent)
+   *  2) Create a hidden temp directory next to the target
+   *  3) Write JSON to a temp file, then rename temp file -> final name inside temp dir
+   *  4) Rename temp dir -> final run dir (atomic on HDFS)
+   */
+  private def persistRunFolderAtomically(finalRunDirStr: String, fileName: String, content: String): Unit = blocking {
+    val (fs, finalRunDir) = pathStringToFsWithPath(finalRunDirStr)
 
-    val umask = FsPermission.getUMask(fs.getConf)
-    FsPermission.getFileDefault.applyUMask(umask)
+    // Ensure final run directory exists
+    if (!fs.exists(finalRunDir)) {
+      fs.mkdirs(finalRunDir)
+      try fs.setPermission(finalRunDir, permission) catch { case _: Throwable => () }
+    }
 
-    logDebug(s"Writing lineage to $path")
-    using(outputStream) {
-      _.write(content.getBytes("UTF-8"))
+    // Write the file atomically inside the run directory
+    val finalFileInRunDir = new Path(finalRunDir, fileName)
+    writeFileAtomically(fs, finalFileInRunDir, content.getBytes(StandardCharsets.UTF_8))
+  }
+
+  /**
+   * Write a single file atomically via temp-sibling + rename.
+   * If the final file already exists, it is removed first so repeated attempts
+   * for the same Spark application can refresh the lineage payload.
+   * The rename is retried up to [[RenameRetries]] times with a short delay to handle
+   * transient HDFS failures (e.g. a bad datanode causing a slow or failed pipeline ack).
+   */
+  private def writeFileAtomically(fs: FileSystem, finalPath: Path, bytes: Array[Byte]): Unit = {
+    val parent = finalPath.getParent
+    if (!fs.exists(parent)) {
+      fs.mkdirs(parent)
+    }
+
+    val tmpFile = new Path(parent, s".${finalPath.getName}.tmp-${UUID.randomUUID().toString}")
+    val replication = fs.getDefaultReplication(finalPath)
+    val blockSize = fs.getDefaultBlockSize(finalPath)
+
+    logDebug(s"Creating temp file $tmpFile")
+    val out = fs.create(tmpFile, permission, true, bufferSize, replication, blockSize, null)
+    try {
+      out.write(bytes)
+      // Best-effort durability hints; on HDFS these are meaningful.
+      out.hflush()
+      out.hsync()
+    } finally {
+      out.close()
+    }
+
+    try fs.setPermission(tmpFile, permission) catch { case _: Throwable => () }
+
+    if (fs.exists(finalPath) && !fs.delete(finalPath, false)) {
+      try fs.delete(tmpFile, false) catch { case _: Throwable => () }
+      throw new RuntimeException(s"Failed to delete existing lineage file before rewrite: $finalPath")
     }
+
+    logDebug(s"Renaming $tmpFile -> $finalPath (atomic on HDFS)")
+
+    @scala.annotation.tailrec
+    def renameWithRetry(attemptsLeft: Int): Unit = {
+      if (fs.rename(tmpFile, finalPath)) {
+        logDebug(s"Rename succeeded: $tmpFile -> $finalPath")
+      } else if (attemptsLeft > 0) {
+        logWarning(s"Rename failed for $tmpFile -> $finalPath, retrying in ${RenameRetryDelayMs}ms ($attemptsLeft attempt(s) left)")
+        Thread.sleep(RenameRetryDelayMs)
+        renameWithRetry(attemptsLeft - 1)
+      } else {
+        try fs.delete(tmpFile, false) catch { case _: Throwable => () }
+        throw new RuntimeException(
+          s"Failed to atomically rename $tmpFile to $finalPath after ${RenameRetries + 1} attempt(s)")
+      }
+    }
+
+    renameWithRetry(RenameRetries)
+  }
+
+  // Kept for compatibility: atomic-at-file-level write for a direct full path
+  private def persistToHadoopFs(content: String, fullLineagePath: String): Unit = blocking {
+    val (fs, path) = pathStringToFsWithPath(fullLineagePath)
+    val parentDir = path.getParent
+    if (!fs.exists(parentDir)) {
+      fs.mkdirs(parentDir)
+    }
+    writeFileAtomically(fs, path, content.getBytes(StandardCharsets.UTF_8))
   }
 }
 
@@ -105,12 +218,14 @@
   private val FilePermissionsKey = "filePermissions"
   private val BufferSizeKey = "fileBufferSize"
 
+  private val RenameRetries = 3
+  private val RenameRetryDelayMs = 1000L
+
   /**
    * Converts string full path to Hadoop FS and Path, e.g.
    * `s3://mybucket1/path/to/file` -> S3 FS + `path/to/file`
-   * `/path/on/hdfs/to/file` -> local HDFS + `/path/on/hdfs/to/file`
-   *
-   * Note, that non-local HDFS paths are not supported in this method, e.g. hdfs://nameservice123:8020/path/on/hdfs/too.
+   * `file:///path/to/file` -> local FS + `/path/to/file`
+   * `/path/on/hdfs/to/file` -> HDFS + `/path/on/hdfs/to/file`
    *
    * @param pathString path to convert to FS and relative path
    * @return FS + relative path
@@ -119,14 +234,21 @@
     pathString.toSimpleS3Location match {
       case Some(s3Location) =>
         val s3Uri = new URI(s3Location.asSimpleS3LocationString) // s3://<bucket>
-        val s3Path = new Path(s"/${s3Location.path}") // /<text-file-object-path>
-
+        val s3Path = new Path(s"/${s3Location.path}")            // /<text-file-object-path>
         val fs = FileSystem.get(s3Uri, HadoopConfiguration)
         (fs, s3Path)
 
-      case None => // local hdfs location
-        val fs = FileSystem.get(HadoopConfiguration)
-        (fs, new Path(pathString))
+      case None =>
+        // Check if it's an explicit file:// URI
+        if (pathString.startsWith("file://") || pathString.startsWith("file:/")) {
+          val uri = new URI(pathString)
+          val fs = FileSystem.get(uri, HadoopConfiguration)
+          (fs, new Path(uri.getPath))
+        } else {
+          // Default HDFS location
+          val fs = FileSystem.get(HadoopConfiguration)
+          (fs, new Path(pathString))
+        }
     }
   }
 }