Dennis-Mircea opened a new pull request, #28014:
URL: https://github.com/apache/flink/pull/28014
<!--
*Thank you very much for contributing to Apache Flink - we are happy that
you want to help us improve Flink. To help the community review your
contribution in the best possible way, please go through the checklist below,
which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a
hassle. In order to uphold a high standard of quality for code contributions,
while at the same time managing a large number of contributions, we need
contributors to prepare the contributions well, and give reviewers enough
contextual information for the review. Please also understand that
contributions that do not follow this guide will take longer to review and thus
typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA
issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are
made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the
pull request", where *FLINK-XXXX* should be replaced by the actual issue
number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following
this pattern: `[hotfix] [docs] Fix typo in event time introduction` or
`[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the
pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean
verify` passes. You can set up Azure Pipelines CI to do that following [this
guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from
multiple issues.
- Each commit in the pull request has a meaningful commit message
(including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and
this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
FLIP link:
https://docs.google.com/document/d/1TlyTc6fvYGG1xlO-IFlGBC97CxilUBFdZBpLOBPdKD8/edit?usp=sharing
Task-level `numRecordsOut` is a single scalar today, which hides how much
data a task routes to each of its downstream vertices when it has more than one
network output (side outputs, multi-sink, broadcast fan-out). Consumers that
need per-edge throughput, most notably the Kubernetes autoscaler, which
computes per-vertex target parallelism, cannot distinguish traffic per
downstream and therefore over- or under-provision affected vertices.
This PR exposes a per-downstream-target `numRecordsOut` breakdown
end-to-end: it is registered as an individual live metric
(`numRecordsOut.<targetJobVertexId>`) for reporters, included in the archived
`IOMetrics` snapshot, aggregated per job vertex, and surfaced on the REST
`/jobs/:jobid`, `/jobs/:jobid/vertices/:vertexid`, and subtask-attempt
responses as a new `write-records-per-target` field. The aggregate
`numRecordsOut`/`write-records` scalar is unchanged.
## Brief change log
- **Planner wiring:** `NonChainedOutput` now carries the downstream
`JobVertexID`, populated by `StreamingJobGraphGenerator#createOrReuseOutput`,
so every network output knows which vertex it feeds.
- **Metric registration:** `TaskIOMetricGroup` gains two overloads:
`reuseRecordsOutputCounter(Counter, jobVertexId)` (aggregate-contributing +
individual metric) and `registerNumRecordsOutPerTarget(Counter, jobVertexId)`
(target-only, does not contribute to the aggregate). Both emit the
`numRecordsOut.<targetJobVertexId>` metric consumable by
Prometheus/JMX/OTel/etc.
- **Operator wiring:** `OperatorChain#createOutputCollector` installs a
per-target counter on each `RecordWriterOutput`. The single-output path uses
the aggregate-contributing overload; the multi-output / broadcast fan-out path
uses the target-only overload and lets `BroadcastingOutputCollector` own the
aggregate (preventing double-count on broadcast). Missing target ids are logged
once and fall back to the aggregate-only counter; metric wiring never fails a
task.
- **Archival path fix:** `Execution#updateAccumulatorsAndMetrics` now
preserves the new `numRecordsOutPerTarget` map when rebuilding the lean
archived `IOMetrics` (previously it was silently dropped, which would have made
the breakdown invisible to REST consumers).
- **REST surface:**
- `IOMetrics` carries the new `Map<String, Long> numRecordsOutPerTarget`.
- `MutableIOMetrics` folds per-subtask maps into the vertex-level view via
per-key sum.
- `IOMetricsInfo` adds a serialized `write-records-per-target` field.
- `JobDetailsHandler`, `JobVertexTaskManagersHandler`, and
`SubtaskExecutionAttemptDetailsInfo` populate the new field.
- Regenerated `rest_v1_dispatcher.yml` and `rest_v1_dispatcher.html`.
## Verifying this change
This change added tests and can be verified as follows:
- **Unit:**
- `TaskIOMetricGroupTest`: covers aggregate-contributing vs target-only
registration semantics, snapshot propagation, and that the per-target map
reports live counter values.
- `StreamingJobGraphGeneratorTest`: asserts every `NonChainedOutput`
carries a real downstream `JobVertexID` for both normal pipelines and broadcast
fan-outs.
- `JobDetailsHandlerTest`: asserts per-subtask per-target maps are merged
per key into the vertex-level `IOMetricsInfo`, and verifies back-compat when
the map is empty.
- `SubtaskExecutionAttemptDetailsInfoTest`: Jackson round-trip for both
empty and populated per-target maps.
- **Integration (new):** `PerTargetNumRecordsOutITCase` in `flink-tests`
runs real jobs on a `MiniClusterExtension` and asserts the full pipeline
(`TaskIOMetricGroup` -> `IOMetrics` -> `MutableIOMetrics` -> `IOMetricsInfo` ->
JSON) via `RestClusterClient#getJobDetails`:
- **Side output topology** (`source -> process (split by i % 3) ->
{main-sink, side-sink}`): aggregate `write-records = 20`, per-target map
contains both downstream `JobVertexID`s with the expected split (13 main / 7
side), and per-target sum equals the aggregate.
- **Broadcast topology** (`source.broadcast() -> 3 x (map ->
DiscardingSink)`): aggregate stays at `NUM_RECORDS = 20` (no double-count),
each of the 3 per-target entries equals `NUM_RECORDS` independently, directly
validating the target-only registration semantics.
- **ArchUnit:** `flink-architecture-tests-production` and
`flink-architecture-tests-test` pass with no new violations; no
`archunit-violations/` store needed changes.
## 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
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- 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? yes
- If yes, how is the feature documented? JavaDocs
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR,
please
change the checkbox below to `[X]` followed by the name of the tool, and
uncomment the
"Generated-by" line. See the ASF Generative Tooling Guidance for details:
https://www.apache.org/legal/generative-tooling.html
You are responsible for the quality and correctness of every change in this
PR
regardless of the tooling used. Low-effort AI-generated PRs will be closed.
See
AGENTS.md for the full guidance.
-->
- [X] Yes
Generated-by: 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]
