tagadearpit opened a new issue, #12677:
URL: https://github.com/apache/seatunnel/issues/12677
## Search before asking
- [x] Searched existing issues and pull requests; no existing fix for this
reproduction was found.
## What happened
In Zeta checkpoint coordinator
(`seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java`),
`completePendingCheckpoint()` cleans up old checkpoints as follows:
```java
if (completedCheckpointIds.size()
% coordinatorConfig.getStorage().getMaxRetainedCheckpoints()
== 0
&& completedCheckpointIds.size()
/ coordinatorConfig.getStorage().getMaxRetainedCheckpoints()
> 1) {
List<String> needDeleteCheckpointId = new ArrayList<>();
for (int i = 0;
i < coordinatorConfig.getStorage().getMaxRetainedCheckpoints();
i++) {
needDeleteCheckpointId.add(completedCheckpointIds.removeFirst());
}
checkpointStorage.deleteCheckpoint(
String.valueOf(completedCheckpoint.getJobId()),
String.valueOf(completedCheckpoint.getPipelineId()),
needDeleteCheckpointId);
}
```
`CHECKPOINT_STORAGE_MAX_RETAINED` (`ServerConfigOptions.java:232`) is
documented as **"The maximum number of retained checkpoints"** (default `20`) —
i.e. the user expectation is that storage never holds more than that many at
once.
However, cleanup only fires when the in-memory `completedCheckpointIds`
deque size is an *exact multiple* of `max-retained` **and** that multiple is `>
1`. It then deletes a whole batch of `max-retained` entries at once, rather
than trimming down to the limit continuously.
For example, with `max-retained = 3`:
| checkpoints completed | deque size | checkpoints actually in storage |
|---|---|---|
| 1–3 | 1–3 | 1, 2, 3 (no cleanup — `3/3 = 1`, not `> 1`) |
| 4, 5 | 4, 5 | up to **5** retained — already over the configured limit of
3 |
| 6 | 6 → cleanup fires | deletes 1, 2, 3 → back down to 3 |
| 7, 8 | 7, 8 | up to **5** again |
| 9 | cleanup fires | back to 3 |
Instead of a hard cap of `max-retained`, storage oscillates between
`max-retained` and `2 * max-retained - 1`. With the default of 20, up to **39**
checkpoints can accumulate in storage before a single batch delete — nearly
double the configured limit.
## SeaTunnel Version
dev (current HEAD: 3ac7003fb90d35e5945f94834742247fabd3dec7)
## SeaTunnel Config
```yaml
seatunnel:
engine:
checkpoint:
storage:
max-retained: 3
```
## Running Command / reproduction
Minimal reproduction of the branch logic in isolation:
```java
Deque<String> completedCheckpointIds = new ArrayDeque<>();
int maxRetained = 3;
int peakSize = 0;
for (int checkpointId = 1; checkpointId <= 30; checkpointId++) {
completedCheckpointIds.addLast(String.valueOf(checkpointId));
peakSize = Math.max(peakSize, completedCheckpointIds.size());
if (completedCheckpointIds.size() % maxRetained == 0
&& completedCheckpointIds.size() / maxRetained > 1) {
for (int i = 0; i < maxRetained; i++) {
completedCheckpointIds.removeFirst();
}
}
}
System.out.println("configured max-retained = " + maxRetained);
System.out.println("peak retained observed = " + peakSize); // prints 5, not
3
```
Or integration-style: configure `checkpoint.storage.max-retained: 3`, run a
streaming job long enough to complete checkpoints 4 and 5, and list the
checkpoint storage directory between checkpoint 5 and 6 — it contains 5
checkpoints instead of the configured limit of 3.
## Suggested Fix
Drop the batch/modulo logic and trim down to `maxRetained` every time:
```java
int maxRetained = coordinatorConfig.getStorage().getMaxRetainedCheckpoints();
if (completedCheckpointIds.size() > maxRetained) {
List<String> needDeleteCheckpointId = new ArrayList<>();
while (completedCheckpointIds.size() > maxRetained) {
needDeleteCheckpointId.add(completedCheckpointIds.removeFirst());
}
checkpointStorage.deleteCheckpoint(
String.valueOf(completedCheckpoint.getJobId()),
String.valueOf(completedCheckpoint.getPipelineId()),
needDeleteCheckpointId);
}
```
This enforces the documented invariant ("at most `max-retained`
checkpoints") after every single checkpoint completion.
## Java or Scala Version
JDK 17
## Are you willing to submit PR?
- [x] Yes
## Code of Conduct
- [x] I agree to follow the Apache 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]
