[Fix][Engine] Add defensive null checks to JobDAGInfo toJsonObject

[Isha2790]

Purpose of this pull request

This pull request introduces comprehensive defensive null-checks in JobDAGInfo.java inside the toJsonObject() method to prevent potential NullPointerException (NPE) runtime failures during internal job configuration serialization or cluster state reporting via the REST API.

Specifically, this PR ensures that:

• Deep nested paths (e.g., vertexInfo.getType().getType()) check for intermediary null references.
• Potential null wrappers like jobId, vertexId, and list elements are safely handled with clean ternary operators rather than crashing the thread with unboxing or direct method access failures.
• Optional parameters like envOptions fallback safely to an empty JSON structure when absent.

Does this PR introduce any user-facing change?

No. This is a purely internal stabilization fix that hardens runtime execution safety. It does not alter behavior or configuration formats for users.

How was this patch tested?

• This changes basic string-formatting serialization utility methods inside the seatunnel-engine-core sub-module.
• The module has been compiled and run locally to verify code correctness.
• The codebase formatting guidelines have been strictly preserved by executing mvn spotless:apply locally before final commit.

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If necessary, please update incompatible-changes.md to describe the incompatibility caused by this PR.
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config

[DanielLeens]

Thanks for your first contribution to Apache SeaTunnel! The approach of adding defensive null checks to JobDAGInfo.toJsonObject() makes sense — when JobDAGInfo is constructed via the no-arg constructor (e.g. in tests or deserialization edge cases), pipelineEdges can be null and the entrySet() call would throw an NPE.

I have one concern before we can merge:

Issue 1: Incomplete null-check coverage in toJsonObject()

• Location: seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/job/JobDAGInfo.java:67-70
• The PR adds null guards for pipelineEdges, entry.getValue(), and individual edge objects, but the method still dereferences vertexInfoMap (line 69), jobId (line 67), and envOptions (line 68) without any null check. Since JobDAGInfo has a @NoArgsConstructor, all fields can be null. If the goal is defensive null safety, these fields need the same treatment. For example, vertexInfoMap.entrySet() will NPE if vertexInfoMap is null — and the existing test BaseServiceTableMetricsTest already creates JobDAGInfo instances with only vertexInfoMap set and leaves pipelineEdges null.
• Suggested fix: add null guards for vertexInfoMap (skip the loop if null), jobId (use String.valueOf(jobId) or a null check), and envOptions (pass an empty JsonObject if null).

CI note: The Build check currently shows action_required — this is expected for first-time contributors. Once a maintainer approves the workflow run, CI will proceed. No action needed from you on that front.

Keep up the good work! Looking forward to the next iteration.

[DanielLeens]

Thanks for your first contribution to Apache SeaTunnel! The approach of adding defensive null checks to JobDAGInfo.toJsonObject() makes sense — when JobDAGInfo is constructed via the no-arg constructor (e.g. in tests or deserialization edge cases), pipelineEdges can be null and the entrySet() call would throw an NPE.

I have one concern before we can merge:

Issue 1: Incomplete null-check coverage in toJsonObject()

• Location: seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/job/JobDAGInfo.java:67-70
• The PR adds null guards for pipelineEdges, entry.getValue(), and individual edge objects, but the method still dereferences vertexInfoMap (line 69), jobId (line 67), and envOptions (line 68) without any null check. Since JobDAGInfo has a @NoArgsConstructor, all fields can be null. If the goal is defensive null safety, these fields need the same treatment. For example, vertexInfoMap.entrySet() will NPE if vertexInfoMap is null — and the existing test BaseServiceTableMetricsTest already creates JobDAGInfo instances with only vertexInfoMap set and leaves pipelineEdges null.
• Suggested fix: add null guards for vertexInfoMap (skip the loop if null), jobId (use String.valueOf(jobId) or a null check), and envOptions (pass an empty JsonObject if null).

CI note: The Build check currently shows action_required — this is expected for first-time contributors. Once a maintainer approves the workflow run, CI will proceed. No action needed from you on that front.

Keep up the good work! Looking forward to the next iteration.

[dybyte]

Please enable CI following the instruction.

[DanielLeens]

Thanks for the CI ping.

The current Build run is still action_required, so please enable CI workflows in your fork first using this exact check entry:
https://github.com/apache/seatunnel/runs/86324931510

One important note from Daniel's side: enabling the workflow is not enough by itself for this PR yet. My earlier blocker on JobDAGInfo.toJsonObject() still stands on the current unchanged head, because the method still dereferences jobId, envOptions, and vertexInfoMap without the same null-safe treatment that was added for pipelineEdges. So after CI is enabled, I still need a code update for that null-coverage gap before I can clear the review.

[SEZ9]

Thanks for working on this. I re-reviewed the latest head 5ac02727f834 from scratch.

What this PR fixes

• User pain: REST/state-reporting endpoints that serialize a job's DAG can crash the handler thread with an NPE when JobDAGInfo contains null maps, null edge lists, or null edges, returning 500s instead of job info.
• Fix approach: Wrap the pipelineEdges iteration in toJsonObject() with null guards at three levels (the map itself, each entry's edge list, and each edge), and fall back to an empty-string JSON key when the pipeline id key is null.
• One-line summary: The change does harden the pipelineEdges loop, but it is incomplete against its own stated goals — edge.getInputVertexId()/getTargetVertexId() can still NPE on .toString() inside the new guard, the promised protections for jobId, vertexId, envOptions, and nested vertexInfo.getType().getType() paths are absent from the diff, and serializing a null pipeline id as key "" emits a malformed payload that downstream REST/UI consumers cannot map to a pipeline.

Runtime chain I rechecked

GET /hazelcast/rest/maps/job-info/{jobId}
  RestHttpGetCommandProcessor.handleJobInfoById()  RestHttpGetCommandProcessor.java
    RestHttpGetCommandProcessor.convertToJson(jobInfo, jobId)  RestHttpGetCommandProcessor.java
      DAGUtils.getJobDAGInfo(jobImmutableInformation, ...)  DAGUtils.java  (constructs JobDAGInfo with pipelineEdges/vertexInfoMap)
      JobDAGInfo.toJsonObject()  JobDAGInfo.java:47
        pipelineEdges null-guard + entry iteration  JobDAGInfo.java:50-63
          edge.getInputVertexId().toString() / edge.getTargetVertexId().toString()  JobDAGInfo.java:57-58  <-- still unguarded NPE if vertex ids are null
        pipelineEdgesJsonObject.add(entry.getKey() != null ? ... : "", jsonArray)  JobDAGInfo.java:62  <-- emits "" key for null pipeline id

Findings

Issue 1: NPE is still possible inside the newly guarded block: edge.getInputVertexId()/getTargetVertexId() can be null and .toString() will throw

• Location: seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/job/JobDAGInfo.java:57
• Why it matters: - Root cause: the new if (edge != null) guard only protects against a null Edge element, but the very next lines still dereference edge.getInputVertexId().toString() and edge.getTargetVertexId().toString() without any check. Edge vertex IDs are Long wrappers that can legitimately be null (e.g., a source/sink boundary edge or a partially built DAG), which is exactly the class of failure this PR claims to fix.
• Impact: an NPE thrown here bubbles out of toJsonObject() and breaks the Zeta REST endpoints / job info reporting that serialize JobDAGInfo, so the 'defensive' fix does not actually remove the crash scenario it targets — it only moves it two lines down.
• Fix: extract the IDs first (Long inputId = edge.getInputVertexId();), and either skip the edge with a log.warn when either ID is null, or serialize a null-safe string via String.valueOf(inputId). Skipping with a warning is preferable so consumers do not receive edges with fabricated "null" vertex IDs.
• Evidence: seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/job/JobDAGInfo.java:58 edgeJsonObject.add("targetVertexId", edge.getTargetVertexId().toString());; seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/job/JobDAGInfo.java:55 if (edge != null) {
• Suggested fix: The edge != null guard is not sufficient here: edge.getInputVertexId() and edge.getTargetVertexId() return Long wrappers that can themselves be null, so .toString() on the next two lines can still throw an NPE — the exact failure mode this PR sets out to prevent. Please null-check the IDs as well, e.g. skip the edge with a log.warn("Edge with null vertex id in pipeline {} skipped", entry.getKey()) when either ID is null, or use String.valueOf(...) if you intend to serialize them regardless.
• Severity: High

Issue 2: PR description promises null checks for jobId/vertexId/envOptions and nested type paths, but the diff only guards the pipelineEdges loop — most of the claimed NPE fixes are missing

• Location: seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/job/JobDAGInfo.java:50
• Why it matters: - Root cause: the description states this PR handles nulls for jobId, vertexId, envOptions, and deep paths like vertexInfo.getType().getType(), yet the only changed file/lines are the pipelineEdges loop in JobDAGInfo.toJsonObject(). None of those other paths appear in the diff.
• Impact: reviewers and users reading the changelog would believe the REST API DAG serialization is fully hardened, while the majority of the described NPE surfaces remain unguarded in the Zeta engine's job info reporting.
• Fix: either implement the remaining guards described in the PR body (jobId, envOptions fallback, vertex type null handling) in this PR, or rewrite the PR description to accurately scope it to the pipelineEdges loop only. Also please link the GitHub issue that reproduces the NPE, per the contribution checklist.
• Evidence: seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/job/JobDAGInfo.java:55 if (edge != null) {
• Suggested fix: The PR description says this change adds null guards for jobId, vertexId, envOptions, and vertexInfo.getType().getType(), but the diff only covers the pipelineEdges loop. Please either add the remaining guards described in the PR body or update the description to match the actual scope, and link the issue that reproduces the NPE.
• Severity: High

Issue 3: A null pipeline id is serialized as an empty-string JSON key "", producing a malformed DAG payload that REST/UI consumers cannot correlate to any pipeline.

• Location: seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/job/JobDAGInfo.java:63
• Why it matters: - Root cause: line 63 substitutes "" when entry.getKey() is null: pipelineEdgesJsonObject.add(entry.getKey() != null ? entry.getKey().toString() : "", jsonArray).
• Impact: the pipelineEdges map in the REST response is keyed by pipeline id (an integer). Emitting "": [...] silently changes the JSON contract — the SeaTunnel web UI and any automation parsing the job DAG will encounter a key that parses to no valid pipeline id, which is arguably worse than a loud failure because the edges get attached to a phantom pipeline.
• A null key here can only come from an engine-side construction bug (pipeline ids are assigned by the LogicalDag/plan builder), so masking it hides the real defect.
• Fix: skip the entry with a WARN log (log.warn("Skip pipeline edges with null pipeline id in job {}", jobId)) instead of inventing an empty key, keeping the JSON contract intact.
• Suggested fix: Falling back to an empty-string key changes the JSON contract of pipelineEdges (keys are pipeline ids consumed by the UI/REST clients). A "": [...] entry cannot be correlated to any pipeline and silently corrupts the payload. Since a null key can only come from an engine-side construction bug, please skip the entry with a WARN log instead of emitting an empty key.
• Severity: Medium

Review conclusion

Conclusion: can merge after the blocking items are fixed

1. Blocking items

• Issue 1: NPE is still possible inside the newly guarded block: edge.getInputVertexId()/getTargetVertexId() can be null and .toString() will throw
• Issue 2: PR description promises null checks for jobId/vertexId/envOptions and nested type paths, but the diff only guards the pipelineEdges loop — most of the claimed NPE fixes are missing

2. Suggested (non-blocking) follow-ups

• Issue 3: A null pipeline id is serialized as an empty-string JSON key "", producing a malformed DAG payload that REST/UI consumers cannot correlate to any pipeline.

Nice progress overall — once the blocking points above are addressed this should be in good shape. Happy to discuss.

[DanielLeens]

Thanks @SEZ9. I rechecked the same head 5ac02727f834 after your review, and I agree the current null-safety fix is still incomplete.

Your additional point about edge.getInputVertexId() and edge.getTargetVertexId() is valid: the new edge != null guard is not enough if either vertex-id wrapper itself is null before .toString() is called. I also agree the PR description currently overstates the scope of the fix relative to the actual diff on this head.

So from Daniel's side the current unchanged head still needs a code update before merge:

1. close the remaining vertex-id null-dereference gap inside the guarded edge loop
2. narrow or complete the PR scope so the description matches the actual null-safety coverage on this revision

The Build workflow is still action_required here:
https://github.com/apache/seatunnel/runs/86324931510

But enabling CI alone will not clear the PR yet — the source-side blocker remains open on this head.

[DanielLeens]

Rechecked the current head 5ac02727f834 against the latest dev after the CI / merge-gate fact changed.

Runtime path rechecked

same runtime path as the previous Daniel review on this unchanged head

Merge conclusion

Conclusion: still needs fixes before merge

1. Blocking items

• The source-side blockers from my previous Daniel review still stand on this unchanged head.
• Separately, this branch is behind upstream (compare status: diverged, ahead_by=1, behind_by=10), and the current failing gate is FAILURE gate: Build: https://github.com/apache/seatunnel/runs/86324931510

2. Lowest-cost next step

• Please sync with the latest dev and rerun the failing gate first so we can separate baseline CI noise from the remaining code-side fixes.
• After that, if the same source blocker is still unresolved, I would still keep this PR out of merge-ready state from Daniel's side.