Kafka add gauge v1 #33408

Naireen · 2024-12-17T23:00:23Z

Add per worker gauge support to add per backlog partition for kafka with java legacy worker for Dataflow

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.

codecov · 2024-12-18T00:02:39Z

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 58.95%. Comparing base (8da9bbe) to head (b7b7e2a).
Report is 190 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #33408      +/-   ##
============================================
+ Coverage     58.86%   58.95%   +0.08%     
- Complexity     3112     3183      +71     
============================================
  Files          1130     1133       +3     
  Lines        174419   174576     +157     
  Branches       3343     3366      +23     
============================================
+ Hits         102680   102916     +236     
+ Misses        68392    68312      -80     
- Partials       3347     3348       +1

Flag	Coverage Δ
java	`70.33% <ø> (+0.43%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

Naireen · 2024-12-18T04:18:38Z

R: @sjvanrossum for the kafka io part, thanks in advance!

github-actions · 2024-12-18T04:19:55Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

Naireen · 2024-12-18T16:50:20Z

Run Java PreCommit

Naireen · 2024-12-18T17:56:46Z

R: @johnjcasey for the sdk portion of it.

.../java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverter.java

Naireen · 2024-12-19T19:48:13Z

Run Java_Pulsar_IO_Direct PreCommit

Naireen · 2024-12-19T20:57:30Z

Run Java PreCommit

Naireen · 2024-12-19T20:57:43Z

Run Java_GCP_IO_Direct PreCommit

Naireen · 2024-12-19T21:32:27Z

Run Java PreCommit

Naireen · 2024-12-19T21:53:22Z

Run Java PreCommit

Naireen · 2024-12-19T21:53:31Z

Run Java_GCP_IO_Direct PreCommit

Naireen · 2024-12-19T21:57:51Z

Run Java_GCP_IO_Direct PreCommit

Naireen · 2024-12-19T22:39:42Z

Run Java PreCommit

sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/NoOpGauge.java

...a/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverterTest.java

sjvanrossum · 2025-01-07T23:41:54Z

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

+    /**
+     * @param topicName topicName
+     * @param partitionId partitionId for the topic Only included in the metric key if
+     *     'supportsMetricsDeletion' is enabled.
+     * @param backlog backlog for the topic Only included in the metric key if
+     *     'supportsMetricsDeletion' is enabled.
+     */


" Only" -> ". Only"?

Removed, thanks for catching that.

sjvanrossum · 2025-01-07T23:49:51Z

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

@@ -71,11 +79,17 @@ abstract class KafkaMetricsImpl implements KafkaMetrics {

    abstract HashMap<String, ConcurrentLinkedQueue<Duration>> perTopicRpcLatencies();

+    static ConcurrentHashMap<String, Gauge> backlogGauges = new ConcurrentHashMap<String, Gauge>();
+
+    abstract HashMap<String, Long> perTopicPartitionBacklogs();


If an instance of this class may be concurrently updated, then HashMap needs to be replaced (ditto for the existing HashMap fields). Use ConcurrentHashMap instead.

Slightly unrelated, but why doesn't perTopicRpcLatencies use a gauge or sum as the value type?

What would the sum represent? the sum of latencies? but each individual one is important, and a sum would lose information.
A gauge isn't quite clear either, if you have two concurrent rpcs that completed, what value do you return?

A histogram of values provides more information (and allows us to see the spread of values)

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

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

sjvanrossum · 2025-01-08T00:11:22Z

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

@@ -743,6 +747,16 @@ private void reportBacklog() {
    backlogElementsOfSplit.set(splitBacklogMessages);
  }

+  private void reportBacklogMetrics() {


Looks like this can be merged with reportBacklog (potentially rename that method to reportBacklogMetrics updateBacklogMetrics).

No, I explicitly moved it out to be separate, since reportBacklog() is called twice, and we only need to do this once (when we advance to the next record).

...a/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverterTest.java

sjvanrossum · 2025-01-17T14:00:16Z

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

+    static ConcurrentHashMap<String, Histogram> latencyHistograms =
+        new ConcurrentHashMap<String, Histogram>();


Suggested change

static ConcurrentHashMap<String, Histogram> latencyHistograms =

new ConcurrentHashMap<String, Histogram>();

private static final Map<String, Histogram> LATENCY_HISTOGRAMS =

new ConcurrentHashMap<>();

sjvanrossum · 2025-01-17T14:01:11Z

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


-    abstract HashMap<String, ConcurrentLinkedQueue<Duration>> perTopicRpcLatencies();
+    static ConcurrentHashMap<String, Gauge> backlogGauges = new ConcurrentHashMap<String, Gauge>();


Suggested change

static ConcurrentHashMap<String, Gauge> backlogGauges = new ConcurrentHashMap<String, Gauge>();

private static final Map<String, Gauge> BACKLOG_GAUGES = new ConcurrentHashMap<>();

Hmmm... I searched for backlogGauges, but I'm not finding any uses of this it.
Is this still needed?

sjvanrossum · 2025-01-17T15:11:25Z

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

@@ -86,13 +100,26 @@ public void updateSuccessfulRpcMetrics(String topic, Duration elapsedTime) {
        if (latencies == null) {
          latencies = new ConcurrentLinkedQueue<Duration>();
          latencies.add(elapsedTime);
-          perTopicRpcLatencies().put(topic, latencies);
+          perTopicRpcLatencies().putIfAbsent(topic, latencies);


If I'm not mistaken, L99-106 can be replaced with perTopicRpcLatencies().computeIfAbsent(topic, ConcurrentLinkedQueue::new).add(elapsedTime); which reduces locking operations. It also changes what happens in L100-104, which currently drops elapsedTime if the call at 103 observes that a write for that key had happened in between.

It won't accept
perTopicRpcLatencies().computeIfAbsent(topic, ConcurrentLinkedQueue::new).add(elapsedTime);

saying that it's unable to infer the type of the argument. I tried a couple of different variations, and couldn't get it to work with a lambda. It think the issue is that it takes in a string, and we want a ConcurrentQueue of Duration

This is what the compiler wants, and I don't know how to pass in a lambda matching that signature
String,Function<? super String,? extends ConcurrentLinkedQueue<Duration>> (my java knowledge is a limited.)

sjvanrossum · 2025-01-17T15:15:13Z

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

@@ -743,6 +747,16 @@ private void reportBacklog() {
    backlogElementsOfSplit.set(splitBacklogMessages);
  }

+  private void reportBacklogMetrics() {


Naireen · 2025-01-17T19:10:43Z

Run Java_IOs_Direct PreCommit

github-actions bot added java build io runners dataflow kafka labels Dec 17, 2024

Naireen force-pushed the kafka_add_counters_V1 branch 4 times, most recently from 2791b23 to b91a77f Compare December 18, 2024 01:21

Naireen marked this pull request as ready for review December 18, 2024 04:18

Naireen force-pushed the kafka_add_counters_V1 branch from b91a77f to 948dfe6 Compare December 18, 2024 07:34

johnjcasey requested changes Dec 19, 2024

View reviewed changes

.../java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverter.java Outdated Show resolved Hide resolved

Naireen force-pushed the kafka_add_counters_V1 branch from 11bee27 to b9d2f2b Compare December 19, 2024 19:54

johnjcasey approved these changes Dec 19, 2024

View reviewed changes

Naireen mentioned this pull request Jan 6, 2025

Kafka add counters v1 uw2 #33503

Draft

3 tasks

sjvanrossum suggested changes Jan 8, 2025

View reviewed changes

Naireen force-pushed the kafka_add_counters_V1 branch 3 times, most recently from ae2d885 to 0972ff3 Compare January 13, 2025 17:37

sjvanrossum suggested changes Jan 17, 2025

View reviewed changes

github-actions bot added python go website docker extensions learning gcp jdbc yaml prism labels Jan 17, 2025

Naireen added 4 commits January 17, 2025 18:23

add counter stuff

b0a5147

Address John's comments about separting conversion and validation checks

e5cad4e

address Steven's comments

f02a194

another round of comments

925290b

Naireen force-pushed the kafka_add_counters_V1 branch from f33b235 to 925290b Compare January 17, 2025 18:23

github-actions bot removed python go website docker extensions learning gcp jdbc yaml prism labels Jan 17, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Kafka add gauge v1 #33408

Kafka add gauge v1 #33408

Naireen commented Dec 17, 2024

codecov bot commented Dec 18, 2024

Naireen commented Dec 18, 2024

github-actions bot commented Dec 18, 2024

Naireen commented Dec 18, 2024

Naireen commented Dec 18, 2024

Naireen commented Dec 19, 2024

Naireen commented Dec 19, 2024

Naireen commented Dec 19, 2024

Naireen commented Dec 19, 2024

Naireen commented Dec 19, 2024

Naireen commented Dec 19, 2024

Naireen commented Dec 19, 2024

Naireen commented Dec 19, 2024

sjvanrossum Jan 7, 2025

Naireen Jan 9, 2025

sjvanrossum Jan 7, 2025

sjvanrossum Jan 8, 2025

Naireen Jan 9, 2025

sjvanrossum Jan 8, 2025

Naireen Jan 9, 2025

sjvanrossum Jan 17, 2025

sjvanrossum Jan 17, 2025

sjvanrossum Jan 17, 2025

sjvanrossum Jan 17, 2025

Naireen Jan 17, 2025

sjvanrossum Jan 17, 2025

Naireen Jan 17, 2025

sjvanrossum Jan 17, 2025

Naireen commented Jan 17, 2025

		static ConcurrentHashMap<String, Histogram> latencyHistograms =
		new ConcurrentHashMap<String, Histogram>();


		abstract HashMap<String, ConcurrentLinkedQueue<Duration>> perTopicRpcLatencies();
		static ConcurrentHashMap<String, Gauge> backlogGauges = new ConcurrentHashMap<String, Gauge>();

	static ConcurrentHashMap<String, Gauge> backlogGauges = new ConcurrentHashMap<String, Gauge>();
	private static final Map<String, Gauge> BACKLOG_GAUGES = new ConcurrentHashMap<>();

Kafka add gauge v1 #33408

Are you sure you want to change the base?

Kafka add gauge v1 #33408

Conversation

Naireen commented Dec 17, 2024

Add per worker gauge support to add per backlog partition for kafka with java legacy worker for Dataflow

GitHub Actions Tests Status (on master branch)

codecov bot commented Dec 18, 2024

Codecov Report

Naireen commented Dec 18, 2024

github-actions bot commented Dec 18, 2024

Naireen commented Dec 18, 2024

Naireen commented Dec 18, 2024

Naireen commented Dec 19, 2024

Naireen commented Dec 19, 2024

Naireen commented Dec 19, 2024

Naireen commented Dec 19, 2024

Naireen commented Dec 19, 2024

Naireen commented Dec 19, 2024

Naireen commented Dec 19, 2024

Naireen commented Dec 19, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Naireen commented Jan 17, 2025