Kafka add counters v1 uw2 #33503

Naireen · 2025-01-06T17:40:08Z

Add per worker metadata on metrics propaged over FnApi. This is on top of #33408

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

Naireen · 2025-01-06T23:18:33Z

Run Java PreCommit

Naireen · 2025-01-06T23:18:56Z

Run Java_IOs_Direct PreCommit

Naireen · 2025-01-06T23:19:12Z

Run Java_Pulsar_IO_Direct PreCommit

Naireen · 2025-01-06T23:19:32Z

Run Python_Coverage PreCommit

Naireen · 2025-01-07T00:26:26Z

Run Java PreCommit

Naireen · 2025-01-07T17:13:08Z

Run Java PreCommit

Naireen · 2025-01-07T18:36:56Z

Run Java PreCommit

Naireen · 2025-01-07T19:55:45Z

tests failing due to #28957

Naireen · 2025-01-07T19:58:28Z

Run Java PreCommit

github-actions · 2025-01-07T20:35:17Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

Naireen · 2025-01-07T20:48:52Z

assign set of reviewers @johnjcasey @sjvanrossum

flakey tests - want to get the review started in the meanwhile, but will wait on the tests to pass before I merge.

github-actions · 2025-01-07T20:50:06Z

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @kennknowles for label java.
R: @damccorm for label build.
R: @damondouglas for label io.
R: @sjvanrossum for label kafka.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

Naireen · 2025-01-07T20:50:33Z

Run Java PreCommit

Naireen · 2025-01-07T22:09:16Z

Run Java PreCommit

Naireen · 2025-01-07T22:51:12Z

Run Java PreCommit

damccorm · 2025-01-13T21:51:12Z

Would you mind adding some more context on what this change is doing?

This is on top of #33408

2 questions:

Does this mean that this is blocked on that PR or just related?
Is it better to have the same person/people review both PRs?

Naireen · 2025-03-04T21:34:05Z

Run Java PreCommit

Naireen · 2025-03-04T21:34:23Z

Run Java_GCP_IO_Direct PreCommit

scwhittle · 2025-03-05T14:48:32Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java

+     *     'supportsMetricsDeletion' is enabled.
+     */
+    @Override
+    public void recordBacklogBytes(String topicName, int partitionId, long backlogBytes) {


It's confusing how
KafkaSinkMetrics.createBacklogGauge(MetricName.named("KafkaSink", backlogs.getKey()));
and
Metrics.gauge(KafkaSinkMetrics.getMetricGaugeName(topicName, partitionId));
are different from just looking here.

From looking into the code, one sets perWorkerMetric when creating the DelegatingMetric. This is confusing because it is still a per-worker metric we are just reporting here, it is just in the MetricName.

Could we change things to be consistent? For example, what if KafkaMetrics always set the per-worker label and then the dataflow metric container looks for per-worker metric label and handles them differently if it wants to instead of the separate perWorkerGauge method?

It seems complicated that this kafkasinkmetrics has to know about the two possible reporting paths and which one works with different runners. For example, what if the non-sdf kakfa impl is run on v2 via the SDF wrapper? Or what if we supported SDF impl on v1?

scwhittle

Thanks! I like how it is keeping the interfaces simpler.

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingStepMetricsContainer.java

scwhittle · 2025-03-07T09:26:50Z

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingStepMetricsContainer.java

@@ -388,7 +372,8 @@ private void deleteStaleCounters(
  @VisibleForTesting
  Iterable<PerStepNamespaceMetrics> extractPerWorkerMetricUpdates() {
    ConcurrentHashMap<MetricName, Long> counters = new ConcurrentHashMap<MetricName, Long>();
-    ConcurrentHashMap<MetricName, Long> gauges = new ConcurrentHashMap<MetricName, Long>();
+    ConcurrentHashMap<MetricName, Long> per_worker_gauges =


nit: I don't think this (or counters) needs to be Concurrent. How about using ImmutableMapBuilder and then build when passing to convert?

I agree, that should be done for all of them, will do that in the followup clean up CL.

scwhittle · 2025-03-07T09:35:10Z

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingStepMetricsContainer.java

-          gauges.put(k, val);
-          v.reset();
+          // Check if metric name has the per worker label set
+          if (k.getLabels().containsKey("PER_WORKER_METRIC")


I think this could be optimized by keeping the perWorkerGauges and populating it by doing this labels check when the gauge is created.

Could add a comment about possible optimization and do it when fixing the other metric types.

Oh that makes sense, so we don't have to check each time we process an update.

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingStepMetricsContainer.java

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaSinkMetrics.java

...rs/core-java/src/main/java/org/apache/beam/runners/core/metrics/MonitoringInfoConstants.java

scwhittle · 2025-03-10T11:20:39Z

Run Java PreCommit

scwhittle · 2025-03-10T17:55:36Z

Run Java PreCommit

scwhittle · 2025-03-10T17:55:50Z

Run Java_PVR_Prism_Loopback PreCommit

scwhittle · 2025-03-10T17:56:00Z

Run Python_Dataframes PreCommit

Naireen · 2025-03-10T19:36:11Z

Run Python_Dataframes PreCommit 3.10

Naireen · 2025-03-10T19:36:21Z

Run Java PreCommit

Naireen · 2025-03-11T00:12:42Z

Run Java PreCommit

Naireen · 2025-03-11T01:08:03Z

Run Java PreCommit

Naireen · 2025-03-11T01:31:48Z

Run Java PreCommit

github-actions bot added java build model io runners dataflow kafka core labels Jan 6, 2025

Naireen force-pushed the kafka_add_counters_V1_uw2 branch 3 times, most recently from 5b28665 to 5f28d22 Compare January 6, 2025 22:17

Naireen marked this pull request as ready for review January 7, 2025 19:58

github-actions bot added the Next Action: Reviewers label Jan 7, 2025

Naireen marked this pull request as draft January 14, 2025 18:28

scwhittle requested changes Mar 5, 2025

View reviewed changes

Naireen added 4 commits March 6, 2025 01:59

add counter stuff

4c39d3a

Add metadata on Kafka sink metrics

6c42cf4

address comments

301aa6a

address comments on method differences

8620b36

Naireen force-pushed the kafka_add_counters_V1_uw2 branch from eb4432e to 8620b36 Compare March 6, 2025 01:59

reduce bloat in metrics api

1c582f5

Naireen mentioned this pull request Mar 6, 2025

[Feature Request]: Unify metrics that are to be aggregated on google worker and on the portable runner #34195

Open

17 tasks

github-actions bot added the dataflow label Mar 6, 2025

Naireen force-pushed the kafka_add_counters_V1_uw2 branch from 043392b to bdc7b6f Compare March 6, 2025 22:15

refactor to remove per worker gauge method

8e7785b

Naireen force-pushed the kafka_add_counters_V1_uw2 branch from bdc7b6f to 8e7785b Compare March 6, 2025 22:40

scwhittle requested changes Mar 7, 2025

View reviewed changes

Naireen force-pushed the kafka_add_counters_V1_uw2 branch from 727229b to 08a259f Compare March 10, 2025 09:53

scwhittle approved these changes Mar 10, 2025

View reviewed changes

...rs/core-java/src/main/java/org/apache/beam/runners/core/metrics/MonitoringInfoConstants.java Outdated Show resolved Hide resolved

Address nits

8e89e97

Naireen force-pushed the kafka_add_counters_V1_uw2 branch from 08a259f to 8e89e97 Compare March 10, 2025 10:04

Naireen mentioned this pull request Mar 11, 2025

Refactor per worker histogram metrics #34244

Open

3 tasks

scwhittle merged commit 53baae8 into apache:master Mar 11, 2025
116 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Kafka add counters v1 uw2 #33503

Kafka add counters v1 uw2 #33503

Naireen commented Jan 6, 2025

Naireen commented Jan 6, 2025

Naireen commented Jan 6, 2025

Naireen commented Jan 6, 2025

Naireen commented Jan 6, 2025

Naireen commented Jan 7, 2025

Naireen commented Jan 7, 2025

Naireen commented Jan 7, 2025

Naireen commented Jan 7, 2025

Naireen commented Jan 7, 2025

github-actions bot commented Jan 7, 2025

Naireen commented Jan 7, 2025

github-actions bot commented Jan 7, 2025

Naireen commented Jan 7, 2025

Naireen commented Jan 7, 2025

Naireen commented Jan 7, 2025

damccorm commented Jan 13, 2025

Naireen commented Mar 4, 2025

Naireen commented Mar 4, 2025

scwhittle Mar 5, 2025

scwhittle left a comment

scwhittle Mar 7, 2025

Naireen Mar 10, 2025

scwhittle Mar 7, 2025

Naireen Mar 10, 2025

scwhittle commented Mar 10, 2025

scwhittle commented Mar 10, 2025

scwhittle commented Mar 10, 2025

scwhittle commented Mar 10, 2025

Naireen commented Mar 10, 2025

Naireen commented Mar 10, 2025

Naireen commented Mar 11, 2025

Naireen commented Mar 11, 2025

Naireen commented Mar 11, 2025

Kafka add counters v1 uw2 #33503

Kafka add counters v1 uw2 #33503

Conversation

Naireen commented Jan 6, 2025

Add per worker metadata on metrics propaged over FnApi. This is on top of #33408

GitHub Actions Tests Status (on master branch)

Naireen commented Jan 6, 2025

Naireen commented Jan 6, 2025

Naireen commented Jan 6, 2025

Naireen commented Jan 6, 2025

Naireen commented Jan 7, 2025

Naireen commented Jan 7, 2025

Naireen commented Jan 7, 2025

Naireen commented Jan 7, 2025

Naireen commented Jan 7, 2025

github-actions bot commented Jan 7, 2025

Naireen commented Jan 7, 2025

github-actions bot commented Jan 7, 2025

Naireen commented Jan 7, 2025

Naireen commented Jan 7, 2025

Naireen commented Jan 7, 2025

damccorm commented Jan 13, 2025

Naireen commented Mar 4, 2025

Naireen commented Mar 4, 2025

scwhittle Mar 5, 2025

Choose a reason for hiding this comment

scwhittle left a comment

Choose a reason for hiding this comment

scwhittle Mar 7, 2025

Choose a reason for hiding this comment

Naireen Mar 10, 2025

Choose a reason for hiding this comment

scwhittle Mar 7, 2025

Choose a reason for hiding this comment

Naireen Mar 10, 2025

Choose a reason for hiding this comment

scwhittle commented Mar 10, 2025

scwhittle commented Mar 10, 2025

scwhittle commented Mar 10, 2025

scwhittle commented Mar 10, 2025

Naireen commented Mar 10, 2025

Naireen commented Mar 10, 2025

Naireen commented Mar 11, 2025

Naireen commented Mar 11, 2025

Naireen commented Mar 11, 2025