DanielLeens opened a new pull request, #13100:
URL: https://github.com/apache/seatunnel/pull/13100
Yes, contributing to SeaTunnel.
### Purpose of this pull request
Add an E2E regression guard for #12256, fixed by #12260 ("[Fix][Zeta] Clean
empty checkpoint barrier counters").
Before #12260, `SinkAggregatedCommitterTask#notifyCheckpointComplete`
removed a checkpoint's `checkpointBarrierCounter` entry only while iterating
`checkpointCommitInfoMap`. A checkpoint whose writers return no commit info
never gets a `checkpointCommitInfoMap` entry, so its barrier counter was never
removed: a long-running streaming job leaked one map entry per such checkpoint,
forever.
### Why existing tests miss it
-
`SinkAggregatedCommitterTaskTest#testCheckpointBarrierCountersAreCleanedWithoutCommitInfo`
covers the fix only on a directly-instantiated task with hand-seeded maps.
-
`SinkAggregatedCommitterTaskCheckpointCacheLeakIT#testCommitInfoCacheStaysBoundedAcrossManyCheckpoints`
runs a real job, but its LocalFile writers always return `Optional.of(...)`
from `prepareCommit`, so every checkpoint has a `checkpointCommitInfoMap` entry
and the old removal path still clears the counter.
### What the test does
- Adds a small test-only sink, `EmptyCommitInfoAggregatedTest`
(`emptycommit/EmptyCommitInfoSink` + `@AutoService` factory, the same pattern
as `workerrestart/FinalCheckpointCloseHoldSink`). It provides an aggregated
committer, so Zeta deploys a `SinkAggregatedCommitterTask`, but its writers
always return `Optional.empty()`. The committer counts `commit()` calls (one
per completed checkpoint) in a static counter, which works because the test
cluster runs in the test JVM.
- I added a dedicated sink instead of a flag on the shared `InMemorySink` in
`seatunnel-e2e-common`. That way no other E2E test is affected.
- Adds one method,
`testCheckpointBarrierCounterStaysBoundedAcrossEmptyCheckpoints`, to the
existing IT. It runs a streaming FakeSource job with 1s checkpoints and waits
until 30 checkpoints complete (180s budget). Throughout, it reads the live
committer task's maps by reflection, the same way the existing method does. It
asserts that `checkpointBarrierCounter.size() <= MAX_BOUNDED_CACHE_SIZE` (3,
the existing bound) and that `commitInfoCache` and `checkpointCommitInfoMap`
stay empty, which guards the test's premise.
- The existing test method and its assertions are unchanged.
### How it fails on pre-fix code
With #12260 reverted, each completed empty checkpoint leaves one counter
entry behind. After about 4 checkpoints the in-loop bound assertion fails, and
by 30 checkpoints the map would hold about 30 entries against a bound of 3.
### Does this PR introduce any user-facing change?
No. The change is test-only.
### How was this patch tested?
Verification: spotless only locally; GitHub CI is the verification.
### Check list
* [x] Test-only change under `seatunnel-e2e/**`; no production code, docs or
dependencies changed.
🤖 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]
