DanielLeens opened a new pull request, #12144:
URL: https://github.com/apache/seatunnel/pull/12144
### Purpose of this pull request
Test-only. Adds `CheckpointStorageRetentionPruneTest` in
`seatunnel-engine-server`, which verifies that checkpoint retention pruning
physically removes the pruned checkpoint files from the backing store, not just
from the storage API's view of the checkpoint set.
**Background: apache/seatunnel#5046, fixed by #5054.** Both
`HdfsStorage#deleteCheckpoint` overloads built the delete target from the bare
file name instead of `<namespace>/<jobId>/<fileName>`. `FileSystem#delete`
found nothing and returned `false` without throwing, the surrounding `catch
(Exception)` kept the failure invisible, and `CheckpointCoordinator` kept
"pruning" its in-memory id queue while the checkpoint files accumulated on disk
on every retention cycle. For long-running streaming jobs that is slow-burn
disk and inode exhaustion.
**The gap.** Nothing in the tree verifies that pruned checkpoints are
actually gone from storage:
- `CheckpointStorageTest#testStreamJobWithCancel` runs a streaming job at
`checkpoint.interval = 1000` for about 10 s (already crossing the module's
`max-retained: 3` prune threshold) but only asserts
`getAllCheckpoints(jobId).size() == 0` after cancel; it never observes mid-run
pruning.
- `LocalFileStorageTest` and the `AbstractFileCheckPointTest` subclasses
only call `deleteCheckpoint(jobId)` in `@AfterAll` teardown, with no assertions.
- The `deleteCheckpoint(jobId, pipelineId, List<String> checkpointIdList)`
overload that `CheckpointCoordinator#completePendingCheckpoint` actually calls
for retention was not exercised anywhere, and no test ever inspected the
backing store directly.
**What the new test does.** It sits next to `CheckpointStorageTest` (same
package, same `AbstractSeaTunnelServerTest` base, same
`stream_fake_to_console_with_checkpoint.conf`) and therefore runs the real
`hdfs` storage plugin backed by `fs.defaultFS: file:///` from the module's
`seatunnel.yaml`, i.e. the exact implementation that carried #5046:
1. Overrides `loadSeaTunnelConfig()` to set `max-retained = 2`.
`JobMaster#createJobCheckpointConfig` copies the value into the per-job config,
so the first prune fires after four completed checkpoints, a few seconds into
the job.
2. Derives the on-disk directory from the running storage instance,
`AbstractCheckpointStorage#getStorageParentDirectory()` + `jobId`, which
resolves to `/tmp/seatunnel/checkpoint_snapshot/<jobId>` with the module's
config. The plugin consumes and removes the `namespace` key from the plugin
config map during `initStorage`, so the instance is the only reliable source of
that path.
3. Polls `getCheckpointsByJobIdAndPipelineId` every 200 ms and accumulates
every id the API ever reported. Once a previously listed id disappears while
the job is still `RUNNING`, retention has pruned it.
4. Load-bearing assertion: lists `<namespace>/<jobId>` directly with
`java.nio.file.Files#list`, parsing
`<timestamp>-<random>-<pipelineId>-<checkpointId>.ser` independently of the
storage code, and asserts that the pruned ids have no file on disk, the
retained ids do, the pruned ids are the oldest ones (a restore must not lose
progress), and the file count never exceeds `2 * max-retained`.
5. Cancels the job and keeps the sibling test's full-wipe expectation
(`getAllCheckpoints == 0`), additionally verified on disk.
All waits are bounded with Awaitility (120 s for the prune, 30 s for the
wipe), so a regression fails with a descriptive message instead of hanging.
**Why the on-disk listing and not just the API count.** A storage
implementation that tracks ids in memory, or whose delete silently fails as in
#5046, can still report the expected checkpoint set through its API while the
backing store keeps growing. The assertion was checked for teeth locally with a
temporary storage wrapper (not committed) that hides pruned ids from
`getCheckpointsByJobIdAndPipelineId` but turns both id-based `deleteCheckpoint`
overloads into no-ops: the committed test fails after the bound with
```
pruned checkpoint files still exist in
/tmp/seatunnel/checkpoint_snapshot/<jobId>, pruned=[1, 2, ..., 116], onDisk=[1,
2, ..., 118]
```
Against the literal #5046 behaviour (delete no-ops while the API lists the
same directory) the API view and the disk view coincide, so the prune is never
observed and the test fails on `retention has not pruned any previously listed
checkpoint yet` instead; that case is reasoned from the code rather than
executed.
Related: #12096 fixes retention after coordinator recreation and adds
plugin-level unit tests for the delete overloads. This PR covers the
steady-state prune path end to end through the server, touches no main code,
and has no file overlap with it.
### Does this PR introduce _any_ user-facing change?
No. Test-only change; no `src/main` or `pom.xml` modifications.
### How was this patch tested?
- `./mvnw test -pl seatunnel-engine/seatunnel-engine-server
-Dtest=CheckpointStorageRetentionPruneTest` on JDK 8: 4 of 4 consecutive runs
pass, test time 10.55 s / 10.41 s / 10.38 s / 10.49 s. In the traced run the
coordinator completed checkpoints 1 to 4 within about 2.5 s of the job entering
`RUNNING`, pruned ids 1 and 2 at checkpoint 4, the test observed the prune on
the next 200 ms poll, and the job directory was gone after cancel; no job
directory was left behind by any passing run.
- Mutation check described above: one run against the no-op-delete wrapper,
which failed as intended on the on-disk assertion after the 120 s bound.
- `./mvnw spotless:check -pl seatunnel-engine/seatunnel-engine-server`
passes.
### Check list
* [ ] If any new Jar binary package adding in your PR, please add License
Notice according
[New License
Guide](https://github.com/apache/seatunnel/blob/dev/docs/en/developer/new-license.md)
* [ ] If necessary, please update the documentation to describe the new
feature. https://github.com/apache/seatunnel/tree/dev/docs
* [ ] If necessary, please update `incompatible-changes.md` to describe the
incompatibility caused by this PR.
* [ ] If you are contributing the connector code, please check that the
following files are updated:
1. Update
[plugin-mapping.properties](https://github.com/apache/seatunnel/blob/dev/plugin-mapping.properties)
and add new connector information in it
2. Update the pom file of
[seatunnel-dist](https://github.com/apache/seatunnel/blob/dev/seatunnel-dist/pom.xml)
3. Add ci label in
[label-scope-conf](https://github.com/apache/seatunnel/blob/dev/.github/workflows/labeler/label-scope-conf.yml)
4. Add e2e testcase in
[seatunnel-e2e](https://github.com/apache/seatunnel/tree/dev/seatunnel-e2e/seatunnel-connector-v2-e2e/)
5. Update connector
[plugin_config](https://github.com/apache/seatunnel/blob/dev/config/plugin_config)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
