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)

sjvanrossum · 2025-01-07T23:59:59Z

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

@@ -40,6 +42,7 @@ public class KafkaSinkMetrics {

  // Base Metric names
  private static final String RPC_LATENCY = "RpcLatency";
+  private static final String ESTIAMTED_BACKLOG_SIZE = "EstimatedBacklogSize";


Nit: ESTIAMTED_BACKLOG_SIZE -> ESTIMATED_BACKLOG_SIZE

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

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

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

    for (PartitionState<K, V> p : partitionStates) {
      long pBacklog = p.approxBacklogInBytes();
      if (pBacklog == UnboundedReader.BACKLOG_UNKNOWN) {
        return UnboundedReader.BACKLOG_UNKNOWN;
      }
-      backlogBytes += pBacklog;
    }

    return backlogBytes;


This method should remain as it is, right? getSplitBacklogBytes and may affect autoscaling behavior according to the docs and this change would have this override return either 0 or UnboundedReader.BACKLOG_UNKNOWN instead of the split's approximate backlog bytes across assigned partitions or UnboundedReader.BACKLOG_UNKNOWN.

yes I agree, this was deleted accidentally. Reverted that.

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).

Naireen · 2025-01-09T22:57:27Z

Flink test is failing, likely due to #33510,

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 added 2 commits December 23, 2024 21:41

add counter stuff

bc8996c

Address John's comments about separting conversion and validation checks

e9e2b16

Naireen mentioned this pull request Jan 6, 2025

Kafka add counters v1 uw2 #33503

Open

3 tasks

sjvanrossum suggested changes Jan 8, 2025

View reviewed changes

Naireen force-pushed the kafka_add_counters_V1 branch 5 times, most recently from 042a247 to 9e11bb6 Compare January 9, 2025 21:22

address Steven's comments

8f9a278

Naireen force-pushed the kafka_add_counters_V1 branch from 9e11bb6 to 8f9a278 Compare January 9, 2025 21:55

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 7, 2025

Naireen Jan 9, 2025

sjvanrossum Jan 8, 2025

Naireen Jan 9, 2025

sjvanrossum Jan 8, 2025

Naireen Jan 9, 2025

Naireen commented Jan 9, 2025

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

Naireen commented Jan 9, 2025