[Bug][Zeta] checkpoint max-retained is not enforced after coordinator recreation
- Dominant language
- Java
- Stars
- 9.7k
- Forks
- 2.4k
- Avg merge
- 3d 9h
- Merged PRs (30d)
- 204
Description
### Search before asking
- [x] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found no similar open issue.
I also checked closed issue #5046. That issue fixed checkpoint-file deletion in `HdfsStorage`. This report has a different failure mode: after active-master failover recreates `CheckpointCoordinator`, the new coordinator loses the IDs of checkpoints already present in storage, so the retention logic never selects those old files for deletion.
### What happened
In a two-member Zeta cluster using a shared NFS directory through the `localfile` checkpoint storage plugin, a streaming job was configured with `max-retained: 3`.
The job completed checkpoints normally. After two member failures that recreated the job's checkpoint coordinator, checkpointing continued and the job remained `RUNNING`, but the same job directory contained eight checkpoint files at the same time:
```text
...-1-1.ser
...-1-2.ser
...-1-3.ser
...-1-4.ser
...-1-5.ser
...-1-6.ser
...-1-7.ser
...-1-8.ser
```
The files were visible with identical hashes from both members, so this was not a local-mount visibility problem. Cancelling the job removed the whole job checkpoint directory, but that does not protect a long-running job from storage growth across repeated active-master failovers.
Expected behavior:
- `max-retained: 3` is enforced across coordinator recreation and active-master failover.
- At most three completed checkpoints for the pipeline remain in storage while the job is running.
- The latest checkpoint remains recoverable while older checkpoints are deleted safely.
Actual behavior:
- Each new `CheckpointCoordinator` starts with an empty `completedCheckpointIds` queue.
- Existing checkpoint IDs in storage are not restored into that queue.
- Cleanup only considers IDs completed by the current coordinator instance.
- Checkpoints written by previous coordinator instances remain invisible to the retention decision and accumulate.
The runtime reproduction was first observed on a downstream Zeta build that preserves this checkpoint implementation. I then checked current upstream `dev` commit `0301f7cd576f3d4918ef857bf2579ca6ae14ddf9`; the same deterministic lifecycle gap is still present:
1. `CheckpointCoordinator` initializes `completedCheckpointIds` as an empty `ArrayDeque`.
2. `completePendingCheckpoint` only appends the checkpoint completed by the current coordinator instance.
3. The deletion list is generated only from that in-memory queue.
4. The restore/constructor path does not repopulate the queue from `CheckpointStorage#getAllCheckpoints(jobId)`.
Relevant code:
- `CheckpointCoordinator` queue initialization: https://github.com/apache/seatunnel/blob/0301f7cd576f3d4918ef857bf2579ca6ae14ddf9/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java#L224-L226
- Retention/deletion logic: https://github.com/apache/seatunnel/blob/0301f7cd576f3d4918ef857bf2579ca6ae14ddf9/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java#L1326-L1354
- Existing storage enumeration API: https://github.com/apache/seatunnel/blob/0301f7cd576f3d4918ef857bf2579ca6ae14ddf9/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-api/src/main/java/org/apache/seatunnel/engine/checkpoint/storage/api/CheckpointStorage.java#L51-L54
This means every coordinator generation can leave another group of old checkpoints behind. Repeated failovers can therefore grow checkpoint storage without respecting the configured retention bound.
### SeaTunnel Version
Current `dev` source at `0301f7cd576f3d4918ef857bf2579ca6ae14ddf9` (`3.0.0-SNAPSHOT`).
Runtime evidence was collected from a downstream Zeta build with the same `CheckpointCoordinator` retention lifecycle.
### SeaTunnel Config
```conf
seatunnel:
engine:
backup-count: 1
queue-type: blockingqueue
job-schedule-strategy: WAIT
slot-service:
dynamic-slot: true
checkpoint:
interval: 30000
timeout: 60000
storage:
type: localfile
max-retained: 3
plugin-config:
namespace: /mnt/shared/seatunnel/checkpoint
```
Job configuration:
```conf
env {
job.mode = "STREAMING"
parallelism = 4
}
source {
FakeSource {
parallelism = 4
row.num = 500
split.num = 200
split.read-interval = 500
schema {
fields {
id = "bigint"
payload = "string"
event_time = "timestamp"
}
}
}
}
sink {
Console {
parallelism = 4
log.print.data = false
}
}
```
### Running Command
```shell
# Start two Zeta members that use the same shared checkpoint namespace.
bin/seatunnel-cluster.sh
# Submit the streaming job and wait for at least two completed checkpoints.
bin/seatunnel.sh --config checkpoint-retention.conf --async
# Stop the current active-master member and wait for the remaining member to
# restore the job and complete more checkpoints. Restart the stopped member,
# then repeat against the new active master.
# While the job is still RUNNING, count the retained files.
find /mnt/shared/seatunnel/checkpoint/ -type f -name '*.ser' -print | sort
```
### Error Exception
```log
No exception is emitted. The job stays RUNNING and checkpoints keep completing,
but the retained file count exceeds checkpoint.storage.max-retained after
CheckpointCoordinator recreation.
```
### Zeta or Flink or Spark Version
Zeta `3.0.0-SNAPSHOT` source verified at `0301f7cd576f3d4918ef857bf2579ca6ae14ddf9`.
### Java or Scala Version
Runtime reproduction: OpenJDK 8. The affected coordinator logic is Java-version independent.
### Screenshots
Not applicable. The retained-file timeline and source links above provide the reproducible evidence.
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.