Hi @zhangshenghang , PTAL when you have time. Thanks!

Hi @dybyte, thanks for working on this. I reviewed the PR from a local branch against the current dev code path, including the full Zeta job-state lifecycle rather than only the diff.

What This PR Solves

• User pain point: today Zeta job state events are only emitted for terminal states, so an external platform cannot reliably observe PENDING, SCHEDULED, RUNNING, FAILING, CANCELING, or DOING_SAVEPOINT through the event stream.
• Fix approach: the PR adds event-report-http.report-non-terminal-job-state with default false, parses it into EngineConfig, and lets PhysicalPlan emit non-terminal JobStateEvents when the flag is enabled.
• One-line summary: this is a useful opt-in feature for job lifecycle observability, but the current hook point emits an incorrect event sequence on the normal startup path.

1. Code Change Review

1.1 Core Logic Analysis

Local review basis:

• Branch: seatunnel-review-10133
• Head: 468b149148c0db9245444910deb3a58cd7706818
• Merge base with upstream/dev: c84f62fb123e0f7aaf747f507806d6760e8f313c
• git diff --stat upstream/dev...HEAD: 8 files changed, 174 insertions, 16 deletions
• git diff --check upstream/dev...HEAD: clean
• Local git merge-tree against current upstream/dev: no conflict markers found
• GitHub checks: Build, labeler, Notify test workflow, and Label PRs when reviewed are all successful

The important code changes are:

• EngineConfig.java: adds reportNonTerminalJobState=false.
• YamlSeaTunnelDomConfigProcessor.java: parses report-non-terminal-job-state under event-report-http.
• PhysicalPlan.java: stores EngineConfig, extracts reportJobStateEvent(), and calls it at the end of stateProcess().
• JobStateEventTest.java: adds coverage for non-terminal event emission.
• English and Chinese event-listener docs describe the new option.

Before this PR, terminal events were emitted only in the terminal branch of stateProcess():

case FAILED:
case CANCELED:
case SAVEPOINT_DONE:
case FINISHED:
    stopJobStateProcess();
    jobEndFuture.complete(new JobResult(jobStatus, errorBySubPlan.get()));
    jobMaster
            .getCoordinatorService()
            .getEventProcessor()
            .process(
                    new JobStateEvent(
                            jobImmutableInformation.getJobId(),
                            jobImmutableInformation.getJobConfig().getName(),
                            jobStatus));
    return;

After this PR, the event reporting is moved to a common helper called from the tail of stateProcess():

case FAILED:
case CANCELED:
case SAVEPOINT_DONE:
case FINISHED:
    stopJobStateProcess();
    jobEndFuture.complete(new JobResult(jobStatus, errorBySubPlan.get()));
    break;
default:
    throw new IllegalArgumentException("Unknown Job State: " + jobStatus);
}
reportJobStateEvent(jobStatus);

private void reportJobStateEvent(JobStatus jobStatus) {
    try {
        if (jobStatus.isEndState()
                || (this.engineConfig != null
                        && this.engineConfig.isReportNonTerminalJobState())) {
            jobMaster
                    .getCoordinatorService()
                    .getEventProcessor()
                    .process(
                            new JobStateEvent(
                                    jobId,
                                    jobImmutableInformation.getJobConfig().getName(),
                                    jobStatus));
        }
    } catch (Exception e) {
        log.warn("Failed to report job {} state event", jobId, e);
    }
}

Key findings:

• The normal Zeta path does hit this PR: CoordinatorService.submitJob() writes PENDING, JobMaster.run() calls PhysicalPlan.startJob(), and PhysicalPlan drives SCHEDULED/RUNNING plus terminal states.
• The new config is default-off, so existing terminal-only behavior is mostly preserved.
• The feature is meant to report job state changes, but the implementation reports from stateProcess(), which is a state-processing function and can recursively advance the state.
• Because of that, the normal startup path misses PENDING, emits RUNNING before SCHEDULED, and emits RUNNING twice.
• The new test checks the count and last non-terminal status, but does not assert the complete event sequence, so it does not catch this issue.

Complete runtime flow:

Job submission
  -> CoordinatorService.submitJob(...) [L619-L691]
      -> new JobMaster(..., engineConfig, seaTunnelServer) [L635-L648]
      -> jobMaster.init(..., false) [L663-L664]
          -> PhysicalPlan constructor stores INITIALIZING/CREATED timestamps and CREATED state [PhysicalPlan L103-L119]
          -> PhysicalPlan.setJobMaster(jobMaster) stores engineConfig [L136-L140]
      -> pendingJobQueue.put(pendingJobInfo) [CoordinatorService L668-L670]
      -> updateJobState(PENDING) [L679]
          -> updateStateTimestamps(PENDING) + runningJobStateIMap.set(PENDING) [PhysicalPlan L268-L272]
          -> stateProcess()
              -> isRunning=false, so it returns before reportJobStateEvent() [L320-L324]
              -> PENDING is not emitted

Job start
  -> JobMaster.run() [JobMaster L540-L542]
      -> PhysicalPlan.startJob() [PhysicalPlan L308-L312]
          -> isRunning=true [L309]
          -> updateJobState(SCHEDULED) [L311]
              -> stateProcess()
                  -> jobStatus=SCHEDULED [L325]
                  -> startSubPlanStateProcess() [L332-L339]
                  -> updateJobState(RUNNING) [L340]
                      -> stateProcess()
                          -> jobStatus=RUNNING [L325]
                          -> reportJobStateEvent(RUNNING) [L360-L375]
                  -> reportJobStateEvent(SCHEDULED) [L360-L375]
          -> stateProcess() again [L312]
              -> current jobStatus=RUNNING
              -> reportJobStateEvent(RUNNING) again [L360-L375]

Pipeline completion/failure/cancel/savepoint
  -> addPipelineEndCallback() selects FINISHED/FAILED/CANCELED/SAVEPOINT_DONE [L172-L187]
      -> updateJobState(...)
          -> stateProcess()
              -> terminal branch completes jobEndFuture [L350-L355]
              -> reportJobStateEvent(terminal status) [L360-L375]

1.2 Compatibility Impact

Verdict: compatible by default, but the opt-in behavior needs a correctness fix.

• API: no public API/SPI method changes.
• Config: adds event-report-http.report-non-terminal-job-state, default false.
• Defaults: unchanged for existing users.
• Protocol: no REST/RPC protocol change.
• Serialization: JobStateEvent format is unchanged.
• Historical behavior: terminal events still emit by default. Event-handler exceptions are now caught and logged instead of being allowed to escape from terminal reporting, which is acceptable for a best-effort event path.

1.3 Performance / Side Effects

CPU, memory, and GC impact should be low because job state transitions are low frequency. Network traffic increases only when the option is enabled and an HTTP endpoint is configured. The main side effect is semantic rather than resource-related: the current duplicate RUNNING event creates unnecessary handler work and can confuse downstream systems. Also, custom SPI handlers are still invoked synchronously from the state-machine path, so slow handlers can extend the synchronized state-processing section.

1.4 Error Handling and Logs

The new try/catch around event reporting is reasonable: event handler failures are logged with the job id and stack trace and do not break job lifecycle processing. No sensitive data is logged.

Issue 1: Non-terminal events are missed, out of order, and duplicated on the normal startup path

• Location: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalPlan.java:360
• Description: This code is in the Zeta job state-machine path. The PR reports events at the end of stateProcess(), but stateProcess() is not equivalent to a single state transition. A normal submit path writes PENDING from CoordinatorService.java:679, then stateProcess() returns early because isRunning=false at PhysicalPlan.java:320-324, so PENDING is not emitted. Later, PhysicalPlan.startJob() at PhysicalPlan.java:308-312 calls updateJobState(SCHEDULED), which recursively calls updateJobState(RUNNING), and then startJob() calls stateProcess() once more. The resulting emitted sequence is RUNNING -> SCHEDULED -> RUNNING.
• Risk: Consumers that rebuild job lifecycle from events will see a missing PENDING, a “running before scheduled” sequence, and duplicate RUNNING. That directly undermines the core purpose of this PR and can trigger wrong monitoring, audit, or orchestration behavior.
• Suggested fix: emit reportJobStateEvent(targetState) after updateJobState() successfully persists the target state, not from the tail of stateProcess(). That makes the event stream follow the actual persisted state sequence and avoids duplicate reports. If terminal events must remain after jobEndFuture.complete(), keep a terminal-specific path, but non-terminal events should still be emitted once per successful state write. Please also strengthen the test to collect the full sequence and assert PENDING -> SCHEDULED -> RUNNING -> FINISHED without duplicate RUNNING, plus a failure sequence ending in FAILING -> FAILED.
• Severity: High

2. Code Quality

2.1 Style

The naming and config style are consistent with the surrounding code. The main quality concern is the location of the event emission: stateProcess() processes and advances states recursively, so it is not the right abstraction boundary for “state changed” notifications.

2.2 Tests

The added test proves that some non-terminal events can be observed, but it misses the critical contract:

• It does not assert the complete event order.
• It does not assert that PENDING is emitted.
• It does not assert that RUNNING is emitted only once.
• It does not cover cancel or savepoint event sequences.

I did not run the PR locally per the requested review mode; this is based on local source analysis and the GitHub check results.

2.3 Docs

Both English and Chinese docs were updated consistently:

• docs/en/engines/event-listener.md
• docs/zh/engines/event-listener.md

However, the current implementation does not satisfy the documented PENDING behavior or a reliable state-change sequence yet.

3. Architecture

3.1 Elegance

The feature direction is good and the default-off config is the right compatibility choice. The implementation is not yet precise because it hooks into state processing instead of state transition persistence.

3.2 Maintainability

Moving event emission to the successful updateJobState(targetState) path would make the behavior easier to reason about and test. The current placement makes future state-machine changes likely to accidentally duplicate or reorder events again.

3.3 Extensibility

If we later add transition reasons, duration, or pipeline-level state events, we need a clear “state changed” publication point. The current stateProcess() hook is too indirect for that.

3.4 Historical Compatibility

Existing users are compatible because the option defaults to false. No migration is needed, but users should not rely on non-terminal event ordering until Issue 1 is fixed.

4. Issue Summary

5. Merge Decision

Conclusion: can merge after fixes

1. Blocking items

   • Issue 1: the current hook point produces an incorrect non-terminal state event sequence on the normal Zeta job startup path.

2. Suggested non-blocking improvements

   • Add cancel and savepoint sequence tests, such as CANCELING -> CANCELED and DOING_SAVEPOINT -> SAVEPOINT_DONE.
   • Consider documenting that custom EventHandler implementations should avoid long blocking work because handlers are invoked from the state-machine path.

Overall, I like the feature goal and the compatibility shape. Once the event emission is tied to successful state transitions and the test asserts the real sequence, this should be a solid improvement.

Hi @DanielLeens , I have addressed the issues.