Resolve PR comments

Author: pflooky

app/src/main/scala/io/github/datacatering/datacaterer/core/sink/SinkFactory.scala

@@ -530,22 +530,36 @@ class SinkFactory(
   private def consolidateCsvWithHeaders(partFiles: Array[java.nio.file.Path], targetPath: java.nio.file.Path): Unit = {
     LOGGER.debug(s"Consolidating CSV files with header handling")
     val targetFile = Files.newOutputStream(targetPath)
+    val lineSeparator = System.lineSeparator().getBytes("UTF-8")
+    
     try {
-      partFiles.zipWithIndex.foreach { case (partFile, index) =>
+      var headerWritten = false
+      
+      partFiles.foreach { partFile =>
         val lines = Files.readAllLines(partFile)
-        if (index == 0) {
-          // First file: write all lines including header
-          lines.forEach(line => {
-            targetFile.write(line.getBytes("UTF-8"))
-            targetFile.write('\n')
-          })
+        
+        if (lines.isEmpty) {
+          LOGGER.debug(s"Skipping empty part file: ${partFile.getFileName}")
+        } else if (!headerWritten) {
+          // First non-empty file: write all lines including header
+          for (i <- 0 until lines.size()) {
+            targetFile.write(lines.get(i).getBytes("UTF-8"))
+            if (i < lines.size() - 1 || partFiles.length > 1) {
+              // Write line separator except for the last line of the last file
+              targetFile.write(lineSeparator)
+            }
+          }
+          headerWritten = true
         } else {
           // Subsequent files: skip first line (header) and write the rest
           if (lines.size() > 1) {
-            lines.subList(1, lines.size()).forEach(line => {
-              targetFile.write(line.getBytes("UTF-8"))
-              targetFile.write('\n')
-            })
+            for (i <- 1 until lines.size()) {
+              targetFile.write(lines.get(i).getBytes("UTF-8"))
+              if (i < lines.size() - 1 || partFiles.indexOf(partFile) < partFiles.length - 1) {
+                // Write line separator except for the last line of the last file
+                targetFile.write(lineSeparator)
+              }
+            }
           }
         }
       }

app/src/test/scala/io/github/datacatering/datacaterer/core/generator/metadata/datasource/opendatacontractstandard/OpenDataContractStandardDataSourceMetadataTest.scala

@@ -37,7 +37,8 @@ class OpenDataContractStandardDataSourceMetadataTest extends SparkSuite {
     assertResult(true)(resultCols.exists(_.field == "txn_ref_dt"))
     val txnDateCol = resultCols.filter(_.field == "txn_ref_dt").head
     val txnCluster = if (isVersion2) Map(CLUSTERING_POSITION -> "-1") else Map()
-    // v2 uses isNullable directly; v3 inverts required field
+    // v2: isNullable=false in test data → IS_NULLABLE="false"
+    // v3: required=false in test data → IS_NULLABLE="true" (mapper inverts: !false = true)
     val isNullableValue = if (isVersion2) "false" else "true"
     val expectedTxnDateMetadata = Map(
       IS_PRIMARY_KEY -> "false",
@@ -59,7 +60,8 @@ class OpenDataContractStandardDataSourceMetadataTest extends SparkSuite {
     assertResult(true)(resultCols.exists(_.field == "rcvr_id"))
     val rcvrIdCol = resultCols.filter(_.field == "rcvr_id").head
     val rcvrIdCluster = if (isVersion2) Map(CLUSTERING_POSITION -> "1") else Map()
-    // v2 uses isNullable directly; v3 inverts required field
+    // v2: isNullable=false in test data → IS_NULLABLE="false"
+    // v3: required=false in test data → IS_NULLABLE="true" (mapper inverts: !false = true)
     val expectedRcvrIdMetadata = Map(
       IS_PRIMARY_KEY -> "true",
       IS_NULLABLE -> isNullableValue,
@@ -77,7 +79,8 @@ class OpenDataContractStandardDataSourceMetadataTest extends SparkSuite {
     assertResult(true)(resultCols.exists(_.field == "rcvr_cntry_code"))
     val countryCodeCol = resultCols.filter(_.field == "rcvr_cntry_code").head
     val countryCodeCluster = if (isVersion2) Map(CLUSTERING_POSITION -> "-1") else Map()
-    // v2 uses isNullable directly; v3 inverts required field
+    // v2: isNullable=false in test data → IS_NULLABLE="false"
+    // v3: required=false in test data → IS_NULLABLE="true" (mapper inverts: !false = true)
     val expectedCountryCodeMetadata = Map(
       IS_PRIMARY_KEY -> "false",
       IS_NULLABLE -> isNullableValue,