danielnadean opened a new issue, #11777:
URL: https://github.com/apache/seatunnel/issues/11777
### Search before asking
- [x] I searched open and closed SeaTunnel issues and STIPs for Zeta member
removal, worker failover, retry counting, dirty jobs, recovery attempts, and
failure history.
- [x] STIP-15 covers dirty **data** collection, while this proposal defines
an operational health label for a Zeta job. STIP-33 covers task failure
history, but does not define job-level member-loss recovery counting or
dirty-job semantics.
## Summary
This STIP proposes a job-scoped **recovery-degraded** label, also referred
to as a **dirty job**, for SeaTunnel Zeta.
A job becomes dirty when the number of job-level deployment attempts
associated with unexpected or unclassified Zeta member loss reaches a
configured threshold. The label is additive metadata: it does not replace
`JobStatus`, change retry limits, or stop the job.
The proposal provides:
- a precise distinction between member loss, recovery episodes, and
deployment attempts;
- job-level counting that does not multiply one recovery by the number of
pipelines or task groups;
- a stable attempt identity that survives active-master changes;
- HA-backed running state and finished-job history;
- explicit `CLEAN`, `DIRTY`, `UNKNOWN`, and `DISABLED` evaluation results;
- additive REST fields for running and finished jobs;
- an optional future path for trusted planned-leave exclusion.
The term `dirty job` in this STIP is unrelated to dirty-data collection. It
describes a job whose runtime continuity has degraded because member-loss
recovery has happened too many times.
## Motivation
In a Zeta cluster with two or more members, a worker process crash, machine
failure, network isolation, Kubernetes Pod restart, or rolling replacement can
remove a Hazelcast member. The current recovery path is approximately:
```text
Hazelcast memberRemoved
-> SeaTunnelServer.memberRemoved(...)
-> CoordinatorService.memberRemoved(...)
-> failedTaskOnMemberRemoved(...)
-> task groups on the lost address become FAILED
-> SubPlan checks the restore limit
-> prepareRestorePipeline()
-> resources are requested and the pipeline is deployed again
```
The engine already restores pipelines and keeps an in-memory
`pipelineRestoreNum`, but it does not provide a durable job-level answer to
these questions:
1. How many job-level deployment attempts were caused by member loss?
2. Did one member-loss event affect multiple pipelines without being
over-counted?
3. Did one failed deployment lead to another deployment attempt without
being under-counted?
4. Did the tracking remain complete across an active-master change?
5. Can operators query the result after the job finishes?
This leaves repeatedly recovered streaming jobs looking healthy as long as
their primary `JobStatus` returns to `RUNNING`.
## Current behavior and source boundaries
The current source establishes several important constraints:
- `SeaTunnelServer.memberRemoved(...)` forwards the Hazelcast membership
event to the active coordinator.
- `CoordinatorService.failedTaskOnMemberRemoved(...)` finds task groups
deployed on the lost address and marks eligible states as `FAILED`.
- `SubPlan.prepareRestorePipeline()` increments `pipelineRestoreNum` before
resource release, resource request, and redeployment.
- `pipelineRestoreNum` is pipeline-scoped and in-memory. It is not a
job-level member-loss counter and is not exposed through finished-job history.
- `MembershipServiceEvent` provides the lost member address and UUID, but
not a Kubernetes termination reason, process exit code, or trusted
planned-maintenance flag.
- The current Kubernetes resource manager does not classify member-removal
causes.
Therefore, `memberRemoved` proves that a Hazelcast member left the cluster.
It does **not** prove that a machine failed.
## Goals
1. Mark a job dirty after `X` job-level deployment attempts associated with
member loss.
2. Count deployment attempts, not task-group failures, pipeline count, or
raw member-removal count.
3. Preserve member UUID and address as diagnostic context without using
address as member identity.
4. Avoid mixing ordinary connector, data, or user-code failures into this
counter.
5. Preserve the tracking state across active-master changes and
partition-owner changes.
6. Expose the same evaluation summary for running and finished jobs.
7. Keep the existing job state machine, retry behavior, checkpoint format,
and REST fields backward compatible.
8. Fail open for job recovery but fail closed for observability: tracking
failure must not block failover, and it must not be reported as `CLEAN`.
## Non-goals
The first implementation will not:
- add a new `JobStatus` such as `DIRTY_RUNNING`;
- stop, cancel, or suspend a dirty job automatically;
- replace Hazelcast membership failure detection with a new heartbeat
detector;
- diagnose whether a member left because of OOM, machine failure, network
partition, or Kubernetes policy;
- count ordinary business failures that have no member-loss cause;
- add Prometheus metrics or Web UI filtering in the first backend PR;
- implement trusted planned-leave exclusion in the first backend PR;
- change `job.retry.times` semantics.
## Terminology
### Member-loss incident
A `MemberLossIncident` records that one Hazelcast member UUID left and
affected one job. Its stable identity is:
```text
jobId + lostMemberUuid
```
The incident is diagnostic and idempotent. It is not the dirty counter.
### Recovery episode
A `RecoveryEpisode` is one continuous member-loss recovery process for a
job. Multiple member-loss incidents and multiple affected pipelines may belong
to the same episode.
The episode begins when the coordinator registers the first affected
member-loss incident. It completes when all affected pipelines have deployed
successfully and recovered, or aborts when the job can no longer restore.
### Deployment attempt
A `RecoveryDeploymentAttempt` is one concrete redeployment attempt for one
affected pipeline. Restore preparation allocates a stable identity:
```text
attemptId = episodeId + pipelineId + prepareSequence
```
`prepareSequence` is incremented atomically in job-scoped dirty state. The
attempt is counted only when the prepared pipeline reaches the recovery
`DEPLOYING` stage after resource allocation.
`prepareRestorePipeline()` alone does not count because no task group has
been redeployed at that point.
### Job-level attempt generation
Within one recovery episode, every affected pipeline maintains its own
confirmed prepare sequence. The episode's job-level generation is the maximum
confirmed sequence:
- multiple pipelines entering their first deployment attempt produce
generation `1`, not `N`;
- if any pipeline requires a second deployment attempt, generation becomes
`2`;
- confirming the same stable attempt ID again is a no-op.
The job's `recoveryAttemptCount` increases only when a confirmed generation
advances. It continues accumulating across completed episodes for the same job
ID.
## Dirty evaluation
Proposed evaluation values:
| Value | Meaning |
| --- | --- |
| `DISABLED` | The job did not enable dirty-job tracking. |
| `CLEAN` | Tracking is complete and the threshold has not been reached. |
| `DIRTY` | The threshold has been reached. This is irreversible for the
current job ID. |
| `UNKNOWN` | Tracking is incomplete or inconsistent, so the job must not be
reported as clean. |
The primary job status remains independent. For example, a job can be both
`RUNNING` and `DIRTY`.
## Cause classification and Kubernetes semantics
The first implementation uses these cause semantics:
| Scenario | Classification | Count after deployment starts |
| --- | --- | --- |
| OOM, process crash, liveness restart | `UNCLASSIFIED` | Yes |
| Manual Pod deletion without prior trusted intent | `UNCLASSIFIED` | Yes |
| Rolling update without a trusted drain protocol | `UNCLASSIFIED` | Yes |
| Ordinary task failure without `memberRemoved` | No incident | No |
Unknown cause is intentionally counted because the engine cannot prove that
the interruption was planned.
A later phase may add `PLANNED_VERIFIED`, but only through an authenticated
and authorized control plane that registers the exact member UUID, request ID,
actor, reason, creation time, and short expiration time before termination. A
Pod `preStop` hook or JVM shutdown hook must not be trusted to classify its own
exit.
If one recovery episode contains any `UNCLASSIFIED` incident, its deployment
attempts are counted. The episode may be excluded only when every incident is
`PLANNED_VERIFIED`.
## State model
The first implementation introduces a dedicated HA-backed job state keyed by
job ID:
```text
engine_jobDirtyState: IMap<Long, DirtyJobState>
```
`DirtyJobState` contains at least:
- job ID and configured threshold;
- `dirty`, `evaluationStatus`, and `trackingComplete`;
- cumulative `recoveryAttemptCount`;
- recovery episode and member-loss incident counts;
- the active recovery episode;
- a bounded recent-episode history;
- first dirty time and last counted time;
- the last processed member-event sequence.
All updates for one job use a single-key `EntryProcessor` or an equivalent
atomic operation. The processor must update synchronous backups. The map
configuration must be installed consistently on every member before the map is
first obtained.
The state is bounded. Incident and episode history has a configured maximum,
task-group details are sampled, and stale incidents that never reach deployment
are expired or marked aborted.
## Tracking completeness and active-master recovery
Job-scoped atomic updates do not form a distributed transaction with task
and pipeline state. This proposal uses idempotent, at-least-once convergence
instead of changing the existing job state machine into a distributed
transaction.
A global monotonically increasing member-event sequence is stored using an
existing serializable type in running engine state. For every `memberRemoved`
event:
1. The active coordinator increments the global sequence before scanning
jobs.
2. Every dirty-enabled running job advances its own processed sequence after
the event is handled.
3. Affected jobs register or merge the incident in the same atomic
dirty-state update.
4. If dirty-state update fails, the job watermark remains behind the global
sequence.
5. A missing dirty state or a lagging watermark evaluates to `UNKNOWN`,
never `CLEAN`.
Dirty state is initialized before a dirty-enabled job becomes schedulable.
This prevents a clean default from being inferred for a job that never created
its state.
After an active-master change, reconciliation runs before restored pipelines
are scheduled:
- a persisted `PREPARED` attempt is compared with the pipeline state;
- `DEPLOYING` or `RUNNING` confirms the original stable attempt ID
idempotently;
- pre-deployment states retain the binding and wait for normal deployment;
- terminal non-restorable state aborts the episode;
- missing state, watermark lag, incompatible serialization, or an
unexplained transition evaluates to `UNKNOWN`.
Tracking-store failure must be logged and surfaced, but it must not block
the existing task failure or failover path.
## Configuration
The dirty threshold is separate from the retry limit:
```hocon
env {
job.retry.times = 5
job.dirty-task.threshold = 2
}
```
Proposed options:
| Option | Default | Meaning |
| --- | ---: | --- |
| `job.dirty-task.threshold` | `-1` | Disabled when less than or equal to
zero. |
| `engine.dirty-task.event-history-size` | `10` | Maximum recent episodes
retained per job. |
| `engine.dirty-task.pending-incident-ttl` | `10m` | Maximum lifetime of an
incident that never reaches deployment. |
`job.retry.times` remains the restore limit. `job.dirty-task.threshold` only
controls the health label.
## REST and history contract
Existing REST fields remain unchanged. Running and finished job detail adds
an optional object:
```json
{
"dirtyTask": {
"dirty": true,
"evaluationStatus": "DIRTY",
"trackingComplete": true,
"threshold": 2,
"recoveryAttemptCount": 2,
"recoveryEpisodeCount": 1,
"memberLossIncidentCount": 1,
"activeEpisodeId": null,
"lastLeaveClassification": "UNCLASSIFIED",
"lastLostMemberUuid": "23cc7ea6-6374-4f2b-8a33-df75b5e58f52",
"lastLostAddress": "10.0.0.12:5801",
"lastCountedTime": 1786500000000
}
}
```
Job lists should expose only `dirty`, `evaluationStatus`, and
`recoveryAttemptCount`. Recent episode detail belongs only in job detail to
avoid unbounded list responses.
When a job finishes, `JobHistoryService` snapshots dirty state together with
the finished-job lifecycle. Running dirty state is removed only after the final
history write. Older finished-job records without this field deserialize as
tracking disabled or absent, without changing existing fields.
## Compatibility and rollout
This proposal is additive and disabled by default:
- existing jobs require no configuration change;
- `JobStatus`, retry limits, checkpoint/savepoint payloads, and existing
REST fields are unchanged;
- no connector SPI changes are required;
- old finished-job records remain readable;
- no new state object is created for disabled jobs.
Because Hazelcast partition owners deserialize map values and entry
processors, the first release requires a homogeneous engine cluster:
1. Upgrade every member to binaries that contain the new serializers and
tracker while keeping the option disabled.
2. Confirm no old member remains.
3. Enable the option only for newly submitted jobs.
Before downgrading to a binary that does not understand the new types, stop
creating enabled jobs and allow existing enabled jobs to finish and clean their
running dirty state.
## Performance considerations
Existing member-removal handling already scans running jobs. This proposal
keeps the same big-O scan but adds one lightweight watermark update per
dirty-enabled job and one bounded state update for each affected job.
The implementation must measure member-removal processing latency with at
least 100 and 1,000 concurrent jobs. The state retained per job is bounded by
the configured history size and sampled task-group details.
## Implementation plan
### Phase 1: backend dirty-job tracking
1. Add configuration and bounded dirty-state models.
2. Add backup-aware atomic tracker operations and global member-event
sequencing.
3. Register member-loss incidents before existing task groups are failed.
4. Allocate stable attempt IDs during restore preparation.
5. Confirm attempts only when recovery reaches `DEPLOYING`.
6. Complete or retain attempt context according to deployment outcome.
7. Reconcile active state before scheduling after an active-master change.
8. Persist finished history and expose additive REST fields.
9. Add unit and integration coverage for idempotency, failure paths, HA
recovery, and compatibility.
### Phase 2: trusted planned-leave exclusion
This phase is independent and requires a reviewed authenticated operations
control plane. It is not part of the first backend PR.
### Later optional work
- Prometheus metrics and alert transitions;
- Web UI badge, filtering, and episode detail;
- faster stale-worker suspicion as a separate failure-detection proposal.
## Acceptance criteria
- [ ] One member-loss episode affecting multiple pipelines counts one
first-generation job attempt.
- [ ] A second deployment attempt advances the job count exactly once.
- [ ] Duplicate `memberRemoved` delivery and duplicate attempt confirmation
are idempotent.
- [ ] Ordinary task failures without an active member-loss recovery context
are not counted.
- [ ] Restore preparation without reaching `DEPLOYING` is not counted.
- [ ] A completed episode cannot be consumed by a later ordinary business
failure.
- [ ] A Kubernetes Pod restart without trusted planned-leave intent is
classified as `UNCLASSIFIED` and counted after deployment starts.
- [ ] Active-master change before or during attempt confirmation neither
loses nor duplicates the count.
- [ ] Partition-owner change preserves state through a synchronous backup.
- [ ] Missing state, watermark lag, or tracker failure returns `UNKNOWN`,
not `CLEAN`.
- [ ] Reaching the threshold does not alter the primary job status or retry
behavior.
- [ ] Running and finished jobs expose the same summary semantics.
- [ ] The feature is disabled by default and old finished-job records remain
readable.
- [ ] Mixed-version clusters cannot enable the feature before every member
has upgraded.
- [ ] Member-removal processing remains bounded under the defined
concurrency test matrix.
## Usage Scenario
Operators running long-lived Zeta streaming jobs need to identify jobs that
repeatedly recover after member loss even when those jobs return to `RUNNING`.
The proposed contract answers:
- Has this job exceeded the acceptable number of member-loss deployment
attempts?
- How many job-level attempts and recovery episodes occurred?
- Which member UUID and address were involved?
- Is the result complete, or should it be treated as unknown?
- Can the same result be queried after the job finishes?
## Related issues
- #11735: task exception and failure history contract
- #11755: stale task completion across restore generations
- #10804: state-store abstraction for Hazelcast IMap
- #11472: finished metrics after worker failover
- #10461: retry mechanism correctness
## Are you willing to submit a PR?
- [x] Yes, I am willing to submit a Draft PR for Phase 1.
## Code of Conduct
- [x] I agree to follow the Apache Software Foundation Code of Conduct.
--
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]
