aakashsandhyan opened a new pull request, #29089:
URL: https://github.com/apache/flink/pull/29089
> **Draft.** The ASF JIRA issue is not filed yet, so the title carries a
`FLINK-XXXXX`
> placeholder — I will replace it in both the title and the commit message
before marking this
> ready for review. Opened as a draft rather than held back so the analysis
and the measurements
> are available to anyone hitting the same leak. Full `mvn clean verify` /
CI has not run yet
> either; what has been run is stated precisely under "Verifying this
change".
## What is the purpose of the change
Per-split source metric groups introduced by FLIP-513 / FLINK-37410 are
never detached from their
parent when a split finishes, so they accumulate for the lifetime of the job.
FLINK-37596 ("Close metric groups for finished splits") closes the per-split
`watermark` child on
completion. That part works — it deregisters the child's gauges from the
registry and from every
reporter. What it does not do is release the group *objects*:
`AbstractMetricGroup#close()` clears
a group's own children and metrics but never removes the group from its
**parent's** map.
For groups that live as long as the job that is harmless. For groups created
per transient entity
it leaks. Any source that mints a unique split id per split retains two
emptied metric groups,
their `QueryScopeInfo` and their scope strings per finished split, without
bound. A Paimon
streaming read guarantees unique ids — one per (snapshot × bucket).
Measured on a TaskManager with 1.4 days uptime:
```
GenericValueMetricGroup 301,792 key='split' (99.90% of leaked
groups)
GenericMetricGroup 301,709 name='watermark'
OperatorQueryScopeInfo 603,120 (2 per split)
HashMap$Node 7,964,229
=> 983 MB of a 1.67 GB live heap
```
Visible without a heap dump as `...split.<uuid>-<seq>.watermark.<metric>`
metric identifiers
growing linearly with uptime. In production the heap fills, GC saturates
(SerialGC measured at
418–1,340 ms/s), the source starves and consumer lag climbs. A restart
clears it completely and
it starts over.
This is **not** FLINK-35321, where the sink committer's
`pendingCommittables` gauge re-registers
under a stable name and is replaced by key — log spam, almost no heap. A job
with a `DISCARD`
sink and no committer at all still leaks these groups.
## Brief change log
- `AbstractMetricGroup#closeAndRemoveGroup(String)` — removes the named
sub-group from this
group's map while holding this group's monitor, then closes the child
*outside* that monitor so
the map mutation completes before any callout into the registry and
reporters.
- Returns early when this group is already closed: `close()` sets `closed =
true` before iterating
`groups.values()`, so a child detaching during a cascading teardown would
otherwise mutate the
map being iterated. `close()` clears the whole map anyway, making the
early return correct.
- Removal is keyed by name so it is O(1). Removing by value
(`groups.values().remove(child)`)
would be O(n) per finished split — quadratic across hundreds of thousands
of splits.
- `InternalSourceSplitMetricGroup` retains the per-split group in addition
to its watermark child.
- `InternalSourceSplitMetricGroup#onSplitFinished()` asks the enclosing
`split` KEY group to close
and remove the per-split group, rather than closing only the watermark
child. Closing the split
group still cascades into the watermark child, so the gauge deregistration
added by FLINK-37596
is preserved.
- `@VisibleForTesting AbstractMetricGroup#numberOfSubgroups()` so the
regression test can assert
on the parent's child count. Happy to drop this in favour of reflection if
reviewers prefer.
Metric names are unchanged: the KEY/VALUE group pair is still built by the
same
`addGroup(SPLIT, splitId)` call. Assembling it via `addGroup(SPLIT)` instead
would create a
GENERIC child and alter emitted names, which is why that call is left as-is.
Closing the split group *without* detaching it would not be sufficient — the
emptied group, its
`QueryScopeInfo` and its scope arrays would remain in the parent map,
leaving roughly half the
per-split footprint behind. A reporter-side filter such as
`metrics.reporter.<name>.filter.excludes` stops the reporter maps growing
but not the group tree,
so it addresses metric cardinality rather than this leak.
## Verifying this change
This change added tests and can be verified as follows:
- Added
`InternalSourceSplitMetricGroupTest#testSplitMetricGroupsAreDetachedWhenSplitsFinish`:
creates five per-split metric groups, finishes each, and asserts the
`split` group holds no
children. Without this change it fails with `expected: 0 but was: 5`.
Verified in both
directions — the test was run against an unpatched tree to confirm it
actually catches the
regression, not only that it passes with the fix.
- Manually verified on a running job: a Paimon streaming read with a
`DISCARD` sink (no sink
committer at all), on Flink 2.2.0 with this change backported, sustaining
~424,000 split
completions per hour. Three class histograms over 62 minutes, during which
roughly 440,000
splits finished:
| class | t0 | +14m | +62m | delta |
|---|---|---|---|---|
| `GenericValueMetricGroup` | 339 | 303 | 288 | **−51** |
| `GenericMetricGroup` | 256 | 220 | 205 | **−51** |
| `InternalSourceSplitMetricGroup` | 51 | 15 | **0** | **−51** |
| `QueryScopeInfo$OperatorQueryScopeInfo` | 215 | 143 | 113 | **−102** |
| `TimerGauge` | 332 | 172 | 96 | −236 |
| `HashMap$Node` | 126,633 | 124,665 | 105,501 | −21,132 |
Monotonically decreasing across all three captures, and the counts move in
exact proportion:
51 splits released means 51 value groups, 51 watermark groups, 51 wrappers
and 102
`QueryScopeInfo` (two per split). Roughly 440,000 splits finished during
that hour; unpatched,
the same interval would have added on the order of 880,000 group objects,
and these counts
could only ever grow. `InternalSourceSplitMetricGroup` reaching 0 is the
clearest single
signal — with no live splits, nothing per-split is retained at all.
Note `InternalSourceSplitMetricGroup` was never the leaking object —
`SourceOperator#splitFinished` already removes the wrapper from its own
map. It is included to
show the wrapper lifecycle is unchanged by this patch, and because its
count tracks live splits.
**Build status, stated precisely:** `mvn clean verify` on the whole project
has *not* been run.
What was run, using the repository's Maven wrapper so the pinned Maven
version is honoured:
`./mvnw -pl flink-runtime -am test
-Dtest=InternalSourceSplitMetricGroupTest` — BUILD SUCCESS,
3 tests, 0 failures, with `checkstyle` reporting 0 violations and
`spotless:check` passing.
Full CI verification is still outstanding.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): **no**
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: **no** — both
`AbstractMetricGroup` and `InternalSourceSplitMetricGroup` are
`@Internal`
- The serializers: **no**
- The runtime per-record code paths (performance sensitive): **no** —
`onSplitFinished()` is
called once per split, from `SourceOperator#splitFinished`, never per
record
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing,
Kubernetes/Yarn, ZooKeeper: **no**
- The S3 file system connector: **no**
## Documentation
- Does this pull request introduce a new feature? **no** — it is a
memory-leak fix
- If yes, how is the feature documented? **not applicable**
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Claude Opus 5)
--
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]
