DanielLeens commented on PR #10364:
URL: https://github.com/apache/seatunnel/pull/10364#issuecomment-4538752338
Thanks for keeping this work alive. I reviewed the current head (`475c3b48`)
from scratch against the engine scheduling, restore/takeover, serialization,
and test paths. I also read the earlier discussion, and I largely agree with
the blockers already raised by @DanielCarter-stack — after tracing the current
call chain locally, I still do not think this revision is safe to merge yet.
# What this PR fixes
- User pain: when one job contains multiple pipelines, SeaTunnel currently
has no explicit way to cap how many pipelines are allowed to run at the same
time.
- Fix approach: this PR adds `env.pipeline_concurrency`, threads it through
`JobConfig`, and changes `PhysicalPlan` from “start every CREATED pipeline” to
“start up to N pipelines”.
- One-line summary: the direction makes sense, but the current head changes
engine scheduling and serialization boundaries without closing the correctness
and compatibility gaps.
# Runtime chain I checked
~~~text
Normal startup
-> JobMaster.run() [JobMaster.java:540-542]
-> PhysicalPlan.startJob() [PhysicalPlan.java:337-341]
-> stateProcess() [PhysicalPlan.java:349-364]
-> startNextPipelines() [PhysicalPlan.java:406-417]
Restore / master-takeover path
-> CoordinatorService submits a RESTORE job
[CoordinatorService.java:311-321]
-> each SubPlan.restorePipelineState() is called first
[SubPlan.java:522-560]
-> RUNNING pipelines continue from the restored state
-> JobMaster.run() then calls PhysicalPlan.startJob() again
-> startNextPipelines() still sees runningPipelineCount = 0 and
nextPipelineIndex = 0
-> serial / limited mode can start a later CREATED pipeline even though an
earlier pipeline is already RUNNING
Completion accounting
-> FINISHED callback in PhysicalPlan.addPipelineEndCallback()
[PhysicalPlan.java:178-200]
-> finishedPipelineNum is incremented once in the FINISHED branch
-> and then incremented again unconditionally at line 200
~~~
# Findings
Issue 1: pipeline completion is counted twice, so the job-level end-state
decision becomes incorrect
- Location:
`seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalPlan.java:178-200`
- Why this matters: a single FINISHED pipeline increments
`finishedPipelineNum` in the FINISHED branch and then increments it again
immediately afterward at line 200.
- Risk: the job can satisfy `finishedPipelineNum == pipelineList.size()`
earlier than it should, which can prematurely drive the job into its
terminal-state branch.
- Better fix: increment the completed-pipeline counter exactly once per
finished pipeline and reuse that value for both logging and end-state checks.
- Severity: High
- Related review history: `+1` to @DanielCarter-stack's earlier point here.
Issue 2: `JobConfig` changes the `IdentifiedDataSerializable` field order
without any compatibility handling
- Location:
`seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/JobConfig.java:37,52-64`
- Why this matters: this is a real engine serialization boundary. The PR
inserts `pipelineConcurrency` between `jobContext` and `envOptions`, so old
readers and new writers no longer agree on the binary layout.
- Risk: rolling upgrades, takeover, and cross-version deserialization can
misread the payload and break recovery.
- Better fix: add explicit version-aware serialization, or at minimum
introduce a backward-compatible read path before changing the layout.
- Severity: High
- Related review history: `+1` to @DanielCarter-stack's earlier point here.
Issue 3: the restore / takeover path can bypass the new concurrency limit
- Location: `CoordinatorService.java:311-321`, `SubPlan.java:522-560`,
`PhysicalPlan.java:91-94,337-363,406-417`
- Why this matters: on master switch, RUNNING pipelines are restored first,
but `PhysicalPlan.startJob()` still restarts the job-level scheduling path with
`runningPipelineCount = 0` and `nextPipelineIndex = 0`. In serial / limited
mode, that means a later CREATED pipeline can be started even though an earlier
pipeline is already RUNNING.
- Risk: the limit that is supposed to serialize or cap pipeline execution is
violated exactly on the restore/takeover path, which is one of the most
sensitive runtime paths in SeaTunnel engine.
- Better fix: rebuild the scheduler counters from the current pipeline
states before calling `startNextPipelines()`, and do not consume
`nextPipelineIndex` as if a skipped non-CREATED pipeline were a free slot.
- Severity: High
- Related review history: `+1` to @DanielCarter-stack's earlier
restore-state concern, with the concrete call chain above.
Issue 4: the new E2E tests only assert log strings, so they do not prove
that concurrency control actually works
- Location:
`seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/PipelineConcurrencyIT.java:43-48,61-68,82-87,100-104`
- Why this matters: all four tests pass as long as the expected log text
appears. They would still pass if serial / limited execution were violated but
the logging still said `[Serial Mode]` or `[Limited Concurrency=2]`.
- Risk: false-green coverage on a scheduler feature. This is not a classic
flaky `sleep()` problem, but it is still a weak signal and a real regression
hole.
- Better fix: add behavior-level assertions, for example using timestamps or
explicit task markers to prove that no more than `N` pipelines are running
concurrently.
- Severity: Medium
- Related review history: this is related to @dybyte's earlier request for
stronger tests.
# Test stability note
- Rating: `Risk present`
- Basis: `PipelineConcurrencyIT.java:43-48,61-68,82-87,100-104` only checks
log strings, so the wait/assertion signal is too weak for the behavior this PR
is changing.
- This is not the highest-flakiness pattern, but it is still a test-quality
risk because the suite can go green while the scheduler logic is wrong.
# CI note
- The branch is currently `CONFLICTING`.
- The latest `Build` check is still failing, so even aside from the
source-level blockers above, the branch is not in a merge-ready state yet.
# Merge conclusion
### Conclusion: can merge after fixes
1. Blocking items
- Issue 1: fix the double increment in `finishedPipelineNum`.
- Issue 2: make the `JobConfig` serialization change backward compatible.
- Issue 3: make the restore / takeover path honor the configured concurrency
limit.
- Please also rebase the branch cleanly and get the current `Build` green.
2. Suggested non-blocking follow-up
- Issue 4: strengthen the tests so they verify actual concurrency behavior
rather than only log text.
- I would also recommend adding clearer comments around the new scheduler
state fields, since this is core engine lifecycle logic.
Overall, I like the goal of making pipeline concurrency configurable, but
the current revision still has correctness gaps on the main engine path. Happy
to re-review once you push an update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
