rmetzger commented on PR #29564:
URL: https://github.com/apache/flink/pull/29564#issuecomment-6492276047
> **Note:** This review was written by **Claude Opus 5.5** (an AI model),
posted at the request of @rmetzger.
Thanks for the PR. Summary: the issue is real, and the PR works as
described. But I think the fix should go into the `MetricStore` instead of
hiding split metrics from the REST API by default.
### The issue is valid
`MetricStore` only adds entries. `MetricFetcherImpl.queryMetrics` → `addAll`
→ `map.put`. It only removes whole TaskManagers, jobs, subtasks and old
attempts. Each split metric is stored 3 times (attempt, subtask and task store).
I reproduced it on a MiniCluster with a recent master build, which includes
FLINK-40549. I used a `FileSource` in continuous mode, 100 files, parallelism
2, and polled REST every 100 ms:
- After all splits finished, the TaskManager had 0 split metrics registered
(checked with a reporter).
- `/subtasks/metrics` still listed 700 split metric names (100 × 7). The
response grew from 3.5 KB to 54 KB and never shrinks. That is about 500 bytes
per finished split. `/vertices/:id/metrics` grows the same way.
- The autoscaler requests the full name listing for each source vertex on
every metric collection (`ScalingMetricCollector#queryAggregatedMetricNames`).
So the >100 MB responses you saw are expected for long-running jobs.
Small corrections to the Jira ticket:
- Kafka splits normally don't finish in unbounded mode, so Kafka is a weak
example. Better examples are `FileSource` in continuous mode, streaming reads
from Iceberg or Paimon, and CDC snapshot chunks.
- There are 7 metrics per split, and the names are `*TimeMsPerSecond`.
### The PR
I compiled the 4 changed classes against the same build and ran the same job:
- Default: no split metrics in REST, and reporters still receive them.
- Empty value: the old behavior comes back.
My concerns:
1. **Split metrics are also hidden while the split is active.** In my run, 2
splits were active on the TaskManager and REST showed none of their metrics.
This removes part of FLIP-513 from the UI by default, instead of cleaning up
finished splits. FLIP-513 says "No existing behavior will be changed". I think
the FLIP author (Efrat Levitan) and Piotr Nowojski should agree to this.
2. **New `@PublicEvolving` option plus a changed default.** This probably
needs a dev@ discussion. FLINK-33804 asked for a similar option, and a dev@
proposal was requested there. It also can't normally be backported to the
affected 2.1/2.2/2.3 patch releases.
3. **The root cause stays.** Any other metric group that is created and
closed at runtime still piles up (FLINK-28852). If the store is fixed later,
the option is no longer needed, but removing a public option takes a
deprecation cycle.
### Suggested alternative
Fix `MetricStore` so it drops metrics that are no longer reported:
- When a TaskManager's dump arrives, check each execution attempt that
appears in it. Remove the keys that the dump no longer contains from that
attempt's attempt, subtask and task stores.
- Don't touch attempts that are missing from the dump. Finished or cancelled
tasks then keep their last values, as they do today. That should avoid the
conflict with FLINK-28588 that FLINK-28852 was closed for, because that
proposal cleared the whole store.
- Skip this cleanup when `MetricQueryService#enforceSizeLimit` dropped a
whole metric type from the dump. Otherwise all gauges of that TaskManager would
disappear. This needs a flag in `MetricSerializationResult`.
This is a plain bug fix with no new option, so it can be backported. A
filter option for the query service could still be useful on its own, for
example for sources with thousands of active splits. But I'd treat that as a
separate feature with an empty default.
### Smaller points, if the current approach is kept
- The default `*.operator.split.watermark:*:*` must match the private group
names in `InternalSourceSplitMetricGroup`. The tests hard-code
`addGroup("split", "split-0").addGroup("watermark")`. Using
`InternalSourceSplitMetricGroup.wrap(...)` in the test would make a renamed
group fail the test.
- In filter patterns, `.` matches any character.
`DefaultReporterFilters#convertToPattern` does `replaceAll("\\.", "\\.")`,
which changes nothing. This bug already exists, and it is harmless for this
default.
- The docs note is in the `watermark.currentWatermark` row, but it applies
to all 7 split metrics.
- With an empty list, the log says "Metrics matching [] will not be exposed
through the metric query service".
- `queryMetricNames` polls with `Thread.sleep`.
`CommonTestUtils.waitUntilCondition` already exists for this.
--
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]
