Oct 29408 qa - #918
Conversation
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
…irectory Co-authored-by: Cursor <cursoragent@cursor.com>
- Safely extract writeOptions using Try().getOrElse - Add wildcard fallback to capture unknown V2 WriteCommands like ReplaceData - Add safe null checking for extracting location path Co-authored-by: Cursor <cursoragent@cursor.com>
⛔ Snyk checks have failed. 12 issues have been found so far.
💻 Catch issues earlier using the plugins for VS Code, JetBrains IDEs, Visual Studio, and Eclipse. |
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
WalkthroughThe PR updates Maven modules to Scala 2.11, hardens lineage dispatch and listener error handling, rewrites HDFS lineage persistence to use run-scoped atomic writes, and adds a Java example that runs Spark with the HDFS dispatcher. ChangesScala 2.11 migration and runtime hardening
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant JavaExampleJob_Files
participant SparkLineageInitializer
participant HDFSLineageDispatcher
participant cleanupOldRunFolders
participant persistRunFolderAtomically
participant writeFileAtomically
JavaExampleJob_Files->>SparkLineageInitializer: enableLineageTracking(...)
JavaExampleJob_Files->>HDFSLineageDispatcher: send(plan/event)
HDFSLineageDispatcher->>cleanupOldRunFolders: remove old run directories
HDFSLineageDispatcher->>persistRunFolderAtomically: create runDir and lineage file
persistRunFolderAtomically->>writeFileAtomically: write temp file then rename
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 inconclusive)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala`:
- Line 79: The HDFSLineageDispatcher is ignoring the configured output name
because `fileName` is hardcoded to `lineage_${executionPlanID}.json`. Update
`HDFSLineageDispatcher` so it uses the constructor/configured `fileName` value
when building the HDFS target path, keeping the existing `fileName` symbol and
replacing the inline literal with the configured contract.
- Around line 121-139: The current persistRunFolderAtomically implementation in
HDFSLineageDispatcher only makes the lineage file write atomic, not the whole
run-folder materialization, because it creates finalRunDir before writing the
file. Update persistRunFolderAtomically to follow the documented temp-directory
flow by writing into a hidden temp dir and renaming it to the final run
directory, or else adjust the method/comment to explicitly promise only
file-level atomicity. Ensure the behavior matches the guarantee implied by
writeFileAtomically and pathStringToFsWithPath.
- Around line 107-111: The cleanup in HDFSLineageDispatcher’s status iteration
is too broad because it deletes every directory whose name differs from the
currentRunID, which can remove unrelated run outputs. Update the logic around
the statuses.foreach block to only remove clearly stale temporary/TTL-expired
folders, or move this cleanup out of the write path entirely; keep the check in
HDFSLineageDispatcher scoped to safe, owned temp artifacts instead of sibling
run directories.
- Around line 157-162: The durability hints in writeFileAtomically are being
applied unconditionally, which can break writes on FileSystem implementations
that do not support them. Update the try block in
HDFSLineageDispatcher.writeFileAtomically to treat out.hflush() and out.hsync()
as best-effort by either checking the filesystem capability first or catching
UnsupportedOperationException around those calls, then continue with the
existing close and temp-file rename flow regardless.
In
`@core/src/main/scala/za/co/absa/spline/harvester/plugin/embedded/DataSourceV2Plugin.scala`:
- Around line 190-194: The fallback URI in DataSourceV2Plugin’s SourceIdentifier
construction currently hardcodes "unknown", which causes different V2 tables to
collide when path and properties.location are missing. Update the URI derivation
logic in the table scanning/identifier-building flow so it uses a table-specific
fallback or leaves the URI absent instead of defaulting to the literal string,
while keeping the provider lookup and SourceIdentifier creation intact.
In
`@examples/src/main/java/za/co/absa/spline/example/batch/JavaExampleJob_Files.java`:
- Around line 31-82: The main method in JavaExampleJob_Files creates a
SparkSession but does not ensure it is closed, leaving resources open on failure
or exit. Update the SparkSession session lifecycle in main so the job uses a
try/finally (or equivalent cleanup) around the Spark work and always calls
session.stop() in the cleanup path, keeping the existing
SparkLineageInitializer.enableLineageTracking and SparkSession.builder setup
intact.
- Line 42: The lineage directory in the Java example is hardcoded to a
Windows-only path, so update JavaExampleJob_Files to build the dispatcher URI
from a portable source such as java.io.tmpdir or a configurable argument instead
of using the fixed file:///C:/spline value. Keep the change localized to the
Spark config setup in the example job so the lineageDispatcher setting works
across Windows, Linux, and macOS.
In `@pom.xml`:
- Around line 147-148: The root build has been switched to Scala 2.11, but the
default cross-build properties still point to 2.12, so update the default Scala
version and binary version properties in pom.xml to match the existing
scala.version and scala.binary.version values; keep the change aligned with the
cross-build configuration used by scala-cross-build-maven-plugin so the default
coordinates and build defaults agree.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 1b62859b-f7ae-4695-a815-982144468b6a
📒 Files selected for processing (11)
bundle-2.2/pom.xmlbundle-2.3/pom.xmlbundle-2.4/pom.xmlcommons/pom.xmlcore/pom.xmlcore/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scalacore/src/main/scala/za/co/absa/spline/harvester/plugin/embedded/DataSourceV2Plugin.scalaexamples/pom.xmlexamples/src/main/java/za/co/absa/spline/example/batch/JavaExampleJob_Files.javaintegration-tests/pom.xmlpom.xml
| // Best-effort cleanup of previous runs | ||
| cleanupOldRunFolders(appDir, runID) | ||
|
|
||
| val fileName = s"lineage_${executionPlanID}.json" |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Keep the configured filename contract effective.
HDFSLineageDispatcher still requires fileName, but this hardcoded lineage_${executionPlanID}.json path ignores the constructor/config value, so existing dispatcher configuration no longer controls output naming.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala`
at line 79, The HDFSLineageDispatcher is ignoring the configured output name
because `fileName` is hardcoded to `lineage_${executionPlanID}.json`. Update
`HDFSLineageDispatcher` so it uses the constructor/configured `fileName` value
when building the HDFS target path, keeping the existing `fileName` symbol and
replacing the inline literal with the configured contract.
| statuses.foreach { status => | ||
| if (status.isDirectory && status.getPath.getName != currentRunID) { | ||
| logInfo(s"Deleting old runID folder: ${status.getPath}") | ||
| fs.delete(status.getPath, true) // recursive delete | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Avoid deleting every non-current run folder.
This removes all sibling run IDs under the same plan directory, which can erase lineage history or another active application’s run output. Restrict cleanup to stale temp folders, TTL-expired folders, or remove this cleanup from the write path.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala`
around lines 107 - 111, The cleanup in HDFSLineageDispatcher’s status iteration
is too broad because it deletes every directory whose name differs from the
currentRunID, which can remove unrelated run outputs. Update the logic around
the statuses.foreach block to only remove clearly stale temporary/TTL-expired
folders, or move this cleanup out of the write path entirely; keep the check in
HDFSLineageDispatcher scoped to safe, owned temp artifacts instead of sibling
run directories.
| * Atomically create a run directory with its lineage JSON file inside. | ||
| * Strategy: | ||
| * 1) Ensure parent exists (mkdirs is idempotent) | ||
| * 2) Create a hidden temp directory next to the target | ||
| * 3) Write JSON to a temp file, then rename temp file -> final name inside temp dir | ||
| * 4) Rename temp dir -> final run dir (atomic on HDFS) | ||
| */ | ||
| private def persistRunFolderAtomically(finalRunDirStr: String, fileName: String, content: String): Unit = blocking { | ||
| val (fs, finalRunDir) = pathStringToFsWithPath(finalRunDirStr) | ||
|
|
||
| val umask = FsPermission.getUMask(fs.getConf) | ||
| FsPermission.getFileDefault.applyUMask(umask) | ||
| // Ensure final run directory exists | ||
| if (!fs.exists(finalRunDir)) { | ||
| fs.mkdirs(finalRunDir) | ||
| try fs.setPermission(finalRunDir, permission) catch { case _: Throwable => () } | ||
| } | ||
|
|
||
| // Write the file atomically inside the run directory | ||
| val finalFileInRunDir = new Path(finalRunDir, fileName) | ||
| writeFileAtomically(fs, finalFileInRunDir, content.getBytes(StandardCharsets.UTF_8)) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Match the atomic run-folder guarantee in the implementation.
The comment describes temp-directory materialization, but the code creates finalRunDir first and then writes a file inside it. Consumers can observe an empty/partial run directory after failures. Either implement temp-dir rename or downgrade the guarantee to file-level atomicity.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala`
around lines 121 - 139, The current persistRunFolderAtomically implementation in
HDFSLineageDispatcher only makes the lineage file write atomic, not the whole
run-folder materialization, because it creates finalRunDir before writing the
file. Update persistRunFolderAtomically to follow the documented temp-directory
flow by writing into a hidden temp dir and renaming it to the final run
directory, or else adjust the method/comment to explicitly promise only
file-level atomicity. Ensure the behavior matches the guarantee implied by
writeFileAtomically and pathStringToFsWithPath.
| try { | ||
| out.write(bytes) | ||
| // Best-effort durability hints; on HDFS these are meaningful. | ||
| out.hflush() | ||
| out.hsync() | ||
| } finally { |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify the Hadoop/Spark versions used by this repo before confirming FSDataOutputStream sync behavior.
rg -n '<hadoop.version>|hadoop-common|hadoop-client|spark-core' pom.xml */pom.xmlRepository: AbsaOSS/spline-spark-agent
Length of output: 1612
🏁 Script executed:
#!/bin/bash
set -euo pipefail
FILE="core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala"
echo "== file context =="
sed -n '120,190p' "$FILE"
echo
echo "== hflush/hsync usages =="
rg -n 'hflush|hsync|supports[A-Za-z]+|hasPathCapability|isInstanceOf|FileSystem' core/src/main/scala
echo
echo "== hadoop-related dependency context =="
sed -n '1,120p' pom.xml
echo
sed -n '1,110p' core/pom.xml
echo
for f in bundle-2.2/pom.xml bundle-2.3/pom.xml bundle-2.4/pom.xml bundle-3.0/pom.xml bundle-3.1/pom.xml bundle-3.3/pom.xml bundle-3.4/pom.xml bundle-3.5/pom.xml; do
if [ -f "$f" ]; then
echo
echo "### $f"
rg -n 'hadoop-client|hadoop-common|spark-core|hadoop.version|scala.binary.version' "$f"
fi
doneRepository: AbsaOSS/spline-spark-agent
Length of output: 16242
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Check Hadoop API docs available in repository-local sources if any mention these methods.
rg -n 'def hflush|def hsync|hflush\(|hsync\(' .Repository: AbsaOSS/spline-spark-agent
Length of output: 387
🌐 Web query:
Hadoop FSDataOutputStream hflush hsync unsupported filesystem implementation IOException documentation
💡 Result:
In Hadoop, the behavior of FSDataOutputStream.hflush and hsync on filesystems that do not support these operations is governed by the implementation of the Syncable interface [1][2]. Historically, and in some current implementations, FSDataOutputStream silently downgrades hflush and hsync calls to a standard OutputStream.flush if the underlying stream does not implement Syncable [3][2]. This behavior is discouraged because it can mislead applications into assuming data is visible or durable when it is not [2]. Modern guidance and best practices for filesystem implementers specify that if a filesystem does not support sync/flush operations, it should implement the Syncable interface but throw an UnsupportedOperationException instead of performing a silent downgrade [4][2][5]. Some modern connectors (such as certain S3A implementations) have moved toward throwing UnsupportedOperationException to ensure callers are explicitly aware that the operation is not supported [5]. To verify support for these operations programmatically before invoking them, developers should use the StreamCapabilities interface [4][6][2]. By calling hasCapability("hflush") or hasCapability("hsync"), an application can safely check if the stream supports these guarantees without risking an unexpected exception or a misleading silent failure [4][2].
Citations:
- 1: https://hadoop.apache.org/docs/current/api/org/apache/hadoop/fs/Syncable.html
- 2: https://hadoop.apache.org/docs/current3/hadoop-project-dist/hadoop-common/filesystem/outputstream.html
- 3: https://hadoop.apache.org/docs/r2.7.1/api/src-html/org/apache/hadoop/fs/FSDataOutputStream.html
- 4: https://hadoop.apache.org/docs/current/hadoop-project-dist/hadoop-common/filesystem/outputstream.html
- 5: HadoopPositionOutputStream calls FSDataOutputStream.hflush() in close() apache/parquet-java#3203
- 6: https://hadoop.apache.org/docs/stable2/hadoop-project-dist/hadoop-common/filesystem/filesystem.html
🏁 Script executed:
#!/bin/bash
set -euo pipefail
FILE="core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala"
echo "== file context =="
sed -n '120,190p' "$FILE"
echo
echo "== hflush/hsync usages =="
rg -n 'hflush|hsync|supports[A-Za-z]+|hasPathCapability|isInstanceOf|FileSystem' core/src/main/scalasrc/main/scala 2>/dev/null || true
rg -n 'hflush|hsync|supports[A-Za-z]+|hasPathCapability|isInstanceOf|FileSystem' core/src/main/scala
echo
echo "== hadoop-related dependency context =="
sed -n '1,120p' pom.xml
echo
sed -n '1,110p' core/pom.xml
echo
for f in bundle-2.2/pom.xml bundle-2.3/pom.xml bundle-2.4/pom.xml bundle-3.0/pom.xml bundle-3.1/pom.xml bundle-3.3/pom.xml bundle-3.4/pom.xml bundle-3.5/pom.xml; do
if [ -f "$f" ]; then
echo
echo "### $f"
rg -n 'hadoop-client|hadoop-common|spark-core|hadoop.version|scala.binary.version' "$f"
fi
doneRepository: AbsaOSS/spline-spark-agent
Length of output: 16242
Make hflush/hsync truly best-effort in writeFileAtomically
out.hflush() / out.hsync() are invoked on every FileSystem implementation here, but some supported connectors can reject those calls and abort the write. Guard them by capability (or catch UnsupportedOperationException) and still close + rename the temp file.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala`
around lines 157 - 162, The durability hints in writeFileAtomically are being
applied unconditionally, which can break writes on FileSystem implementations
that do not support them. Update the try block in
HDFSLineageDispatcher.writeFileAtomically to treat out.hflush() and out.hsync()
as best-effort by either checking the filesystem capability first or catching
UnsupportedOperationException around those calls, then continue with the
existing close and temp-file rename flow regardless.
| } else { | ||
| "unknown" | ||
| } | ||
| val provider = Option(tableProps.get("provider")) | ||
| SourceIdentifier(provider, uri) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify how SourceIdentifier URIs participate in source-node identity.
rg -n -C3 'SourceIdentifier|uris|format' core commonsRepository: AbsaOSS/spline-spark-agent
Length of output: 50383
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the V2 plugin logic around the fallback URI.
ast-grep outline core/src/main/scala/za/co/absa/spline/harvester/plugin/embedded/DataSourceV2Plugin.scala --view expanded
sed -n '150,230p' core/src/main/scala/za/co/absa/spline/harvester/plugin/embedded/DataSourceV2Plugin.scala
# Inspect SourceIdentifier and its consumers to see how URIs affect identity/building.
sed -n '1,120p' core/src/main/scala/za/co/absa/spline/harvester/builder/SourceIdentifier.scala
sed -n '1,140p' core/src/main/scala/za/co/absa/spline/harvester/builder/read/PluggableReadCommandExtractor.scala
sed -n '1,140p' core/src/main/scala/za/co/absa/spline/harvester/builder/write/PluggableWriteCommandExtractor.scala
sed -n '1,120p' core/src/main/scala/za/co/absa/spline/harvester/builder/plan/read/ReadNodeBuilder.scala
sed -n '1,120p' core/src/main/scala/za/co/absa/spline/harvester/builder/plan/write/WriteNodeBuilder.scala
# Search for places where SourceIdentifier or its URIs are used as keys or deduped.
rg -n "SourceIdentifier\\(|sourceIdentifier\\.|\\.uris\\b|inputSources|outputSource|distinct\\(|groupBy\\(|toSet\\b|Map\\(" core/src/main/scala core/src/test/scalaRepository: AbsaOSS/spline-spark-agent
Length of output: 50382
Avoid the literal "unknown" fallback. Distinct V2 tables that share a provider and lack both path and properties.location will all emit SourceIdentifier(provider, "unknown"), making them indistinguishable in lineage payloads. Prefer a table-specific fallback or omit the URI when it isn’t available.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@core/src/main/scala/za/co/absa/spline/harvester/plugin/embedded/DataSourceV2Plugin.scala`
around lines 190 - 194, The fallback URI in DataSourceV2Plugin’s
SourceIdentifier construction currently hardcodes "unknown", which causes
different V2 tables to collide when path and properties.location are missing.
Update the URI derivation logic in the table scanning/identifier-building flow
so it uses a table-specific fallback or leaves the URI absent instead of
defaulting to the literal string, while keeping the provider lookup and
SourceIdentifier creation intact.
| public static void main(String[] args) { | ||
|
|
||
|
|
||
|
|
||
| final SparkSession session = SparkSession.builder() | ||
| .appName("java example app") | ||
| .master("local[*]") | ||
| .config("spark.driver.host", "localhost") | ||
| .config("spark.spline.mode", "ENABLED") // ✅ Enable Spline | ||
| .config("spark.spline.lineageDispatcher", "hdfs") // ✅ Use HDFS dispatcher | ||
| .config("spark.spline.lineageDispatcher.hdfs.className", "za.co.absa.spline.harvester.dispatcher.HDFSLineageDispatcher") // 🔥 Fix: Define HDFS Dispatcher | ||
| .config("spark.spline.lineageDispatcher.hdfs.directory", "file:///C:/spline") // ✅ Store lineage files in local storage | ||
| .config("spark.spline.lineageDispatcher.hdfs.fileNamePrefix", "lineage_") // ✅ File naming pattern | ||
| .getOrCreate(); | ||
|
|
||
|
|
||
|
|
||
| // Enable Spline Tracking | ||
| AgentConfig splineConfig = AgentConfig.builder().build(); | ||
| SparkLineageInitializer.enableLineageTracking(session, splineConfig); | ||
| System.out.println("Spline Listeners: " + splineConfig.getConfigurationListeners().size()); | ||
|
|
||
| System.out.println("Using HDFSLineageDispatcher from: " + | ||
| HDFSLineageDispatcher.class.getProtectionDomain().getCodeSource().getLocation()); | ||
|
|
||
|
|
||
|
|
||
|
|
||
|
|
||
|
|
||
| System.out.println("Spline Mode: " + session.conf().get("spark.spline.mode", "NOT SET")); | ||
| System.out.println("Spline HDFS: " + session.conf().get("spark.spline.lineageDispatcher.hdfs.directory", "NOT SET")); | ||
|
|
||
|
|
||
| // Explicitly enable Spline lineage tracking | ||
| // This step is optional - see https://github.com/AbsaOSS/spline-spark-agent#programmatic-initialization | ||
|
|
||
|
|
||
|
|
||
| // Sample Spark Job | ||
| session.read() | ||
| .option("header", "true") | ||
| .option("inferSchema", "true") | ||
| .csv("data/input/batch/wikidata.csv") | ||
| .as("source") | ||
| .write() | ||
| .mode(SaveMode.Overwrite) | ||
| .csv("data/output/batch/java-sample3.csv"); | ||
|
|
||
| // printLineageFiles("/tmp/spline"); | ||
|
|
||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Stop the Spark session in a finally block.
The example creates a local Spark session but never releases it, which can leave local resources and file handles open after failures.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@examples/src/main/java/za/co/absa/spline/example/batch/JavaExampleJob_Files.java`
around lines 31 - 82, The main method in JavaExampleJob_Files creates a
SparkSession but does not ensure it is closed, leaving resources open on failure
or exit. Update the SparkSession session lifecycle in main so the job uses a
try/finally (or equivalent cleanup) around the Spark work and always calls
session.stop() in the cleanup path, keeping the existing
SparkLineageInitializer.enableLineageTracking and SparkSession.builder setup
intact.
| .config("spark.spline.mode", "ENABLED") // ✅ Enable Spline | ||
| .config("spark.spline.lineageDispatcher", "hdfs") // ✅ Use HDFS dispatcher | ||
| .config("spark.spline.lineageDispatcher.hdfs.className", "za.co.absa.spline.harvester.dispatcher.HDFSLineageDispatcher") // 🔥 Fix: Define HDFS Dispatcher | ||
| .config("spark.spline.lineageDispatcher.hdfs.directory", "file:///C:/spline") // ✅ Store lineage files in local storage |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Use a portable local lineage directory.
file:///C:/spline is Windows-specific and maps poorly on Linux/macOS. Build the URI from java.io.tmpdir or accept it as an argument.
Proposed fix
+ final String lineageDir = new File(System.getProperty("java.io.tmpdir"), "spline").toURI().toString();
final SparkSession session = SparkSession.builder()
@@
- .config("spark.spline.lineageDispatcher.hdfs.directory", "file:///C:/spline") // ✅ Store lineage files in local storage
+ .config("spark.spline.lineageDispatcher.hdfs.directory", lineageDir) // ✅ Store lineage files in local storage📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| .config("spark.spline.lineageDispatcher.hdfs.directory", "file:///C:/spline") // ✅ Store lineage files in local storage | |
| final String lineageDir = new File(System.getProperty("java.io.tmpdir"), "spline").toURI().toString(); | |
| .config("spark.spline.lineageDispatcher.hdfs.directory", lineageDir) // ✅ Store lineage files in local storage |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@examples/src/main/java/za/co/absa/spline/example/batch/JavaExampleJob_Files.java`
at line 42, The lineage directory in the Java example is hardcoded to a
Windows-only path, so update JavaExampleJob_Files to build the dispatcher URI
from a portable source such as java.io.tmpdir or a configurable argument instead
of using the fixed file:///C:/spline value. Keep the change localized to the
Spark config setup in the example job so the lineageDispatcher setting works
across Windows, Linux, and macOS.
| <scala.version>2.11.12</scala.version> | ||
| <scala.binary.version>2.11</scala.binary.version> |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== root cross-build properties =="
sed -n '138,178p' pom.xml
echo
echo "== scala-binary-dependent Spark/Delta coordinates in reviewed POMs =="
rg -n 'scala\.binary\.version|spark-(30|31|32|33|34|35)\.version|delta-(30|31|32|33|34|35)\.version' \
pom.xml commons/pom.xml core/pom.xml examples/pom.xml integration-tests/pom.xml bundle-2.2/pom.xml bundle-2.3/pom.xml bundle-2.4/pom.xmlRepository: AbsaOSS/spline-spark-agent
Length of output: 12539
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== pom.xml around Scala/Spark profile matrix =="
sed -n '620,910p' pom.xml
echo
echo "== plugin and cross-build property references =="
rg -n 'defaultScalaBinaryVersion|defaultScalaVersion|scala-cross-build|scala\.binary\.version|scala\.version' pom.xmlRepository: AbsaOSS/spline-spark-agent
Length of output: 16675
🌐 Web query:
scala-cross-build-maven-plugin defaultScalaBinaryVersion defaultScalaVersion semantics
💡 Result:
The scala-cross-build-maven-plugin is a utility designed to assist with Scala cross-version builds in Maven projects [1][2]. Regarding the specific properties defaultScalaBinaryVersion and defaultScalaVersion, these are not core semantic configuration parameters of the plugin itself, but rather common naming conventions used by developers within their Maven POM files to manage cross-compilation profiles [3][4][5]. In the context of cross-building, the following semantics typically apply: 1. Property Semantics: - defaultScalaBinaryVersion: Usually defines the Scala binary version (e.g., 2.12, 2.13) that the project defaults to when no specific profile is activated [3][5]. - defaultScalaVersion: Often maps to the specific full Scala version (e.g., 2.12.18) corresponding to the default binary version [3][5]. 2. Plugin Usage: The scala-cross-build-maven-plugin itself focuses on goals such as changing the Scala version in your pom.xml (via change-version) and restoring it (via restore-version) [1][2]. It relies on Maven profiles to handle the actual switching of dependencies and plugin configurations [1][3]. 3. Best Practices: - Developers typically define these defaults in the properties section of their pom.xml so that IDEs and standard builds have a stable target [3]. - When building for other Scala versions, developers use Maven profiles to override these default properties [3][4]. - While some alternative plugins (like the scala-maven-plugin) have built-in properties like scala.version and scala.compat.version [6], the scala-cross-build-maven-plugin is generally used as a helper to modify the project structure dynamically to achieve similar results [1][2]. If you are implementing this, ensure your POM correctly uses Maven property interpolation to reference these variables in your dependency and plugin sections [3].
Citations:
- 1: https://github.com/wajda/scala-cross-build-maven-plugin
- 2: https://github.com/witokondoria/scala-cross-build-maven-plugin
- 3: https://github.com/kelnos/scala-cross-maven-plugin
- 4: Best way to cross compile davidB/scala-maven-plugin#177
- 5: https://repo1.maven.org/maven2/za/co/absa/abris_2.13/6.4.0/abris_2.13-6.4.0.pom
- 6: https://davidb.github.io/scala-maven-plugin/cc-mojo.html
Align the cross-build defaults with the 2.11 switch. default.scala.binary.version / default.scala.version still point at 2.12 while the root build is now 2.11, so scala-cross-build-maven-plugin and the checked-in coordinates disagree. Update the defaults to the 2.11 values too.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pom.xml` around lines 147 - 148, The root build has been switched to Scala
2.11, but the default cross-build properties still point to 2.12, so update the
default Scala version and binary version properties in pom.xml to match the
existing scala.version and scala.binary.version values; keep the change aligned
with the cross-build configuration used by scala-cross-build-maven-plugin so the
default coordinates and build defaults agree.
Catch and log Spline listener, dispatch, and cleanup failures so unexpected lineage issues do not fail Spark job execution. Co-authored-by: Cursor <cursoragent@cursor.com>
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
core/src/main/scala/za/co/absa/spline/agent/SplineAgent.scala (1)
90-100: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winGate event dispatch on a successful plan send
core/src/main/scala/za/co/absa/spline/agent/SplineAgent.scala:94-98The separate catches still allow an
ExecutionEventto be dispatched after the plan send fails. For HTTP/Kafka backends, that can persist an event without its plan and leave a dangling lineage record.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/src/main/scala/za/co/absa/spline/agent/SplineAgent.scala` around lines 90 - 100, The plan and event dispatch in SplineAgent’s harvest callback are still independent, so an ExecutionEvent can be sent even when the corresponding plan send fails. Update the logic around harvester.harvest(...).foreach and the two withErrorHandling blocks so the event dispatch only runs after lineageDispatcher.send(plan) succeeds, using the existing execution plan dispatch and execution event dispatch symbols to keep the send order explicit and gated.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@core/src/main/scala/za/co/absa/spline/harvester/dispatcher/CompositeLineageDispatcher.scala`:
- Around line 46-61: `CompositeLineageDispatcher.delegate` only fails fast when
a child dispatcher throws, but `HttpLineageDispatcher`,
`KafkaLineageDispatcher`, and `FallbackLineageDispatcher` currently swallow
`NonFatal` errors and only log them. Update those dispatcher implementations so
their dispatch methods propagate failures back to `delegate` when `failOnErrors`
should apply, and keep the handling centralized around the
`delegate`/`failOnErrors` flow rather than hiding exceptions inside the child
dispatchers.
---
Outside diff comments:
In `@core/src/main/scala/za/co/absa/spline/agent/SplineAgent.scala`:
- Around line 90-100: The plan and event dispatch in SplineAgent’s harvest
callback are still independent, so an ExecutionEvent can be sent even when the
corresponding plan send fails. Update the logic around
harvester.harvest(...).foreach and the two withErrorHandling blocks so the event
dispatch only runs after lineageDispatcher.send(plan) succeeds, using the
existing execution plan dispatch and execution event dispatch symbols to keep
the send order explicit and gated.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 65d5eed9-387e-454a-92f4-b9b6f3e94eea
📒 Files selected for processing (7)
core/src/main/scala/za/co/absa/spline/agent/SplineAgent.scalacore/src/main/scala/za/co/absa/spline/harvester/dispatcher/CompositeLineageDispatcher.scalacore/src/main/scala/za/co/absa/spline/harvester/dispatcher/FallbackLineageDispatcher.scalacore/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scalacore/src/main/scala/za/co/absa/spline/harvester/dispatcher/HttpLineageDispatcher.scalacore/src/main/scala/za/co/absa/spline/harvester/dispatcher/KafkaLineageDispatcher.scalacore/src/main/scala/za/co/absa/spline/harvester/listener/SplineQueryExecutionListener.scala
🚧 Files skipped from review as they are similar to previous changes (1)
- core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala
| private def delegate(call: LineageDispatcher => Unit): Unit = { | ||
| delegatees.foreach(withErrorHandling(call)) | ||
| } | ||
| val failures = delegatees.flatMap { disp => | ||
| try { | ||
| call(disp) | ||
| None | ||
| } catch { | ||
| case NonFatal(e) => | ||
| logWarning(s"Proceeding after an error occurred in an underlying dispatcher: ${disp.getClass.getName}", e) | ||
| Some(e) | ||
| } | ||
| } | ||
|
|
||
| private def withErrorHandling(call: LineageDispatcher => Unit): LineageDispatcher => Unit = { disp => | ||
| try call(disp) | ||
| catch { | ||
| case NonFatal(e) => | ||
| if (failOnErrors) throw e | ||
| else logWarning(s"Proceeding after an error occurred in an underlying dispatcher: ${disp.getClass.getName}", e) | ||
| if (failOnErrors) { | ||
| failures.headOption.foreach(e => throw e) | ||
| } | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map relevant files
git ls-files core/src/main/scala/za/co/absa/spline/harvester/dispatcher | sed -n '1,200p'
echo '--- CompositeLineageDispatcher.scala ---'
wc -l core/src/main/scala/za/co/absa/spline/harvester/dispatcher/CompositeLineageDispatcher.scala
cat -n core/src/main/scala/za/co/absa/spline/harvester/dispatcher/CompositeLineageDispatcher.scala | sed -n '1,220p'
echo '--- dispatcher implementations ---'
for f in \
core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HttpLineageDispatcher.scala \
core/src/main/scala/za/co/absa/spline/harvester/dispatcher/KafkaLineageDispatcher.scala \
core/src/main/scala/za/co/absa/spline/harvester/dispatcher/FallbackLineageDispatcher.scala \
core/src/main/scala/za/co/absa/spline/harvester/dispatcher/LineageDispatcher.scala
do
if [ -f "$f" ]; then
echo "### $f"
wc -l "$f"
cat -n "$f" | sed -n '1,240p'
fi
doneRepository: AbsaOSS/spline-spark-agent
Length of output: 26323
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Find all references to failOnErrors and dispatcher send implementations
rg -n "failOnErrors|FailOnErrorsKey|def send\\(|override def send|catch \\{[[:space:]]*case NonFatal|NonFatal\\(" core/src/main/scala/za/co/absa/spline/harvester/dispatcher core/src/main/scala/za/co/absa/spline/harvester | sed -n '1,240p'Repository: AbsaOSS/spline-spark-agent
Length of output: 14139
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect config/docs around failOnErrors semantics
rg -n "failOnErrors|FailOnErrorsKey|fail on errors|failOnErrors" core src docs . | sed -n '1,240p'Repository: AbsaOSS/spline-spark-agent
Length of output: 4066
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Determine whether delegatees can propagate failures through return values instead of exceptions
ast-grep outline core/src/main/scala/za/co/absa/spline/harvester/dispatcher/LineageDispatcher.scala --view expanded || true
ast-grep outline core/src/main/scala/za/co/absa/spline/harvester/dispatcher/CompositeLineageDispatcher.scala --view expanded || trueRepository: AbsaOSS/spline-spark-agent
Length of output: 367
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Quick semantic probe: see if any underlying dispatchers return any value indicating failure
python3 - <<'PY'
from pathlib import Path
files = [
Path("core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HttpLineageDispatcher.scala"),
Path("core/src/main/scala/za/co/absa/spline/harvester/dispatcher/KafkaLineageDispatcher.scala"),
Path("core/src/main/scala/za/co/absa/spline/harvester/dispatcher/FallbackLineageDispatcher.scala"),
]
for p in files:
if p.exists():
print(f"### {p}")
text = p.read_text()
for i, line in enumerate(text.splitlines(), 1):
if "def send" in line or "override def send" in line or "NonFatal" in line or "Try(" in line or "Either" in line:
print(f"{i}: {line}")
PYRepository: AbsaOSS/spline-spark-agent
Length of output: 1581
🏁 Script executed:
#!/bin/bash
set -euo pipefail
for f in \
core/src/main/scala/za/co/absa/spline/harvester/dispatcher/CompositeLineageDispatcher.scala \
core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HttpLineageDispatcher.scala \
core/src/main/scala/za/co/absa/spline/harvester/dispatcher/KafkaLineageDispatcher.scala \
core/src/main/scala/za/co/absa/spline/harvester/dispatcher/FallbackLineageDispatcher.scala \
core/src/main/scala/za/co/absa/spline/harvester/dispatcher/LineageDispatcher.scala
do
if [ -f "$f" ]; then
echo "### $f"
cat -n "$f" | sed -n '1,220p' | sed -n '/def send\|override def send\|delegate(call: LineageDispatcher => Unit)\|NonFatal\|failOnErrors/p'
echo
fi
doneRepository: AbsaOSS/spline-spark-agent
Length of output: 2369
failOnErrors never sees failures from the HTTP/Kafka/Fallback dispatchers
CompositeLineageDispatcher can only fail fast when a child dispatcher throws, but HttpLineageDispatcher, KafkaLineageDispatcher, and FallbackLineageDispatcher catch NonFatal internally and only log. That makes failOnErrors = true ineffective for those implementations. If this setting is supposed to surface dispatch errors, these dispatchers need to propagate failure instead of swallowing it.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@core/src/main/scala/za/co/absa/spline/harvester/dispatcher/CompositeLineageDispatcher.scala`
around lines 46 - 61, `CompositeLineageDispatcher.delegate` only fails fast when
a child dispatcher throws, but `HttpLineageDispatcher`,
`KafkaLineageDispatcher`, and `FallbackLineageDispatcher` currently swallow
`NonFatal` errors and only log them. Update those dispatcher implementations so
their dispatch methods propagate failures back to `delegate` when `failOnErrors`
should apply, and keep the handling centralized around the
`delegate`/`failOnErrors` flow rather than hiding exceptions inside the child
dispatchers.
Co-authored-by: Cursor <cursoragent@cursor.com>
|
4 participants
Summary
ReplaceDataoperations causing Spline to crash.writeOptionsusingTry().getOrElseso missing fields don't throwNoSuchFieldExceptionReplaceDatainstead of crashing.Test plan
UPDATErow-level operations locally, verified Spline generated lineage correctly instead of crashing.Summary by CodeRabbit
New Features
Bug Fixes
file://paths.Chores