From 8d49e2f1b2f04741228edb48e9c7947e80ff3bdd Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Sun, 12 Oct 2025 19:12:25 +0100
Subject: [PATCH 01/25] Added initial implementation for writing lineage
 messages using HDFS Lineage Dispatcher to a centralised location

---
 core/src/main/resources/spline.default.yaml   |  9 ++++
 .../dispatcher/HDFSLineageDispatcher.scala    | 47 +++++++++++++++++--
 2 files changed, 53 insertions(+), 3 deletions(-)

diff --git a/core/src/main/resources/spline.default.yaml b/core/src/main/resources/spline.default.yaml
index 66eef175..a258f496 100644
--- a/core/src/main/resources/spline.default.yaml
+++ b/core/src/main/resources/spline.default.yaml
@@ -146,6 +146,15 @@ spline:
       fileBufferSize: 4096
       # file/directory permissions, provided as umask string, either in octal or symbolic format
       filePermissions: 777
+      # customLineagePath:
+      # OPTIONAL: Custom path for centralized lineage storage. If specified, all lineage files will be written to this location instead of alongside
+      # the target data files. Filenames will include original filename as prefix, followed by timestamp to ensure uniqueness and chronological ordering.
+      # Since each execution plan produces exactly one lineage message, the timestamp alone ensures uniqueness.
+      # Example Output: {fileName}_{timestamp} where filename is location.parq_LINEAGE and timestamp is 1703123456789
+      # Thus, full example includes:
+      # - "/my/unique/location.parq_LINEAGE_1703123456789",
+      # - "s3://my-bucket-locatioon/file.parq_LINEAGE_1703123456789",
+      # - "hdfs://cluster/my/unique/location.parq_LINEAGE_1703123456789"
 
     # -------------------------------------------
     # Open Lineage HTTP dispatcher
diff --git a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
index a52611af..9d0a70a4 100644
--- a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
+++ b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
@@ -40,16 +40,22 @@ import scala.concurrent.blocking
  * 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)`
+ *
+ * When using centralized lineage storage (customLineagePath), filenames are guaranteed to be unique by including:
+ * - Original filename (as prefix)
+ * - Timestamp (epoch milliseconds)
+ * Since each execution plan produces exactly one line message, the timestamp alone ensures uniqueness and provides chronological ordering.
  */
 @Experimental
-class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSize: Int)
+class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSize: Int, customLineagePath: Option[String])
   extends LineageDispatcher
     with Logging {
 
   def this(conf: Configuration) = this(
     filename = conf.getRequiredString(FileNameKey),
     permission = new FsPermission(conf.getRequiredObject(FilePermissionsKey).toString),
-    bufferSize = conf.getRequiredInt(BufferSizeKey)
+    bufferSize = conf.getRequiredInt(BufferSizeKey),
+    customLineagePath = conf.getOptionalString(CustomLineagePathKey)
   )
 
   @volatile
@@ -67,7 +73,7 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
       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 path = resolveLineagePath(event.planId)
       val planWithEvent = Map(
         "executionPlan" -> this._lastSeenPlan,
         "executionEvent" -> event
@@ -80,6 +86,40 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
     }
   }
 
+  /**
+   * Resolves the lineage file path based on configuration.
+   * If customLineagePath is specified, lineage files are written to that centralized location.
+   * Otherwise, lineage files are written alongside the target data file (current behavior).
+   *
+   * @param planId The execution plan ID used for generating unique filenames in centralized mode
+   * @return The full path where the lineage file should be written
+   */
+  private def resolveLineagePath(planId: String): String = {
+    customLineagePath match {
+      case Some(customPath) =>
+        // Centralized mode: write to custom path with unique filename
+        val cleanCustomPath = customPath.stripSuffix("/")
+        val uniqueFilename = generateUniqueFilename(planId)
+        s"$cleanCustomPath/$uniqueFilename"
+      case None =>
+        // Legacy mode: write alongside target data file
+        s"${this._lastSeenPlan.operations.write.outputSource.stripSuffix("/")}/$filename"
+    }
+  }
+
+  /**
+   * Generates a unique filename for centralized lineage storage.
+   * Uses original filename as prefix, followed by timestamp to ensure uniqueness and chronological ordering.
+   * Since each execution plan produces exactly one lineage message, the timestamp alone ensures uniqueness.
+   *
+   * @param planId The execution plan ID (unused, kept for interface compatibility)
+   * @return A unique filename with original filename as prefix, followed by timestamp
+   */
+  private def generateUniqueFilename(planId: String): String = {
+    val timestamp = System.currentTimeMillis()
+    s"${filename}_${timestamp}"
+  }
+
   private def persistToHadoopFs(content: String, fullLineagePath: String): Unit = blocking {
     val (fs, path) = pathStringToFsWithPath(fullLineagePath)
     logDebug(s"Opening HadoopFs output stream to $path")
@@ -104,6 +144,7 @@ object HDFSLineageDispatcher {
   private val FileNameKey = "fileName"
   private val FilePermissionsKey = "filePermissions"
   private val BufferSizeKey = "fileBufferSize"
+  private val CustomLineagePathKey = "customLineagePath"
 
   /**
    * Converts string full path to Hadoop FS and Path, e.g.

From a35a99cf7e962b10510292f51ebf39c7c2781f55 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Sun, 12 Oct 2025 23:25:32 +0100
Subject: [PATCH 02/25] Added creation of the base customLineagePath directory
 ensuring lineage files can be written there

---
 .../dispatcher/HDFSLineageDispatcher.scala    | 99 +++++++++++++++++--
 1 file changed, 89 insertions(+), 10 deletions(-)

diff --git a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
index 9d0a70a4..aaccc340 100644
--- a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
+++ b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
@@ -30,6 +30,8 @@ import za.co.absa.spline.harvester.json.HarvesterJsonSerDe
 import za.co.absa.spline.producer.model.{ExecutionEvent, ExecutionPlan}
 
 import java.net.URI
+import java.time.format.DateTimeFormatter
+import java.time.{Instant, ZoneId}
 import scala.concurrent.blocking
 
 /**
@@ -41,10 +43,22 @@ import scala.concurrent.blocking
  *
  * It is NOT thread-safe, strictly synchronous assuming a predefined order of method calls: `send(plan)` and then `send(event)`
  *
- * When using centralized lineage storage (customLineagePath), filenames are guaranteed to be unique by including:
- * - Original filename (as prefix)
- * - Timestamp (epoch milliseconds)
- * Since each execution plan produces exactly one line message, the timestamp alone ensures uniqueness and provides chronological ordering.
+ * TWO MODES OF OPERATION:
+ *
+ * 1. DEFAULT MODE (customLineagePath = None, null, or empty string):
+ *    Lineage files are written alongside the target data files.
+ *    Example: Writing to /data/output/file.parquet creates /data/output/_LINEAGE
+ *
+ * 2. CENTRALIZED MODE (customLineagePath set to a valid path):
+ *    All lineage files are written to a single centralized location with unique filenames.
+ *    Filename format: {timestamp}_{fileName}_{appId}
+ *    - timestamp: Human-readable UTC timestamp (yyyy-MM-dd_HH-mm-ss-SSS) for chronological sorting and filtering
+ *    - fileName: The configured fileName value (e.g., "my_file.parq_LINEAGE")
+ *    - appId: Spark application ID for traceability
+ *
+ *    The timestamp-first format ensures natural chronological sorting and easy date-based filtering.
+ *    Parent directories are automatically created with proper permissions for multi-user access (HDFS/local).
+ *    For object storage (S3, GCS, Azure), directory creation is skipped since they use key prefixes.
  */
 @Experimental
 class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSize: Int, customLineagePath: Option[String])
@@ -102,28 +116,93 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
         val uniqueFilename = generateUniqueFilename(planId)
         s"$cleanCustomPath/$uniqueFilename"
       case None =>
-        // Legacy mode: write alongside target data file
+        // Default mode: write alongside target data file
         s"${this._lastSeenPlan.operations.write.outputSource.stripSuffix("/")}/$filename"
     }
   }
 
   /**
    * Generates a unique filename for centralized lineage storage.
-   * Uses original filename as prefix, followed by timestamp to ensure uniqueness and chronological ordering.
-   * Since each execution plan produces exactly one lineage message, the timestamp alone ensures uniqueness.
+   * 
+   * Format: {timestamp}_{fileName}_{appId}
+   * Example: 2025-10-12_14-30-45-123_lineage_app-20251012143045-0001
+   *
+   * This format optimizes for operational debugging use cases:
+   * - Timestamp FIRST: Ensures natural chronological sorting (most recent files appear together)
+   * - Application ID: Full traceability to specific Spark application run
+   * Benefits:
+   * - Natural chronological sorting by default
+   * - Easy date-based filtering
+   * - Human-readable: Immediately see when each lineage file was created
+   * - Millisecond precision for uniqueness
    *
    * @param planId The execution plan ID (unused, kept for interface compatibility)
-   * @return A unique filename with original filename as prefix, followed by timestamp
+   * @return A unique filename optimized for filtering and sorting
    */
   private def generateUniqueFilename(planId: String): String = {
-    val timestamp = System.currentTimeMillis()
-    s"${filename}_${timestamp}"
+    val sparkContext = SparkContext.getOrCreate()
+    val appId = sparkContext.applicationId
+    val dateFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd_HH-mm-ss-SSS").withZone(ZoneId.of("UTC"))
+    val timestamp = dateFormatter.format(Instant.now())
+    s"${timestamp}_${filename}_${appId}"
+  }
+
+  /**
+   * Ensures that parent directories exist with proper permissions for multi-user access.
+   * This is critical for centralized lineage storage where different service accounts need write access.
+   *
+   * Note: Object storage systems (S3, GCS, Azure Blob) don't have true directories - they use key prefixes.
+   * Directory creation is automatically skipped for these systems to avoid unnecessary operations.
+   *
+   * @param fs The Hadoop FileSystem
+   * @param path The target file path whose parent directories should be created
+   */
+  private def ensureParentDirectoriesExist(fs: FileSystem, path: Path): Unit = {
+    // Object storage systems (S3, GCS, Azure Blob) don't have real directories - they're just key prefixes
+    // Skip directory creation for these systems to avoid unnecessary operations
+    val fsScheme = fs.getUri.getScheme
+    val isObjectStorage = fsScheme != null && (
+      fsScheme.startsWith("s3") ||      // S3: s3, s3a, s3n
+      fsScheme.startsWith("gs") ||      // Google Cloud Storage: gs
+      fsScheme.startsWith("wasb") ||    // Azure Blob Storage: wasb, wasbs
+      fsScheme.startsWith("abfs") ||    // Azure Data Lake Storage Gen2: abfs, abfss
+      fsScheme.startsWith("adl")        // Azure Data Lake Storage Gen1: adl
+    )
+    
+    if (isObjectStorage) {
+      logDebug(s"Skipping directory creation for object storage filesystem ($fsScheme) - directories are implicit key prefixes")
+      return
+    }
+
+    val parentDir = path.getParent
+    if (parentDir != null && !fs.exists(parentDir)) {
+      try {
+        // Create directories with multi-user friendly permissions to allow all service accounts to write
+        // This uses the same permission object that's already configured for files
+        val permissions = this.permission.toString
+        val dirPermission = new FsPermission(permissions)
+        val created = fs.mkdirs(parentDir, dirPermission)
+        if (created) {
+          logInfo(s"Created parent directories: $parentDir with permissions $permissions")
+        }
+      } catch {
+        case e: org.apache.hadoop.fs.FileAlreadyExistsException =>
+          // Race condition: another process created the directory - this is fine
+          logDebug(s"Directory $parentDir already exists (created by another process)")
+        case e: Exception =>
+          logWarning(s"Failed to create parent directories: $parentDir", e)
+          throw e
+      }
+    }
   }
 
   private def persistToHadoopFs(content: String, fullLineagePath: String): Unit = blocking {
     val (fs, path) = pathStringToFsWithPath(fullLineagePath)
     logDebug(s"Opening HadoopFs output stream to $path")
 
+    // Ensure parent directories exist with proper permissions for multi-user access
+    ensureParentDirectoriesExist(fs, path)
+
     val replication = fs.getDefaultReplication(path)
     val blockSize = fs.getDefaultBlockSize(path)
     val outputStream = fs.create(path, permission, true, bufferSize, replication, blockSize, null)

From 7a824bf3b45ed7aeea31da7c47ef68be98772c82 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Sun, 12 Oct 2025 23:26:58 +0100
Subject: [PATCH 03/25] Updated documentation for Hdfs Dispatcher in
 spline.default.yaml file

---
 core/src/main/resources/spline.default.yaml | 33 ++++++++++++++-------
 1 file changed, 23 insertions(+), 10 deletions(-)

diff --git a/core/src/main/resources/spline.default.yaml b/core/src/main/resources/spline.default.yaml
index a258f496..2277f656 100644
--- a/core/src/main/resources/spline.default.yaml
+++ b/core/src/main/resources/spline.default.yaml
@@ -146,16 +146,29 @@ spline:
       fileBufferSize: 4096
       # file/directory permissions, provided as umask string, either in octal or symbolic format
       filePermissions: 777
-      # customLineagePath:
-      # OPTIONAL: Custom path for centralized lineage storage. If specified, all lineage files will be written to this location instead of alongside
-      # the target data files. Filenames will include original filename as prefix, followed by timestamp to ensure uniqueness and chronological ordering.
-      # Since each execution plan produces exactly one lineage message, the timestamp alone ensures uniqueness.
-      # Example Output: {fileName}_{timestamp} where filename is location.parq_LINEAGE and timestamp is 1703123456789
-      # Thus, full example includes:
-      # - "/my/unique/location.parq_LINEAGE_1703123456789",
-      # - "s3://my-bucket-locatioon/file.parq_LINEAGE_1703123456789",
-      # - "hdfs://cluster/my/unique/location.parq_LINEAGE_1703123456789"
-
+      customLineagePath:
+      # OPTIONAL: Custom path for centralized lineage storage.
+      # If left empty, null, or not specified → DEFAULT MODE: lineage written alongside target data files
+      # If set to a path → CENTRALIZED MODE: all lineage written to this location with unique filenames
+      #
+      # CENTRALIZED MODE filename format: {timestamp}_{fileName}_{appId}
+      # - timestamp: Human-readable UTC timestamp (yyyy-MM-dd_HH-mm-ss-SSS) for natural chronological sorting and easy filtering
+      #   Example: 2025-10-12_14-30-45-123
+      # - fileName: The configured fileName value (e.g., "my_file.parq_LINEAGE")
+      # - appId: Spark application ID for traceability to specific runs
+      #   Example: app-20251012143045-0001
+      #
+      # More examples:
+      # - Local: customLineagePath: /my/centralized/lineage
+      #   Output: /my/centralized/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
+      # - S3: customLineagePath: s3://my-bucket/lineage
+      #   Output: s3://my-bucket/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
+      # - GCS: customLineagePath: gs://my-bucket/lineage
+      #   Output: gs://my-bucket/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
+      # - Azure: customLineagePath: abfss://container@account.dfs.core.windows.net/lineage
+      #   Output: abfss://container@account.dfs.core.windows.net/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
+      # - HDFS: customLineagePath: hdfs://cluster/lineage
+      #   Output: hdfs://cluster/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
     # -------------------------------------------
     # Open Lineage HTTP dispatcher
     # -------------------------------------------

From 96373b41453a7f31d77137ad21f8de34df4ab0a2 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Sun, 12 Oct 2025 23:36:48 +0100
Subject: [PATCH 04/25] Added initial integration tests for HDFS Lineage
 Dispatcher with centralised lineage write function

---
 .../HDFSLineageDispatcherSpec.scala           | 172 +++++++++++++++++-
 1 file changed, 171 insertions(+), 1 deletion(-)

diff --git a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
index a302fe73..9fafc1dc 100644
--- a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
+++ b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
@@ -37,7 +37,7 @@ class HDFSLineageDispatcherSpec
 
   behavior of "HDFSLineageDispatcher"
 
-  it should "save lineage file to a filesystem" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
+  it should "save lineage file to a filesystem in DEFAULT mode (no customLineagePath)" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
     withIsolatedSparkSession(_
       .config("spark.spline.lineageDispatcher", "hdfs")
       .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
@@ -63,4 +63,174 @@ class HDFSLineageDispatcherSpec
     }
   }
 
+  it should "save lineage file in CENTRALIZED mode with customLineagePath" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
+    val centralizedPath = TempDirectory("spline_centralized_", "", pathOnly = true).deleteOnExit()
+
+    withIsolatedSparkSession(_
+      .config("spark.spline.lineageDispatcher", "hdfs")
+      .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
+      .config("spark.spline.lineageDispatcher.hdfs.customLineagePath", centralizedPath.asString)
+    ) { implicit spark =>
+      withLineageTracking { captor =>
+        import spark.implicits._
+        val dummyDF = Seq((1, 2)).toDF
+        val destPath = TempDirectory("spline_", ".parquet", pathOnly = true).deleteOnExit()
+
+        for {
+          (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
+        } yield {
+          // Lineage should NOT be in the destination directory
+          val lineageFileInDest = new File(destPath.asString, "_LINEAGE")
+          lineageFileInDest.exists should be(false)
+
+          // Lineage should be in centralized directory with timestamp-based filename
+          val centralizedDir = new File(centralizedPath.asString)
+          val lineageFiles = centralizedDir.listFiles()
+          lineageFiles should not be null
+          lineageFiles.length should be(1)
+
+          val lineageFile = lineageFiles(0)
+          lineageFile.length should be > 0L
+
+          // Verify filename format: {timestamp}_{fileName}_{appId}
+          val filename = lineageFile.getName
+          // Should match pattern: yyyy-MM-dd_HH-mm-ss-SSS__LINEAGE_app-...
+          filename should include("_LINEAGE_")
+          filename should startWith regex """\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}-\d{3}"""
+
+          val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
+          lineageJson should contain key "executionPlan"
+          lineageJson should contain key "executionEvent"
+          lineageJson("executionPlan")("id") should equal(lineageJson("executionEvent")("planId"))
+        }
+      }
+    }
+  }
+
+  it should "use DEFAULT mode when customLineagePath is empty string" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
+    withIsolatedSparkSession(_
+      .config("spark.spline.lineageDispatcher", "hdfs")
+      .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
+      .config("spark.spline.lineageDispatcher.hdfs.customLineagePath", "") // Empty string
+    ) { implicit spark =>
+      withLineageTracking { captor =>
+        import spark.implicits._
+        val dummyDF = Seq((1, 2)).toDF
+        val destPath = TempDirectory("spline_", ".parquet", pathOnly = true).deleteOnExit()
+
+        for {
+          (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
+        } yield {
+          // Empty string should trigger DEFAULT mode, lineage should be in destination directory
+          val lineageFile = new File(destPath.asString, "_LINEAGE")
+          lineageFile.exists should be(true)
+          lineageFile.length should be > 0L
+
+          val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
+          lineageJson should contain key "executionPlan"
+          lineageJson should contain key "executionEvent"
+        }
+      }
+    }
+  }
+
+  it should "use DEFAULT mode when customLineagePath is whitespace only" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
+    withIsolatedSparkSession(_
+      .config("spark.spline.lineageDispatcher", "hdfs")
+      .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
+      .config("spark.spline.lineageDispatcher.hdfs.customLineagePath", "   ") // Whitespace only
+    ) { implicit spark =>
+      withLineageTracking { captor =>
+        import spark.implicits._
+        val dummyDF = Seq((1, 2)).toDF
+        val destPath = TempDirectory("spline_", ".parquet", pathOnly = true).deleteOnExit()
+
+        for {
+          (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
+        } yield {
+          // Whitespace should trigger DEFAULT mode, lineage should be in destination directory
+          val lineageFile = new File(destPath.asString, "_LINEAGE")
+          lineageFile.exists should be(true)
+          lineageFile.length should be > 0L
+
+          val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
+          lineageJson should contain key "executionPlan"
+          lineageJson should contain key "executionEvent"
+        }
+      }
+    }
+  }
+
+  it should "generate chronologically sortable filenames in CENTRALIZED mode" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
+    val centralizedPath = TempDirectory("spline_centralized_", "", pathOnly = true).deleteOnExit()
+
+    withIsolatedSparkSession(_
+      .config("spark.spline.lineageDispatcher", "hdfs")
+      .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
+      .config("spark.spline.lineageDispatcher.hdfs.customLineagePath", centralizedPath.asString)
+    ) { implicit spark =>
+      withLineageTracking { captor =>
+        import spark.implicits._
+
+        // Write multiple lineage files
+        val futures = for (i <- 1 to 3) yield {
+          val dummyDF = Seq((i, i * 2)).toDF
+          val destPath = TempDirectory(s"spline_$i", ".parquet", pathOnly = true).deleteOnExit()
+          captor.lineageOf(dummyDF.write.save(destPath.asString))
+        }
+
+        for {
+          _ <- futures.head
+          _ <- futures(1)
+          _ <- futures(2)
+        } yield {
+          val centralizedDir = new File(centralizedPath.asString)
+          val lineageFiles = centralizedDir.listFiles().sorted
+          lineageFiles.length should be(3)
+
+          // Verify filenames are in chronological order (lexicographic = chronological due to timestamp-first format)
+          val filenames = lineageFiles.map(_.getName).toList
+          filenames shouldBe filenames.sorted
+
+          // Verify all filenames start with date prefix (yyyy-MM-dd)
+          filenames.foreach { name =>
+            name should fullyMatch regex """\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}-\d{3}_.+"""
+          }
+        }
+      }
+    }
+  }
+
+  it should "create nested directories in CENTRALIZED mode" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
+    val centralizedBasePath = TempDirectory("spline_base_", "", pathOnly = true).deleteOnExit()
+    val nestedPath = new File(centralizedBasePath.asString, "level1/level2/level3").getAbsolutePath
+
+    withIsolatedSparkSession(_
+      .config("spark.spline.lineageDispatcher", "hdfs")
+      .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
+      .config("spark.spline.lineageDispatcher.hdfs.customLineagePath", nestedPath)
+    ) { implicit spark =>
+      withLineageTracking { captor =>
+        import spark.implicits._
+        val dummyDF = Seq((1, 2)).toDF
+        val destPath = TempDirectory("spline_", ".parquet", pathOnly = true).deleteOnExit()
+
+        for {
+          (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
+        } yield {
+          // Verify nested directories were created
+          val nestedDir = new File(nestedPath)
+          nestedDir.exists should be(true)
+          nestedDir.isDirectory should be(true)
+
+          // Verify lineage file was written
+          val lineageFiles = nestedDir.listFiles()
+          lineageFiles should not be null
+          lineageFiles.length should be(1)
+          lineageFiles(0).length should be > 0L
+        }
+      }
+    }
+  }
+
 }

From 2b01b81eaca3b7da3c674d7bb686c30587a52880 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Sun, 12 Oct 2025 23:38:33 +0100
Subject: [PATCH 05/25] Aliging tests to be running same evaluation as when
 using default mode

---
 .../HDFSLineageDispatcherSpec.scala           | 76 +------------------
 1 file changed, 2 insertions(+), 74 deletions(-)

diff --git a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
index 9fafc1dc..58932e68 100644
--- a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
+++ b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
@@ -121,7 +121,6 @@ class HDFSLineageDispatcherSpec
         for {
           (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
         } yield {
-          // Empty string should trigger DEFAULT mode, lineage should be in destination directory
           val lineageFile = new File(destPath.asString, "_LINEAGE")
           lineageFile.exists should be(true)
           lineageFile.length should be > 0L
@@ -129,6 +128,7 @@ class HDFSLineageDispatcherSpec
           val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
           lineageJson should contain key "executionPlan"
           lineageJson should contain key "executionEvent"
+          lineageJson("executionPlan")("id") should equal(lineageJson("executionEvent")("planId"))
         }
       }
     }
@@ -148,7 +148,6 @@ class HDFSLineageDispatcherSpec
         for {
           (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
         } yield {
-          // Whitespace should trigger DEFAULT mode, lineage should be in destination directory
           val lineageFile = new File(destPath.asString, "_LINEAGE")
           lineageFile.exists should be(true)
           lineageFile.length should be > 0L
@@ -156,78 +155,7 @@ class HDFSLineageDispatcherSpec
           val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
           lineageJson should contain key "executionPlan"
           lineageJson should contain key "executionEvent"
-        }
-      }
-    }
-  }
-
-  it should "generate chronologically sortable filenames in CENTRALIZED mode" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
-    val centralizedPath = TempDirectory("spline_centralized_", "", pathOnly = true).deleteOnExit()
-
-    withIsolatedSparkSession(_
-      .config("spark.spline.lineageDispatcher", "hdfs")
-      .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
-      .config("spark.spline.lineageDispatcher.hdfs.customLineagePath", centralizedPath.asString)
-    ) { implicit spark =>
-      withLineageTracking { captor =>
-        import spark.implicits._
-
-        // Write multiple lineage files
-        val futures = for (i <- 1 to 3) yield {
-          val dummyDF = Seq((i, i * 2)).toDF
-          val destPath = TempDirectory(s"spline_$i", ".parquet", pathOnly = true).deleteOnExit()
-          captor.lineageOf(dummyDF.write.save(destPath.asString))
-        }
-
-        for {
-          _ <- futures.head
-          _ <- futures(1)
-          _ <- futures(2)
-        } yield {
-          val centralizedDir = new File(centralizedPath.asString)
-          val lineageFiles = centralizedDir.listFiles().sorted
-          lineageFiles.length should be(3)
-
-          // Verify filenames are in chronological order (lexicographic = chronological due to timestamp-first format)
-          val filenames = lineageFiles.map(_.getName).toList
-          filenames shouldBe filenames.sorted
-
-          // Verify all filenames start with date prefix (yyyy-MM-dd)
-          filenames.foreach { name =>
-            name should fullyMatch regex """\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}-\d{3}_.+"""
-          }
-        }
-      }
-    }
-  }
-
-  it should "create nested directories in CENTRALIZED mode" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
-    val centralizedBasePath = TempDirectory("spline_base_", "", pathOnly = true).deleteOnExit()
-    val nestedPath = new File(centralizedBasePath.asString, "level1/level2/level3").getAbsolutePath
-
-    withIsolatedSparkSession(_
-      .config("spark.spline.lineageDispatcher", "hdfs")
-      .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
-      .config("spark.spline.lineageDispatcher.hdfs.customLineagePath", nestedPath)
-    ) { implicit spark =>
-      withLineageTracking { captor =>
-        import spark.implicits._
-        val dummyDF = Seq((1, 2)).toDF
-        val destPath = TempDirectory("spline_", ".parquet", pathOnly = true).deleteOnExit()
-
-        for {
-          (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
-        } yield {
-          // Verify nested directories were created
-          val nestedDir = new File(nestedPath)
-          nestedDir.exists should be(true)
-          nestedDir.isDirectory should be(true)
-
-          // Verify lineage file was written
-          val lineageFiles = nestedDir.listFiles()
-          lineageFiles should not be null
-          lineageFiles.length should be(1)
-          lineageFiles(0).length should be > 0L
+          lineageJson("executionPlan")("id") should equal(lineageJson("executionEvent")("planId"))
         }
       }
     }

From 6f0ecdf553fddff75dcd70e248bb0e9d2f3fffd7 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Sun, 12 Oct 2025 23:49:40 +0100
Subject: [PATCH 06/25] Minor cleanup in HDFSLineageDispatcher

---
 core/src/main/resources/spline.default.yaml           |  2 --
 .../harvester/dispatcher/HDFSLineageDispatcher.scala  | 11 ++---------
 2 files changed, 2 insertions(+), 11 deletions(-)

diff --git a/core/src/main/resources/spline.default.yaml b/core/src/main/resources/spline.default.yaml
index 2277f656..1708f7c4 100644
--- a/core/src/main/resources/spline.default.yaml
+++ b/core/src/main/resources/spline.default.yaml
@@ -165,8 +165,6 @@ spline:
       #   Output: s3://my-bucket/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
       # - GCS: customLineagePath: gs://my-bucket/lineage
       #   Output: gs://my-bucket/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
-      # - Azure: customLineagePath: abfss://container@account.dfs.core.windows.net/lineage
-      #   Output: abfss://container@account.dfs.core.windows.net/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
       # - HDFS: customLineagePath: hdfs://cluster/lineage
       #   Output: hdfs://cluster/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
     # -------------------------------------------
diff --git a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
index aaccc340..f9c0fc38 100644
--- a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
+++ b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
@@ -130,11 +130,6 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
    * This format optimizes for operational debugging use cases:
    * - Timestamp FIRST: Ensures natural chronological sorting (most recent files appear together)
    * - Application ID: Full traceability to specific Spark application run
-   * Benefits:
-   * - Natural chronological sorting by default
-   * - Easy date-based filtering
-   * - Human-readable: Immediately see when each lineage file was created
-   * - Millisecond precision for uniqueness
    *
    * @param planId The execution plan ID (unused, kept for interface compatibility)
    * @return A unique filename optimized for filtering and sorting
@@ -179,11 +174,9 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
       try {
         // Create directories with multi-user friendly permissions to allow all service accounts to write
         // This uses the same permission object that's already configured for files
-        val permissions = this.permission.toString
-        val dirPermission = new FsPermission(permissions)
-        val created = fs.mkdirs(parentDir, dirPermission)
+        val created = fs.mkdirs(parentDir, permission)
         if (created) {
-          logInfo(s"Created parent directories: $parentDir with permissions $permissions")
+          logInfo(s"Created parent directories: $parentDir with permissions $permission")
         }
       } catch {
         case e: org.apache.hadoop.fs.FileAlreadyExistsException =>

From b0640419d24c50ce9601ec251d38a3d04e527ae3 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Mon, 13 Oct 2025 00:07:57 +0100
Subject: [PATCH 07/25] Fixed issue with unmatched types for resolveLineagePath
 in HDFSLineageDispatcher

---
 .../spline/harvester/dispatcher/HDFSLineageDispatcher.scala     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
index f9c0fc38..434f24bc 100644
--- a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
+++ b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
@@ -87,7 +87,7 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
       throw new IllegalStateException("send(event) must be called strictly after send(plan) method with matching plan ID")
 
     try {
-      val path = resolveLineagePath(event.planId)
+      val path = resolveLineagePath(event.planId.toString)
       val planWithEvent = Map(
         "executionPlan" -> this._lastSeenPlan,
         "executionEvent" -> event

From 578c876e021aa739d825f0d73cf97b4629dfd727 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Mon, 13 Oct 2025 00:19:19 +0100
Subject: [PATCH 08/25] Fixing issues as per SonarQube for
 HDFSLineageDispatcher

---
 .../dispatcher/HDFSLineageDispatcher.scala    | 27 +++++++++----------
 .../HDFSLineageDispatcherSpec.scala           | 27 +++++++++++--------
 2 files changed, 29 insertions(+), 25 deletions(-)

diff --git a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
index 434f24bc..a7a60a4c 100644
--- a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
+++ b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
@@ -69,7 +69,7 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
     filename = conf.getRequiredString(FileNameKey),
     permission = new FsPermission(conf.getRequiredObject(FilePermissionsKey).toString),
     bufferSize = conf.getRequiredInt(BufferSizeKey),
-    customLineagePath = conf.getOptionalString(CustomLineagePathKey)
+    customLineagePath = conf.getOptionalString(CustomLineagePathKey).map(_.trim).filter(_.nonEmpty)
   )
 
   @volatile
@@ -105,15 +105,14 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
    * If customLineagePath is specified, lineage files are written to that centralized location.
    * Otherwise, lineage files are written alongside the target data file (current behavior).
    *
-   * @param planId The execution plan ID used for generating unique filenames in centralized mode
    * @return The full path where the lineage file should be written
    */
-  private def resolveLineagePath(planId: String): String = {
+  private def resolveLineagePath(): String = {
     customLineagePath match {
       case Some(customPath) =>
         // Centralized mode: write to custom path with unique filename
         val cleanCustomPath = customPath.stripSuffix("/")
-        val uniqueFilename = generateUniqueFilename(planId)
+        val uniqueFilename = generateUniqueFilename()
         s"$cleanCustomPath/$uniqueFilename"
       case None =>
         // Default mode: write alongside target data file
@@ -131,10 +130,9 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
    * - Timestamp FIRST: Ensures natural chronological sorting (most recent files appear together)
    * - Application ID: Full traceability to specific Spark application run
    *
-   * @param planId The execution plan ID (unused, kept for interface compatibility)
    * @return A unique filename optimized for filtering and sorting
    */
-  private def generateUniqueFilename(planId: String): String = {
+  private def generateUniqueFilename(): String = {
     val sparkContext = SparkContext.getOrCreate()
     val appId = sparkContext.applicationId
     val dateFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd_HH-mm-ss-SSS").withZone(ZoneId.of("UTC"))
@@ -156,16 +154,17 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
     // Object storage systems (S3, GCS, Azure Blob) don't have real directories - they're just key prefixes
     // Skip directory creation for these systems to avoid unnecessary operations
     val fsScheme = fs.getUri.getScheme
-    val isObjectStorage = fsScheme != null && (
-      fsScheme.startsWith("s3") ||      // S3: s3, s3a, s3n
-      fsScheme.startsWith("gs") ||      // Google Cloud Storage: gs
-      fsScheme.startsWith("wasb") ||    // Azure Blob Storage: wasb, wasbs
-      fsScheme.startsWith("abfs") ||    // Azure Data Lake Storage Gen2: abfs, abfss
-      fsScheme.startsWith("adl")        // Azure Data Lake Storage Gen1: adl
+    val scheme = Option(fsSchemeRaw).map(_.toLowerCase(java.util.Locale.ROOT)).orNull
+    val isObjectStorage = scheme != null && (
+      scheme.startsWith("s3") ||      // S3: s3, s3a, s3n
+      scheme.startsWith("gs") ||      // Google Cloud Storage: gs
+      scheme.startsWith("wasb") ||    // Azure Blob Storage: wasb, wasbs
+      scheme.startsWith("abfs") ||    // Azure Data Lake Storage Gen2: abfs, abfss
+      scheme.startsWith("adl")        // Azure Data Lake Storage Gen1: adl
     )
-    
+  
     if (isObjectStorage) {
-      logDebug(s"Skipping directory creation for object storage filesystem ($fsScheme) - directories are implicit key prefixes")
+      logDebug(s"Skipping directory creation for object storage filesystem ($scheme) - directories are implicit key prefixes")
       return
     }
 
diff --git a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
index 58932e68..191b348c 100644
--- a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
+++ b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
@@ -37,10 +37,15 @@ class HDFSLineageDispatcherSpec
 
   behavior of "HDFSLineageDispatcher"
 
+  val lineageDispatcherConfigKeyName = "spark.spline.lineageDispatcher"
+  val lineageDispatcherConfigValueName = "hdfs"
+  val lineageDispatcherConfigClassNameKeyName = s"$lineageDispatcherConfigKeyName.$lineageDispatcherConfigValueName.className"
+  val lineageDispatcherConfigCustomLineagePathKeyName = s"$lineageDispatcherConfigKeyName.$lineageDispatcherConfigValueName.customLineagePath"
+
   it should "save lineage file to a filesystem in DEFAULT mode (no customLineagePath)" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
     withIsolatedSparkSession(_
-      .config("spark.spline.lineageDispatcher", "hdfs")
-      .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
+      .config(lineageDispatcherConfigKeyName, lineageDispatcherConfigValueName)
+      .config(lineageDispatcherConfigClassNameKeyName, classOf[HDFSLineageDispatcher].getName)
     ) { implicit spark =>
       withLineageTracking { captor =>
         import spark.implicits._
@@ -67,9 +72,9 @@ class HDFSLineageDispatcherSpec
     val centralizedPath = TempDirectory("spline_centralized_", "", pathOnly = true).deleteOnExit()
 
     withIsolatedSparkSession(_
-      .config("spark.spline.lineageDispatcher", "hdfs")
-      .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
-      .config("spark.spline.lineageDispatcher.hdfs.customLineagePath", centralizedPath.asString)
+      .config(lineageDispatcherConfigKeyName, lineageDispatcherConfigValueName)
+      .config(lineageDispatcherConfigClassNameKeyName, classOf[HDFSLineageDispatcher].getName)
+      .config(lineageDispatcherConfigCustomLineagePathKeyName, centralizedPath.asString)
     ) { implicit spark =>
       withLineageTracking { captor =>
         import spark.implicits._
@@ -109,9 +114,9 @@ class HDFSLineageDispatcherSpec
 
   it should "use DEFAULT mode when customLineagePath is empty string" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
     withIsolatedSparkSession(_
-      .config("spark.spline.lineageDispatcher", "hdfs")
-      .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
-      .config("spark.spline.lineageDispatcher.hdfs.customLineagePath", "") // Empty string
+      .config(lineageDispatcherConfigKeyName, lineageDispatcherConfigValueName)
+      .config(lineageDispatcherConfigClassNameKeyName, classOf[HDFSLineageDispatcher].getName)
+      .config(lineageDispatcherConfigCustomLineagePathKeyName, "") // Empty string
     ) { implicit spark =>
       withLineageTracking { captor =>
         import spark.implicits._
@@ -136,9 +141,9 @@ class HDFSLineageDispatcherSpec
 
   it should "use DEFAULT mode when customLineagePath is whitespace only" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
     withIsolatedSparkSession(_
-      .config("spark.spline.lineageDispatcher", "hdfs")
-      .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
-      .config("spark.spline.lineageDispatcher.hdfs.customLineagePath", "   ") // Whitespace only
+      .config(lineageDispatcherConfigKeyName, lineageDispatcherConfigValueName)
+      .config(lineageDispatcherConfigClassNameKeyName, classOf[HDFSLineageDispatcher].getName)
+      .config(lineageDispatcherConfigCustomLineagePathKeyName, "   ") // Whitespace only
     ) { implicit spark =>
       withLineageTracking { captor =>
         import spark.implicits._

From bebf4c9de381b1c60192f8db58a8e0911df64a99 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Mon, 13 Oct 2025 00:23:08 +0100
Subject: [PATCH 09/25] Fix for the issue in fsScheme!

---
 .../spline/harvester/dispatcher/HDFSLineageDispatcher.scala     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
index a7a60a4c..2c851ac3 100644
--- a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
+++ b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
@@ -154,7 +154,7 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
     // Object storage systems (S3, GCS, Azure Blob) don't have real directories - they're just key prefixes
     // Skip directory creation for these systems to avoid unnecessary operations
     val fsScheme = fs.getUri.getScheme
-    val scheme = Option(fsSchemeRaw).map(_.toLowerCase(java.util.Locale.ROOT)).orNull
+    val scheme = Option(fsScheme).map(_.toLowerCase(java.util.Locale.ROOT)).orNull
     val isObjectStorage = scheme != null && (
       scheme.startsWith("s3") ||      // S3: s3, s3a, s3n
       scheme.startsWith("gs") ||      // Google Cloud Storage: gs

From 04eebe36e53ba39c27a020174bdc85914adb85be Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Mon, 13 Oct 2025 00:24:32 +0100
Subject: [PATCH 10/25] Constant for file extension in
 HDFSLineageDispatcherSpec

---
 .../harvester/dispatcher/HDFSLineageDispatcherSpec.scala | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
index 191b348c..21d20c46 100644
--- a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
+++ b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
@@ -41,6 +41,7 @@ class HDFSLineageDispatcherSpec
   val lineageDispatcherConfigValueName = "hdfs"
   val lineageDispatcherConfigClassNameKeyName = s"$lineageDispatcherConfigKeyName.$lineageDispatcherConfigValueName.className"
   val lineageDispatcherConfigCustomLineagePathKeyName = s"$lineageDispatcherConfigKeyName.$lineageDispatcherConfigValueName.customLineagePath"
+  val destFilePathExtension = ".parquet"
 
   it should "save lineage file to a filesystem in DEFAULT mode (no customLineagePath)" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
     withIsolatedSparkSession(_
@@ -50,7 +51,7 @@ class HDFSLineageDispatcherSpec
       withLineageTracking { captor =>
         import spark.implicits._
         val dummyDF = Seq((1, 2)).toDF
-        val destPath = TempDirectory("spline_", ".parquet", pathOnly = true).deleteOnExit()
+        val destPath = TempDirectory("spline_", destFilePathExtension, pathOnly = true).deleteOnExit()
 
         for {
           (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
@@ -79,7 +80,7 @@ class HDFSLineageDispatcherSpec
       withLineageTracking { captor =>
         import spark.implicits._
         val dummyDF = Seq((1, 2)).toDF
-        val destPath = TempDirectory("spline_", ".parquet", pathOnly = true).deleteOnExit()
+        val destPath = TempDirectory("spline_", destFilePathExtension, pathOnly = true).deleteOnExit()
 
         for {
           (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
@@ -121,7 +122,7 @@ class HDFSLineageDispatcherSpec
       withLineageTracking { captor =>
         import spark.implicits._
         val dummyDF = Seq((1, 2)).toDF
-        val destPath = TempDirectory("spline_", ".parquet", pathOnly = true).deleteOnExit()
+        val destPath = TempDirectory("spline_", destFilePathExtension, pathOnly = true).deleteOnExit()
 
         for {
           (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
@@ -148,7 +149,7 @@ class HDFSLineageDispatcherSpec
       withLineageTracking { captor =>
         import spark.implicits._
         val dummyDF = Seq((1, 2)).toDF
-        val destPath = TempDirectory("spline_", ".parquet", pathOnly = true).deleteOnExit()
+        val destPath = TempDirectory("spline_", destFilePathExtension, pathOnly = true).deleteOnExit()
 
         for {
           (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))

From 5f63f61e8032c408615819cb36e4657b83cb2aa6 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Mon, 13 Oct 2025 00:52:42 +0100
Subject: [PATCH 11/25] Removed outputSource filename when writing file to new
 location and using internal SparkContext values

---
 core/src/main/resources/spline.default.yaml     | 15 ++++++++-------
 .../dispatcher/HDFSLineageDispatcher.scala      | 17 ++++++++++-------
 .../dispatcher/HDFSLineageDispatcherSpec.scala  |  8 ++++----
 3 files changed, 22 insertions(+), 18 deletions(-)

diff --git a/core/src/main/resources/spline.default.yaml b/core/src/main/resources/spline.default.yaml
index 1708f7c4..4ea7165b 100644
--- a/core/src/main/resources/spline.default.yaml
+++ b/core/src/main/resources/spline.default.yaml
@@ -151,22 +151,23 @@ spline:
       # If left empty, null, or not specified → DEFAULT MODE: lineage written alongside target data files
       # If set to a path → CENTRALIZED MODE: all lineage written to this location with unique filenames
       #
-      # CENTRALIZED MODE filename format: {timestamp}_{fileName}_{appId}
+      # CENTRALIZED MODE filename format: {timestamp}_{appName}_{appId}
       # - timestamp: Human-readable UTC timestamp (yyyy-MM-dd_HH-mm-ss-SSS) for natural chronological sorting and easy filtering
       #   Example: 2025-10-12_14-30-45-123
-      # - fileName: The configured fileName value (e.g., "my_file.parq_LINEAGE")
+      # - appName: Spark application name for easy identification of which job generated the lineage
+      #   Example: MySparkJob
       # - appId: Spark application ID for traceability to specific runs
       #   Example: app-20251012143045-0001
       #
-      # More examples:
+      # Examples (assuming app name is "MySparkJob"):
       # - Local: customLineagePath: /my/centralized/lineage
-      #   Output: /my/centralized/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
+      #   Output: /my/centralized/lineage/2025-10-12_14-30-45-123_MySparkJob_app-20251012143045-0001
       # - S3: customLineagePath: s3://my-bucket/lineage
-      #   Output: s3://my-bucket/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
+      #   Output: s3://my-bucket/lineage/2025-10-12_14-30-45-123_MySparkJob_app-20251012143045-0001
       # - GCS: customLineagePath: gs://my-bucket/lineage
-      #   Output: gs://my-bucket/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
+      #   Output: gs://my-bucket/lineage/2025-10-12_14-30-45-123_MySparkJob_app-20251012143045-0001
       # - HDFS: customLineagePath: hdfs://cluster/lineage
-      #   Output: hdfs://cluster/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
+      #   Output: hdfs://cluster/lineage/2025-10-12_14-30-45-123_MySparkJob_app-20251012143045-0001
     # -------------------------------------------
     # Open Lineage HTTP dispatcher
     # -------------------------------------------
diff --git a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
index 2c851ac3..49d7fbc0 100644
--- a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
+++ b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
@@ -51,9 +51,9 @@ import scala.concurrent.blocking
  *
  * 2. CENTRALIZED MODE (customLineagePath set to a valid path):
  *    All lineage files are written to a single centralized location with unique filenames.
- *    Filename format: {timestamp}_{fileName}_{appId}
+ *    Filename format: {timestamp}_{appName}_{appId}
  *    - timestamp: Human-readable UTC timestamp (yyyy-MM-dd_HH-mm-ss-SSS) for chronological sorting and filtering
- *    - fileName: The configured fileName value (e.g., "my_file.parq_LINEAGE")
+ *    - appName: Spark application name for easy identification
  *    - appId: Spark application ID for traceability
  *
  *    The timestamp-first format ensures natural chronological sorting and easy date-based filtering.
@@ -87,7 +87,7 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
       throw new IllegalStateException("send(event) must be called strictly after send(plan) method with matching plan ID")
 
     try {
-      val path = resolveLineagePath(event.planId.toString)
+      val path = resolveLineagePath()
       val planWithEvent = Map(
         "executionPlan" -> this._lastSeenPlan,
         "executionEvent" -> event
@@ -108,6 +108,7 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
    * @return The full path where the lineage file should be written
    */
   private def resolveLineagePath(): String = {
+    val outputSource = s"${this._lastSeenPlan.operations.write.outputSource}"
     customLineagePath match {
       case Some(customPath) =>
         // Centralized mode: write to custom path with unique filename
@@ -116,28 +117,30 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
         s"$cleanCustomPath/$uniqueFilename"
       case None =>
         // Default mode: write alongside target data file
-        s"${this._lastSeenPlan.operations.write.outputSource.stripSuffix("/")}/$filename"
+        s"${outputSource.stripSuffix("/")}/$filename"
     }
   }
 
   /**
    * Generates a unique filename for centralized lineage storage.
    * 
-   * Format: {timestamp}_{fileName}_{appId}
-   * Example: 2025-10-12_14-30-45-123_lineage_app-20251012143045-0001
+   * Format: {timestamp}_{appName}_{appId}
+   * Example: 2025-10-12_14-30-45-123_MySparkJob_app-20251012143045-0001
    *
    * This format optimizes for operational debugging use cases:
    * - Timestamp FIRST: Ensures natural chronological sorting (most recent files appear together)
+   * - Application Name: Easy identification of which job generated the lineage
    * - Application ID: Full traceability to specific Spark application run
    *
    * @return A unique filename optimized for filtering and sorting
    */
   private def generateUniqueFilename(): String = {
     val sparkContext = SparkContext.getOrCreate()
+    val appName = sparkContext.appName
     val appId = sparkContext.applicationId
     val dateFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd_HH-mm-ss-SSS").withZone(ZoneId.of("UTC"))
     val timestamp = dateFormatter.format(Instant.now())
-    s"${timestamp}_${filename}_${appId}"
+    s"${timestamp}_${appName}_${appId}"
   }
 
   /**
diff --git a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
index 21d20c46..621a9030 100644
--- a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
+++ b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
@@ -43,7 +43,7 @@ class HDFSLineageDispatcherSpec
   val lineageDispatcherConfigCustomLineagePathKeyName = s"$lineageDispatcherConfigKeyName.$lineageDispatcherConfigValueName.customLineagePath"
   val destFilePathExtension = ".parquet"
 
-  it should "save lineage file to a filesystem in DEFAULT mode (no customLineagePath)" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
+  it should "save lineage file to a filesystem in DEFAULT mode" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
     withIsolatedSparkSession(_
       .config(lineageDispatcherConfigKeyName, lineageDispatcherConfigValueName)
       .config(lineageDispatcherConfigClassNameKeyName, classOf[HDFSLineageDispatcher].getName)
@@ -98,10 +98,10 @@ class HDFSLineageDispatcherSpec
           val lineageFile = lineageFiles(0)
           lineageFile.length should be > 0L
 
-          // Verify filename format: {timestamp}_{fileName}_{appId}
+          // Verify filename format: {timestamp}_{appName}_{appId}
           val filename = lineageFile.getName
-          // Should match pattern: yyyy-MM-dd_HH-mm-ss-SSS__LINEAGE_app-...
-          filename should include("_LINEAGE_")
+          // Should match pattern: yyyy-MM-dd_HH-mm-ss-SSS_{appName}_app-...
+          // AppName and AppId are part of the filename
           filename should startWith regex """\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}-\d{3}"""
 
           val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]

From 8bb78a1e5826613164a833200a05795570ec8eaa Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Mon, 13 Oct 2025 01:13:30 +0100
Subject: [PATCH 12/25] Fixing issues as per static code analysis

---
 .../dispatcher/HDFSLineageDispatcher.scala    | 50 +++++++++++--------
 .../HDFSLineageDispatcherSpec.scala           |  8 +--
 2 files changed, 32 insertions(+), 26 deletions(-)

diff --git a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
index 49d7fbc0..758d8256 100644
--- a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
+++ b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
@@ -33,6 +33,7 @@ import java.net.URI
 import java.time.format.DateTimeFormatter
 import java.time.{Instant, ZoneId}
 import scala.concurrent.blocking
+import scala.util.{Try, Success, Failure}
 
 /**
  * A port of https://github.com/AbsaOSS/spline/tree/release/0.3.9/persistence/hdfs/src/main/scala/za/co/absa/spline/persistence/hdfs
@@ -144,8 +145,8 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
   }
 
   /**
-   * Ensures that parent directories exist with proper permissions for multi-user access.
-   * This is critical for centralized lineage storage where different service accounts need write access.
+   * Ensures that parent directories exist with proper permissions for centralized lineage storage.
+   * This is only called when customLineagePath is specified and is critical for multi-user access.
    *
    * Note: Object storage systems (S3, GCS, Azure Blob) don't have true directories - they use key prefixes.
    * Directory creation is automatically skipped for these systems to avoid unnecessary operations.
@@ -168,25 +169,27 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
   
     if (isObjectStorage) {
       logDebug(s"Skipping directory creation for object storage filesystem ($scheme) - directories are implicit key prefixes")
-      return
-    }
-
-    val parentDir = path.getParent
-    if (parentDir != null && !fs.exists(parentDir)) {
-      try {
-        // Create directories with multi-user friendly permissions to allow all service accounts to write
-        // This uses the same permission object that's already configured for files
-        val created = fs.mkdirs(parentDir, permission)
-        if (created) {
-          logInfo(s"Created parent directories: $parentDir with permissions $permission")
+    } else {
+      logDebug(s"Ensuring parent directories exist for centralized lineage storage: $path")
+      val parentDir = path.getParent
+      if (parentDir != null && !fs.exists(parentDir)) {
+        Try {
+          // Create directories with multi-user friendly permissions to allow all service accounts to write
+          // This uses the same permission object that's already configured for files
+          val created = fs.mkdirs(parentDir, permission)
+          if (created) {
+            logInfo(s"Created parent directories: $parentDir with permissions $permission")
+          }
+        } match {
+          case Success(_) =>
+            // Directory creation succeeded
+          case Failure(e: org.apache.hadoop.fs.FileAlreadyExistsException) =>
+            // Race condition: another process created the directory - this is fine
+            logDebug(s"Directory $parentDir already exists (created by another process)")
+          case Failure(e: RuntimeException) =>
+            logWarning(s"Failed to create parent directories: $parentDir", e)
+            throw e
         }
-      } catch {
-        case e: org.apache.hadoop.fs.FileAlreadyExistsException =>
-          // Race condition: another process created the directory - this is fine
-          logDebug(s"Directory $parentDir already exists (created by another process)")
-        case e: Exception =>
-          logWarning(s"Failed to create parent directories: $parentDir", e)
-          throw e
       }
     }
   }
@@ -195,8 +198,11 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
     val (fs, path) = pathStringToFsWithPath(fullLineagePath)
     logDebug(s"Opening HadoopFs output stream to $path")
 
-    // Ensure parent directories exist with proper permissions for multi-user access
-    ensureParentDirectoriesExist(fs, path)
+    // Only ensure parent directories exist when using centralized mode (customLineagePath is specified)
+    // In default mode, the parent directories should already exist as they're alongside the data files
+    if (customLineagePath.isDefined) {
+      ensureParentDirectoriesExist(fs, path)
+    }
 
     val replication = fs.getDefaultReplication(path)
     val blockSize = fs.getDefaultBlockSize(path)
diff --git a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
index 621a9030..bec2ac8c 100644
--- a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
+++ b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
@@ -91,11 +91,11 @@ class HDFSLineageDispatcherSpec
 
           // Lineage should be in centralized directory with timestamp-based filename
           val centralizedDir = new File(centralizedPath.asString)
-          val lineageFiles = centralizedDir.listFiles()
-          lineageFiles should not be null
-          lineageFiles.length should be(1)
+          val lineageFiles = Option(centralizedDir.listFiles())
+          lineageFiles should be(defined)
+          lineageFiles.get.length should be(1)
 
-          val lineageFile = lineageFiles(0)
+          val lineageFile = lineageFiles.get(0)
           lineageFile.length should be > 0L
 
           // Verify filename format: {timestamp}_{appName}_{appId}

From 7a9370508645c323e56bc0476f97951861ea85fe Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Mon, 13 Oct 2025 01:21:29 +0100
Subject: [PATCH 13/25] Fixing issues in logic for HDFSLineageDispatcher

---
 .../HDFSLineageDispatcherSpec.scala           | 35 +++++++++++--------
 1 file changed, 20 insertions(+), 15 deletions(-)

diff --git a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
index bec2ac8c..c3e943f0 100644
--- a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
+++ b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
@@ -93,21 +93,26 @@ class HDFSLineageDispatcherSpec
           val centralizedDir = new File(centralizedPath.asString)
           val lineageFiles = Option(centralizedDir.listFiles())
           lineageFiles should be(defined)
-          lineageFiles.get.length should be(1)
-
-          val lineageFile = lineageFiles.get(0)
-          lineageFile.length should be > 0L
-
-          // Verify filename format: {timestamp}_{appName}_{appId}
-          val filename = lineageFile.getName
-          // Should match pattern: yyyy-MM-dd_HH-mm-ss-SSS_{appName}_app-...
-          // AppName and AppId are part of the filename
-          filename should startWith regex """\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}-\d{3}"""
-
-          val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
-          lineageJson should contain key "executionPlan"
-          lineageJson should contain key "executionEvent"
-          lineageJson("executionPlan")("id") should equal(lineageJson("executionEvent")("planId"))
+          
+          lineageFiles match {
+            case Some(files) =>
+              files.length should be(1)
+              val lineageFile = files(0)
+              lineageFile.length should be > 0L
+
+              // Verify filename format: {timestamp}_{appName}_{appId}
+              val filename = lineageFile.getName
+              // Should match pattern: yyyy-MM-dd_HH-mm-ss-SSS_{appName}_app-...
+              // AppName and AppId are part of the filename
+              filename should startWith regex """\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}-\d{3}"""
+
+              val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
+              lineageJson should contain key "executionPlan"
+              lineageJson should contain key "executionEvent"
+              lineageJson("executionPlan")("id") should equal(lineageJson("executionEvent")("planId"))
+            case None =>
+              fail("Expected lineage files to be present in centralized directory")
+          }
         }
       }
     }

From 32db7ac3c728f9da1880cb28b6ec773a10cffb8d Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Mon, 13 Oct 2025 01:23:14 +0100
Subject: [PATCH 14/25] Fixing issue with exception handling for mkdirs in
 HDFSLineageDispatcher

---
 .../spline/harvester/dispatcher/HDFSLineageDispatcher.scala   | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
index 758d8256..1a5bf418 100644
--- a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
+++ b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
@@ -189,6 +189,10 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
           case Failure(e: RuntimeException) =>
             logWarning(s"Failed to create parent directories: $parentDir", e)
             throw e
+          case Failure(e) =>
+            // Handle any other exceptions (IOException, etc.)
+            logWarning(s"Failed to create parent directories: $parentDir", e)
+            throw new RuntimeException(s"Failed to create parent directories: $parentDir", e)
         }
       }
     }

From 30f2f40eb9b50a1ea9226c5a66adf325a6811982 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Mon, 13 Oct 2025 06:36:09 +0100
Subject: [PATCH 15/25] Updated integration test for custom lineage path in
 HDFSLineageDispatcherSpec

---
 .../HDFSLineageDispatcherSpec.scala           | 92 ++++++++++---------
 1 file changed, 47 insertions(+), 45 deletions(-)

diff --git a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
index c3e943f0..68430035 100644
--- a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
+++ b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
@@ -69,13 +69,11 @@ class HDFSLineageDispatcherSpec
     }
   }
 
-  it should "save lineage file in CENTRALIZED mode with customLineagePath" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
-    val centralizedPath = TempDirectory("spline_centralized_", "", pathOnly = true).deleteOnExit()
-
+  it should "use DEFAULT mode when customLineagePath is empty string" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
     withIsolatedSparkSession(_
       .config(lineageDispatcherConfigKeyName, lineageDispatcherConfigValueName)
       .config(lineageDispatcherConfigClassNameKeyName, classOf[HDFSLineageDispatcher].getName)
-      .config(lineageDispatcherConfigCustomLineagePathKeyName, centralizedPath.asString)
+      .config(lineageDispatcherConfigCustomLineagePathKeyName, "") // Empty string
     ) { implicit spark =>
       withLineageTracking { captor =>
         import spark.implicits._
@@ -85,44 +83,24 @@ class HDFSLineageDispatcherSpec
         for {
           (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
         } yield {
-          // Lineage should NOT be in the destination directory
-          val lineageFileInDest = new File(destPath.asString, "_LINEAGE")
-          lineageFileInDest.exists should be(false)
+          val lineageFile = new File(destPath.asString, "_LINEAGE")
+          lineageFile.exists should be(true)
+          lineageFile.length should be > 0L
 
-          // Lineage should be in centralized directory with timestamp-based filename
-          val centralizedDir = new File(centralizedPath.asString)
-          val lineageFiles = Option(centralizedDir.listFiles())
-          lineageFiles should be(defined)
-          
-          lineageFiles match {
-            case Some(files) =>
-              files.length should be(1)
-              val lineageFile = files(0)
-              lineageFile.length should be > 0L
-
-              // Verify filename format: {timestamp}_{appName}_{appId}
-              val filename = lineageFile.getName
-              // Should match pattern: yyyy-MM-dd_HH-mm-ss-SSS_{appName}_app-...
-              // AppName and AppId are part of the filename
-              filename should startWith regex """\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}-\d{3}"""
-
-              val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
-              lineageJson should contain key "executionPlan"
-              lineageJson should contain key "executionEvent"
-              lineageJson("executionPlan")("id") should equal(lineageJson("executionEvent")("planId"))
-            case None =>
-              fail("Expected lineage files to be present in centralized directory")
-          }
+          val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
+          lineageJson should contain key "executionPlan"
+          lineageJson should contain key "executionEvent"
+          lineageJson("executionPlan")("id") should equal(lineageJson("executionEvent")("planId"))
         }
       }
     }
   }
 
-  it should "use DEFAULT mode when customLineagePath is empty string" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
+  it should "use DEFAULT mode when customLineagePath is whitespace only" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
     withIsolatedSparkSession(_
       .config(lineageDispatcherConfigKeyName, lineageDispatcherConfigValueName)
       .config(lineageDispatcherConfigClassNameKeyName, classOf[HDFSLineageDispatcher].getName)
-      .config(lineageDispatcherConfigCustomLineagePathKeyName, "") // Empty string
+      .config(lineageDispatcherConfigCustomLineagePathKeyName, "   ") // Whitespace only
     ) { implicit spark =>
       withLineageTracking { captor =>
         import spark.implicits._
@@ -145,28 +123,52 @@ class HDFSLineageDispatcherSpec
     }
   }
 
-  it should "use DEFAULT mode when customLineagePath is whitespace only" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
+  it should "save lineage files in a custom lineage path" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
+    val centralizedPath = TempDirectory("spline_centralized").deleteOnExit()
+
     withIsolatedSparkSession(_
       .config(lineageDispatcherConfigKeyName, lineageDispatcherConfigValueName)
       .config(lineageDispatcherConfigClassNameKeyName, classOf[HDFSLineageDispatcher].getName)
-      .config(lineageDispatcherConfigCustomLineagePathKeyName, "   ") // Whitespace only
+      .config(lineageDispatcherConfigCustomLineagePathKeyName, centralizedPath.asString)
     ) { implicit spark =>
       withLineageTracking { captor =>
         import spark.implicits._
-        val dummyDF = Seq((1, 2)).toDF
-        val destPath = TempDirectory("spline_", destFilePathExtension, pathOnly = true).deleteOnExit()
+        
+        // Test with multiple data writes to verify unique filenames
+        val dummyDF1 = Seq((1, 2)).toDF
+        val dummyDF2 = Seq((3, 4)).toDF
+        val destPath1 = TempDirectory("spline_1_", destFilePathExtension, pathOnly = true).deleteOnExit()
+        val destPath2 = TempDirectory("spline_2_", destFilePathExtension, pathOnly = true).deleteOnExit()
 
         for {
-          (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
+          (_, _) <- captor.lineageOf(dummyDF1.write.save(destPath1.asString))
+          (_, _) <- captor.lineageOf(dummyDF2.write.save(destPath2.asString))
         } yield {
-          val lineageFile = new File(destPath.asString, "_LINEAGE")
-          lineageFile.exists should be(true)
-          lineageFile.length should be > 0L
+          val centralizedDir = new File(centralizedPath.asString)
+          centralizedDir.exists should be(true)
+          centralizedDir.isDirectory should be(true)
+
+          // Should have 2 lineage files (one for each write operation)
+          val lineageFiles = Option(centralizedDir.listFiles()).getOrElse(Array.empty[File])
+          val lineageFilesOnly = lineageFiles.filter(_.isFile)
+          lineageFilesOnly.length should be(2)
+
+          // Verify both files have unique names
+          val filenames = lineageFilesOnly.map(_.getName).toSet
+          filenames.size should be(2) // All filenames should be unique
+
+          // Verify each file has the correct format and content
+          lineageFilesOnly.map { lineageFile =>
+            val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
+            lineageJson should contain key "executionPlan"
+            lineageJson should contain key "executionEvent"
+          }
 
-          val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
-          lineageJson should contain key "executionPlan"
-          lineageJson should contain key "executionEvent"
-          lineageJson("executionPlan")("id") should equal(lineageJson("executionEvent")("planId"))
+          // Verify no lineage files in destination directories
+          val lineageFileInDest1 = new File(destPath1.asString, "_LINEAGE")
+          val lineageFileInDest2 = new File(destPath2.asString, "_LINEAGE")
+          lineageFileInDest1.exists should be(false)
+          lineageFileInDest2.exists should be(false)
         }
       }
     }

From 87936b8df720b518963dd64e14158eecc902c026 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Mon, 13 Oct 2025 22:15:05 +0100
Subject: [PATCH 16/25] Ensuring the edgecase with Spark AppName containing
 non-standard characters is covered and updated integration tests for
 HDFSLineageDispatcherSpec

---
 .../dispatcher/HDFSLineageDispatcher.scala    |  2 +-
 .../HDFSLineageDispatcherSpec.scala           | 94 +++++++++----------
 2 files changed, 44 insertions(+), 52 deletions(-)

diff --git a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
index 1a5bf418..1d42ded1 100644
--- a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
+++ b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
@@ -137,7 +137,7 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
    */
   private def generateUniqueFilename(): String = {
     val sparkContext = SparkContext.getOrCreate()
-    val appName = sparkContext.appName
+    val appName = sparkContext.appName.replaceAll("[^a-zA-Z0-9_-]", "_")
     val appId = sparkContext.applicationId
     val dateFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd_HH-mm-ss-SSS").withZone(ZoneId.of("UTC"))
     val timestamp = dateFormatter.format(Instant.now())
diff --git a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
index 68430035..1b30d1ca 100644
--- a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
+++ b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
@@ -69,55 +69,37 @@ class HDFSLineageDispatcherSpec
     }
   }
 
-  it should "use DEFAULT mode when customLineagePath is empty string" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
-    withIsolatedSparkSession(_
-      .config(lineageDispatcherConfigKeyName, lineageDispatcherConfigValueName)
-      .config(lineageDispatcherConfigClassNameKeyName, classOf[HDFSLineageDispatcher].getName)
-      .config(lineageDispatcherConfigCustomLineagePathKeyName, "") // Empty string
-    ) { implicit spark =>
-      withLineageTracking { captor =>
-        import spark.implicits._
-        val dummyDF = Seq((1, 2)).toDF
-        val destPath = TempDirectory("spline_", destFilePathExtension, pathOnly = true).deleteOnExit()
-
-        for {
-          (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
-        } yield {
-          val lineageFile = new File(destPath.asString, "_LINEAGE")
-          lineageFile.exists should be(true)
-          lineageFile.length should be > 0L
-
-          val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
-          lineageJson should contain key "executionPlan"
-          lineageJson should contain key "executionEvent"
-          lineageJson("executionPlan")("id") should equal(lineageJson("executionEvent")("planId"))
-        }
+  Seq(
+    ("without customLineagePath config", None),
+    ("with empty string customLineagePath", Some("")),
+    ("with whitespace-only customLineagePath", Some("   "))
+  ).foreach { case (scenarioDesc, customPathValue) =>
+    it should s"use DEFAULT mode $scenarioDesc" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
+      val builder = (b: SparkSession.Builder) => {
+        val configured = b
+          .config(lineageDispatcherConfigKeyName, lineageDispatcherConfigValueName)
+          .config(lineageDispatcherConfigClassNameKeyName, classOf[HDFSLineageDispatcher].getName)
+        customPathValue.fold(configured)(path => configured.config(lineageDispatcherConfigCustomLineagePathKeyName, path))
       }
-    }
-  }
-
-  it should "use DEFAULT mode when customLineagePath is whitespace only" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
-    withIsolatedSparkSession(_
-      .config(lineageDispatcherConfigKeyName, lineageDispatcherConfigValueName)
-      .config(lineageDispatcherConfigClassNameKeyName, classOf[HDFSLineageDispatcher].getName)
-      .config(lineageDispatcherConfigCustomLineagePathKeyName, "   ") // Whitespace only
-    ) { implicit spark =>
-      withLineageTracking { captor =>
-        import spark.implicits._
-        val dummyDF = Seq((1, 2)).toDF
-        val destPath = TempDirectory("spline_", destFilePathExtension, pathOnly = true).deleteOnExit()
-
-        for {
-          (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
-        } yield {
-          val lineageFile = new File(destPath.asString, "_LINEAGE")
-          lineageFile.exists should be(true)
-          lineageFile.length should be > 0L
+      
+      withIsolatedSparkSession(builder) { implicit spark =>
+        withLineageTracking { captor =>
+          import spark.implicits._
+          val dummyDF = Seq((1, 2)).toDF
+          val destPath = TempDirectory("spline_", destFilePathExtension, pathOnly = true).deleteOnExit()
+
+          for {
+            (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
+          } yield {
+            val lineageFile = new File(destPath.asString, "_LINEAGE")
+            lineageFile.exists should be(true)
+            lineageFile.length should be > 0L
 
-          val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
-          lineageJson should contain key "executionPlan"
-          lineageJson should contain key "executionEvent"
-          lineageJson("executionPlan")("id") should equal(lineageJson("executionEvent")("planId"))
+            val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
+            lineageJson should contain key "executionPlan"
+            lineageJson should contain key "executionEvent"
+            lineageJson("executionPlan")("id") should equal(lineageJson("executionEvent")("planId"))
+          }
         }
       }
     }
@@ -147,15 +129,25 @@ class HDFSLineageDispatcherSpec
           val centralizedDir = new File(centralizedPath.asString)
           centralizedDir.exists should be(true)
           centralizedDir.isDirectory should be(true)
+          
+          val appId = spark.sparkContext.applicationId
+          val appName = spark.sparkContext.appName
+          val appNameCleaned = appName.replaceAll(r"[^a-zA-Z0-9_-]".r, "_")
 
           // Should have 2 lineage files (one for each write operation)
           val lineageFiles = Option(centralizedDir.listFiles()).getOrElse(Array.empty[File])
-          val lineageFilesOnly = lineageFiles.filter(_.isFile)
+          val lineageFilesOnly = lineageFiles.filter(f => f.isFile && !f.getName.endsWith(".crc"))
           lineageFilesOnly.length should be(2)
 
-          // Verify both files have unique names
-          val filenames = lineageFilesOnly.map(_.getName).toSet
-          filenames.size should be(2) // All filenames should be unique
+          // Verify naming convention aligns with centralized lineage pattern (timestamp_appName_appId)
+          val filenamePattern = """\\d{4}-\\d{2}-\\d{2}_\\d{2}-\\d{2}-\\d{2}-\\d{3}_.+_.+""".r
+          lineageFilesOnly.foreach { file =>
+            val name = file.getName
+            withClue(s"Lineage filename '$name' should follow the timestamp_appName_appId pattern") {
+              filenamePattern.matches(name) shouldBe true
+            }
+            name should include (appId) and include (appNameCleaned)
+          }
 
           // Verify each file has the correct format and content
           lineageFilesOnly.map { lineageFile =>

From 4e699dd468c0344c0303b8343378facb2a61085f Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Mon, 13 Oct 2025 22:24:27 +0100
Subject: [PATCH 17/25] HDFSLineageDispatcherSpec debug for failing integration
 test

---
 .../harvester/dispatcher/HDFSLineageDispatcherSpec.scala   | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
index 1b30d1ca..0d8d551f 100644
--- a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
+++ b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
@@ -134,15 +134,14 @@ class HDFSLineageDispatcherSpec
           val appName = spark.sparkContext.appName
           val appNameCleaned = appName.replaceAll(r"[^a-zA-Z0-9_-]".r, "_")
 
-          // Should have 2 lineage files (one for each write operation)
           val lineageFiles = Option(centralizedDir.listFiles()).getOrElse(Array.empty[File])
-          val lineageFilesOnly = lineageFiles.filter(f => f.isFile && !f.getName.endsWith(".crc"))
-          lineageFilesOnly.length should be(2)
+          println(lineageFiles.map(_.getName).mkString(", "))
 
           // Verify naming convention aligns with centralized lineage pattern (timestamp_appName_appId)
           val filenamePattern = """\\d{4}-\\d{2}-\\d{2}_\\d{2}-\\d{2}-\\d{2}-\\d{3}_.+_.+""".r
-          lineageFilesOnly.foreach { file =>
+          lineageFiles.foreach { file =>
             val name = file.getName
+            println(name)
             withClue(s"Lineage filename '$name' should follow the timestamp_appName_appId pattern") {
               filenamePattern.matches(name) shouldBe true
             }

From 8a4b431a40c9c7911cc8fcbb43aa727d2387e620 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Mon, 13 Oct 2025 22:27:10 +0100
Subject: [PATCH 18/25] HDFSLineageDispatcherSpec debug for failing integration
 test

---
 .../harvester/dispatcher/HDFSLineageDispatcherSpec.scala    | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
index 0d8d551f..fc2b37ef 100644
--- a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
+++ b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
@@ -26,7 +26,7 @@ import za.co.absa.spline.commons.version.Version._
 import za.co.absa.spline.harvester.json.HarvesterJsonSerDe.impl._
 import za.co.absa.spline.test.fixture.SparkFixture
 import za.co.absa.spline.test.fixture.spline.SplineFixture
-
+import org.apache.spark.sql.SparkSession
 import java.io.File
 
 class HDFSLineageDispatcherSpec
@@ -132,13 +132,13 @@ class HDFSLineageDispatcherSpec
           
           val appId = spark.sparkContext.applicationId
           val appName = spark.sparkContext.appName
-          val appNameCleaned = appName.replaceAll(r"[^a-zA-Z0-9_-]".r, "_")
+          val appNameCleaned = appName.replaceAll("[^a-zA-Z0-9_-]", "_")
 
           val lineageFiles = Option(centralizedDir.listFiles()).getOrElse(Array.empty[File])
           println(lineageFiles.map(_.getName).mkString(", "))
 
           // Verify naming convention aligns with centralized lineage pattern (timestamp_appName_appId)
-          val filenamePattern = """\\d{4}-\\d{2}-\\d{2}_\\d{2}-\\d{2}-\\d{2}-\\d{3}_.+_.+""".r
+          val filenamePattern = """\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}-\d{3}_.+_.+""".r
           lineageFiles.foreach { file =>
             val name = file.getName
             println(name)

From 6bc068dd3824dc12ffa071ae3b44eed05fc91935 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Mon, 13 Oct 2025 22:33:47 +0100
Subject: [PATCH 19/25] Fixing issues in HDFSLineageDispatcherSpec

---
 .../harvester/dispatcher/HDFSLineageDispatcherSpec.scala      | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
index fc2b37ef..14b92f12 100644
--- a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
+++ b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
@@ -135,7 +135,7 @@ class HDFSLineageDispatcherSpec
           val appNameCleaned = appName.replaceAll("[^a-zA-Z0-9_-]", "_")
 
           val lineageFiles = Option(centralizedDir.listFiles()).getOrElse(Array.empty[File])
-          println(lineageFiles.map(_.getName).mkString(", "))
+          println("lineageFiles: " + lineageFiles.map(_.getName).mkString(", "))
 
           // Verify naming convention aligns with centralized lineage pattern (timestamp_appName_appId)
           val filenamePattern = """\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}-\d{3}_.+_.+""".r
@@ -149,7 +149,7 @@ class HDFSLineageDispatcherSpec
           }
 
           // Verify each file has the correct format and content
-          lineageFilesOnly.map { lineageFile =>
+          lineageFiles.foreach { lineageFile =>
             val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
             lineageJson should contain key "executionPlan"
             lineageJson should contain key "executionEvent"

From d155a981d5adf4c9920cfb8b198408b32d109fa3 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Mon, 13 Oct 2025 22:41:06 +0100
Subject: [PATCH 20/25] Fixing issues in HDFSLineageDispatcherSpec

---
 .../dispatcher/HDFSLineageDispatcherSpec.scala    | 15 ++++++++-------
 1 file changed, 8 insertions(+), 7 deletions(-)

diff --git a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
index 14b92f12..010184e9 100644
--- a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
+++ b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
@@ -135,21 +135,22 @@ class HDFSLineageDispatcherSpec
           val appNameCleaned = appName.replaceAll("[^a-zA-Z0-9_-]", "_")
 
           val lineageFiles = Option(centralizedDir.listFiles()).getOrElse(Array.empty[File])
-          println("lineageFiles: " + lineageFiles.map(_.getName).mkString(", "))
+          val lineageFilesOnly = lineageFiles.filter(f => f.isFile && !f.getName.endsWith(".crc"))
+          lineageFilesOnly.length should be(2)
 
           // Verify naming convention aligns with centralized lineage pattern (timestamp_appName_appId)
-          val filenamePattern = """\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}-\d{3}_.+_.+""".r
-          lineageFiles.foreach { file =>
+          val filenamePattern = """\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}-\d{3}_.+_.+"""
+          lineageFilesOnly.foreach { file =>
             val name = file.getName
-            println(name)
             withClue(s"Lineage filename '$name' should follow the timestamp_appName_appId pattern") {
-              filenamePattern.matches(name) shouldBe true
+              name.matches(filenamePattern) shouldBe true
             }
-            name should include (appId) and include (appNameCleaned)
+            name should include appId
+            name should include appNameCleaned
           }
 
           // Verify each file has the correct format and content
-          lineageFiles.foreach { lineageFile =>
+          lineageFilesOnly.foreach { lineageFile =>
             val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
             lineageJson should contain key "executionPlan"
             lineageJson should contain key "executionEvent"

From 4c7d7233b73ab92d2716a7e649f7926cbe1e16cf Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Mon, 13 Oct 2025 23:54:09 +0100
Subject: [PATCH 21/25] Fixed integration test and changed the filename to
 avoid clashed from the same Spark application in case it is writing files in
 parallel

---
 .../dispatcher/HDFSLineageDispatcher.scala    | 29 ++++++++++++-------
 .../HDFSLineageDispatcherSpec.scala           | 20 ++++++++-----
 2 files changed, 31 insertions(+), 18 deletions(-)

diff --git a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
index 1d42ded1..e6fcf796 100644
--- a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
+++ b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
@@ -52,10 +52,10 @@ import scala.util.{Try, Success, Failure}
  *
  * 2. CENTRALIZED MODE (customLineagePath set to a valid path):
  *    All lineage files are written to a single centralized location with unique filenames.
- *    Filename format: {timestamp}_{appName}_{appId}
+ *    Filename format: {timestamp}_{planId}_{appId}
  *    - timestamp: Human-readable UTC timestamp (yyyy-MM-dd_HH-mm-ss-SSS) for chronological sorting and filtering
- *    - appName: Spark application name for easy identification
- *    - appId: Spark application ID for traceability
+ *    - planId: Execution plan UUID for guaranteed uniqueness (prevents collisions from concurrent writes)
+ *    - appId: Spark application ID for traceability to Spark UI, logs, and monitoring systems
  *
  *    The timestamp-first format ensures natural chronological sorting and easy date-based filtering.
  *    Parent directories are automatically created with proper permissions for multi-user access (HDFS/local).
@@ -125,23 +125,32 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
   /**
    * Generates a unique filename for centralized lineage storage.
    * 
-   * Format: {timestamp}_{appName}_{appId}
-   * Example: 2025-10-12_14-30-45-123_MySparkJob_app-20251012143045-0001
+   * Format: {timestamp}_{planId}_{appId}
+   * Example: 2025-10-12_14-30-45-123_550e8400-e29b-41d4-a716-446655440000_app-20251012143045-0001
    *
    * This format optimizes for operational debugging use cases:
    * - Timestamp FIRST: Ensures natural chronological sorting (most recent files appear together)
-   * - Application Name: Easy identification of which job generated the lineage
-   * - Application ID: Full traceability to specific Spark application run
+   * - Plan ID: Guarantees uniqueness even for concurrent writes from the same application
+   * - Application ID: Enables immediate filtering by Spark application (correlates with Spark UI, logs, monitoring)
    *
-   * @return A unique filename optimized for filtering and sorting
+   * The planId ensures zero collision risk even when:
+   * - Multiple writes happen in the same millisecond
+   * - The same application writes multiple datasets concurrently
+   * - Different applications run simultaneously
+   *
+   * The appId enables operators to quickly filter all lineage files from a specific Spark application run,
+   * making it trivial to correlate with Spark UI, logs, and monitoring systems where the human-readable
+   * application name is already available.
+   *
+   * @return A unique filename optimized for filtering, sorting, and guaranteed uniqueness
    */
   private def generateUniqueFilename(): String = {
     val sparkContext = SparkContext.getOrCreate()
-    val appName = sparkContext.appName.replaceAll("[^a-zA-Z0-9_-]", "_")
+    val planId = this._lastSeenPlan.id.get.toString
     val appId = sparkContext.applicationId
     val dateFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd_HH-mm-ss-SSS").withZone(ZoneId.of("UTC"))
     val timestamp = dateFormatter.format(Instant.now())
-    s"${timestamp}_${appName}_${appId}"
+    s"${timestamp}_${planId}_${appId}"
   }
 
   /**
diff --git a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
index 010184e9..d87693cb 100644
--- a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
+++ b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
@@ -107,7 +107,8 @@ class HDFSLineageDispatcherSpec
 
   it should "save lineage files in a custom lineage path" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
     val centralizedPath = TempDirectory("spline_centralized").deleteOnExit()
-
+    centralizedPath.delete()
+    
     withIsolatedSparkSession(_
       .config(lineageDispatcherConfigKeyName, lineageDispatcherConfigValueName)
       .config(lineageDispatcherConfigClassNameKeyName, classOf[HDFSLineageDispatcher].getName)
@@ -129,24 +130,27 @@ class HDFSLineageDispatcherSpec
           val centralizedDir = new File(centralizedPath.asString)
           centralizedDir.exists should be(true)
           centralizedDir.isDirectory should be(true)
-          
+
           val appId = spark.sparkContext.applicationId
-          val appName = spark.sparkContext.appName
-          val appNameCleaned = appName.replaceAll("[^a-zA-Z0-9_-]", "_")
 
           val lineageFiles = Option(centralizedDir.listFiles()).getOrElse(Array.empty[File])
           val lineageFilesOnly = lineageFiles.filter(f => f.isFile && !f.getName.endsWith(".crc"))
           lineageFilesOnly.length should be(2)
 
-          // Verify naming convention aligns with centralized lineage pattern (timestamp_appName_appId)
+          // Verify naming convention aligns with centralized lineage pattern (timestamp_planId_appId)
+          // Format: {timestamp}_{planId}_{appId}
+          // Example: 2025-10-12_14-30-45-123_550e8400-e29b-41d4-a716-446655440000_app-20251012143045-0001
           val filenamePattern = """\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}-\d{3}_.+_.+"""
           lineageFilesOnly.foreach { file =>
             val name = file.getName
-            withClue(s"Lineage filename '$name' should follow the timestamp_appName_appId pattern") {
+            withClue(s"Lineage filename '$name' should follow the timestamp_planId_appId pattern") {
               name.matches(filenamePattern) shouldBe true
             }
-            name should include appId
-            name should include appNameCleaned
+            // Verify the appId appears in the filename (ends with appId)
+            withClue(s"Lineage filename '$name' should end with application ID '$appId'") {
+              name.endsWith(appId) shouldBe true
+            }
+
           }
 
           // Verify each file has the correct format and content

From cd535c95a118f8db2fa89a0c325bf1abd85ef234 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Tue, 14 Oct 2025 00:03:21 +0100
Subject: [PATCH 22/25] Added more robust check to ensure no _LINEAGE file is
 created

---
 .../harvester/dispatcher/HDFSLineageDispatcherSpec.scala | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
index d87693cb..ef49ab3c 100644
--- a/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
+++ b/integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala
@@ -161,10 +161,11 @@ class HDFSLineageDispatcherSpec
           }
 
           // Verify no lineage files in destination directories
-          val lineageFileInDest1 = new File(destPath1.asString, "_LINEAGE")
-          val lineageFileInDest2 = new File(destPath2.asString, "_LINEAGE")
-          lineageFileInDest1.exists should be(false)
-          lineageFileInDest2.exists should be(false)
+          val dest1Files = Option(new File(destPath1.asString).listFiles()).getOrElse(Array.empty[File])
+          val dest2Files = Option(new File(destPath2.asString).listFiles()).getOrElse(Array.empty[File])
+          
+          dest1Files.exists(_.getName.contains("_LINEAGE")) should be(false)
+          dest2Files.exists(_.getName.contains("_LINEAGE")) should be(false)
         }
       }
     }

From 0e5c69a3842bcf26daf08f808aeb6603e6c87bc1 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Tue, 14 Oct 2025 00:08:46 +0100
Subject: [PATCH 23/25] Added getOrElse when obtaining planId in
 HDFSLineageDispatcher

---
 .../dispatcher/HDFSLineageDispatcher.scala    | 19 ++++++++++++++++---
 1 file changed, 16 insertions(+), 3 deletions(-)

diff --git a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
index e6fcf796..45391095 100644
--- a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
+++ b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
@@ -79,13 +79,24 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
   override def name = "HDFS"
 
   override def send(plan: ExecutionPlan): Unit = {
+    require(plan != null, "Execution plan cannot be null")
+    require(plan.id.isDefined, "Execution plan must have an ID")
     this._lastSeenPlan = plan
   }
 
   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")
+    if (this._lastSeenPlan == null) {
+      throw new IllegalStateException("send(event) must be called strictly after send(plan) method")
+    }
+    
+    val planId = this._lastSeenPlan.id.getOrElse(
+      throw new IllegalStateException("Execution plan ID is missing")
+    )
+    
+    if (planId != event.planId) {
+      throw new IllegalStateException(s"Plan ID mismatch: expected $planId but event has ${event.planId}")
+    }
 
     try {
       val path = resolveLineagePath()
@@ -146,7 +157,9 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
    */
   private def generateUniqueFilename(): String = {
     val sparkContext = SparkContext.getOrCreate()
-    val planId = this._lastSeenPlan.id.get.toString
+    val planId = this._lastSeenPlan.id.getOrElse(
+      throw new IllegalStateException("Execution plan ID is missing")
+    ).toString
     val appId = sparkContext.applicationId
     val dateFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd_HH-mm-ss-SSS").withZone(ZoneId.of("UTC"))
     val timestamp = dateFormatter.format(Instant.now())

From bebc869c2be839f828953c52cf3fac0c828a103a Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Tue, 14 Oct 2025 00:09:12 +0100
Subject: [PATCH 24/25] Added getOrElse when obtaining planId in
 HDFSLineageDispatcher

---
 .../dispatcher/HDFSLineageDispatcher.scala        | 15 ++-------------
 1 file changed, 2 insertions(+), 13 deletions(-)

diff --git a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
index 45391095..b975a3b1 100644
--- a/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
+++ b/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
@@ -79,24 +79,13 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
   override def name = "HDFS"
 
   override def send(plan: ExecutionPlan): Unit = {
-    require(plan != null, "Execution plan cannot be null")
-    require(plan.id.isDefined, "Execution plan must have an ID")
     this._lastSeenPlan = plan
   }
 
   override def send(event: ExecutionEvent): Unit = {
     // check state
-    if (this._lastSeenPlan == null) {
-      throw new IllegalStateException("send(event) must be called strictly after send(plan) method")
-    }
-    
-    val planId = this._lastSeenPlan.id.getOrElse(
-      throw new IllegalStateException("Execution plan ID is missing")
-    )
-    
-    if (planId != event.planId) {
-      throw new IllegalStateException(s"Plan ID mismatch: expected $planId but event has ${event.planId}")
-    }
+    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 = resolveLineagePath()

From 7bd5b0a25a5f13876576b1bc3c88ce9e151be6f1 Mon Sep 17 00:00:00 2001
From: rkrumins <kruminsrinalds@gmail.com>
Date: Tue, 14 Oct 2025 00:25:55 +0100
Subject: [PATCH 25/25] Updated spline.default.yaml as per up-to-date details
 for HDFSLineageDispatcher

---
 core/src/main/resources/spline.default.yaml | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/core/src/main/resources/spline.default.yaml b/core/src/main/resources/spline.default.yaml
index 4ea7165b..e458802d 100644
--- a/core/src/main/resources/spline.default.yaml
+++ b/core/src/main/resources/spline.default.yaml
@@ -151,23 +151,23 @@ spline:
       # If left empty, null, or not specified → DEFAULT MODE: lineage written alongside target data files
       # If set to a path → CENTRALIZED MODE: all lineage written to this location with unique filenames
       #
-      # CENTRALIZED MODE filename format: {timestamp}_{appName}_{appId}
+      # CENTRALIZED MODE filename format: {timestamp}_{planId}_{appId}
       # - timestamp: Human-readable UTC timestamp (yyyy-MM-dd_HH-mm-ss-SSS) for natural chronological sorting and easy filtering
       #   Example: 2025-10-12_14-30-45-123
-      # - appName: Spark application name for easy identification of which job generated the lineage
-      #   Example: MySparkJob
-      # - appId: Spark application ID for traceability to specific runs
+      # - planId: Execution plan UUID for guaranteed uniqueness (prevents collisions from concurrent writes)
+      #   Example: 550e8400-e29b-41d4-a716-446655440000
+      # - appId: Spark application ID for traceability to specific runs (correlates with Spark UI, logs, monitoring)
       #   Example: app-20251012143045-0001
       #
-      # Examples (assuming app name is "MySparkJob"):
+      # Examples:
       # - Local: customLineagePath: /my/centralized/lineage
-      #   Output: /my/centralized/lineage/2025-10-12_14-30-45-123_MySparkJob_app-20251012143045-0001
+      #   Output: /my/centralized/lineage/2025-10-12_14-30-45-123_550e8400-e29b-41d4-a716-446655440000_app-20251012143045-0001
       # - S3: customLineagePath: s3://my-bucket/lineage
-      #   Output: s3://my-bucket/lineage/2025-10-12_14-30-45-123_MySparkJob_app-20251012143045-0001
+      #   Output: s3://my-bucket/lineage/2025-10-12_14-30-45-123_550e8400-e29b-41d4-a716-446655440000_app-20251012143045-0001
       # - GCS: customLineagePath: gs://my-bucket/lineage
-      #   Output: gs://my-bucket/lineage/2025-10-12_14-30-45-123_MySparkJob_app-20251012143045-0001
+      #   Output: gs://my-bucket/lineage/2025-10-12_14-30-45-123_550e8400-e29b-41d4-a716-446655440000_app-20251012143045-0001
       # - HDFS: customLineagePath: hdfs://cluster/lineage
-      #   Output: hdfs://cluster/lineage/2025-10-12_14-30-45-123_MySparkJob_app-20251012143045-0001
+      #   Output: hdfs://cluster/lineage/2025-10-12_14-30-45-123_550e8400-e29b-41d4-a716-446655440000_app-20251012143045-0001
     # -------------------------------------------
     # Open Lineage HTTP dispatcher
     # -------------------------------------------