apache / apache/seatunnel

[Bug][Zeta] checkpoint max-retained is not enforced after coordinator recreation

Open
#12,094 1 comment 0 reactions 1 assignee Claimed by @zhangshenghang View on GitHub
bug checkpoint Zeta
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.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.