Add sdf kafka poll latencies#34275

Merged

liferoad merged 4 commits intoapache:masterfrom

Naireen:add_sdf_kafka_poll_latencies

Mar 20, 2025

Contributor

Naireen commented Mar 13, 2025

Adds sdf poll metrics to kafka for runner v2

Should be merged after #34244

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.

github-actions bot added java io runners dataflow gcp kafka core labels

Contributor Author

Naireen commented Mar 13, 2025

Contributor

github-actions bot commented Mar 13, 2025

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

scwhittle requested changes

View reviewed changes

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

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

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

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java Outdated Show resolved Hide resolved

Naireen force-pushed the add_sdf_kafka_poll_latencies branch from 6f9eb14 to 31585e7 Compare

March 13, 2025 16:59

github-actions bot removed dataflow gcp labels

Contributor Author

Naireen commented Mar 14, 2025

Run Java PreCommit

Naireen requested a review from scwhittle

March 14, 2025 08:01

scwhittle requested changes

View reviewed changes

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

-                      rawRecords = poll(consumer, kafkaSourceDescriptor.getTopicPartition());
+                      KafkaMetrics kafkaMetrics = KafkaSinkMetrics.kafkaMetrics();
+                      rawRecords = poll(consumer, kafkaSourceDescriptor.getTopicPartition(), kafkaMetrics);
+                      kafkaMetrics.flushBufferedMetrics();

Contributor

scwhittle Mar 14, 2025

what about not flushing here and just falling through?

would it be ok to just flush once outside the while loop with a try { } finally { kafkaMetrics.flushBufferedMetrics(); } block?

Contributor Author

Naireen Mar 14, 2025

I didn't do it because of the returns, but apparently the finally block will run and then return, so that will work. I have modified the code.

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java Outdated Show resolved Hide resolved

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java Show resolved Hide resolved

Contributor Author

Naireen commented Mar 14, 2025

Run Java PreCommit

Naireen requested a review from scwhittle

March 14, 2025 23:58

Contributor Author

Naireen commented Mar 14, 2025

Run Java PreCommit

scwhittle requested changes

View reviewed changes

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

-                          } else {
-                            Preconditions.checkStateNotNull(this.extractOutputTimestampFn);
-                            outputTimestamp = extractOutputTimestampFn.apply(kafkaRecord);
+                      KafkaMetrics kafkaMetrics = KafkaSinkMetrics.kafkaMetrics();

Contributor

scwhittle Mar 17, 2025

can we instead have the metrics lifetime longer? Seems like avoiding allocation/flush per pull will be better for performance.

KafkaMetrics kafkaMetrics = KafkaSinkMetrics.kafkaMetrics();
try {
while (true) {

}
} finally {
}

Contributor Author

Naireen Mar 20, 2025

So the way the container is implemented is that once its flushed, we dont reuse it. I'm less usre of the context behind this decision, I dont see why we cant change it, as I agree it would make it more efficient, I'll leave that for a subsequent PR.

Contributor

scwhittle Mar 20, 2025

I still think it would only be flushed once. it would just live the whole time and be flushed at the end once.

Contributor Author

Naireen Mar 20, 2025

Done.

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java Outdated Show resolved Hide resolved

Naireen force-pushed the add_sdf_kafka_poll_latencies branch from 5ad81c7 to 4ac4b06 Compare

March 20, 2025 18:29

Naireen requested a review from scwhittle

March 20, 2025 19:06

liferoad added this to the 2.64.0 Release milestone

Naireen force-pushed the add_sdf_kafka_poll_latencies branch from 4ac4b06 to a900a43 Compare

March 20, 2025 19:15

Naireen added 3 commits

March 20, 2025 19:32


          add kafka sdf metrics

18139bf


          address comments

2f70906


          address more comments

552bf7f

Naireen force-pushed the add_sdf_kafka_poll_latencies branch from a900a43 to ad9471a Compare

March 20, 2025 19:32

scwhittle requested changes

View reviewed changes

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

-                          } else {
-                            Preconditions.checkStateNotNull(this.extractOutputTimestampFn);
-                            outputTimestamp = extractOutputTimestampFn.apply(kafkaRecord);
+                      KafkaMetrics kafkaMetrics = KafkaSinkMetrics.kafkaMetrics();

Contributor

scwhittle Mar 20, 2025

I still think it would only be flushed once. it would just live the whole time and be flushed at the end once.


          address comments

e72dc1e

Naireen force-pushed the add_sdf_kafka_poll_latencies branch from ad9471a to e72dc1e Compare

March 20, 2025 20:00

Naireen requested a review from scwhittle

March 20, 2025 20:31

scwhittle approved these changes

View reviewed changes

Contributor

scwhittle left a comment

Thanks!

Contributor Author

Naireen commented Mar 20, 2025

Run Java PreCommit)

Contributor Author

Naireen commented Mar 20, 2025

Run Java PreCommit

Contributor

liferoad commented Mar 20, 2025

https://github.com/apache/beam/actions/runs/13978267141/job/39140793274?pr=34275 is not related to this PR.

Contributor

liferoad commented Mar 20, 2025 •

edited

Loading

testAtLeastOnceWithAutoSchemaUpdate[1] (org.apache.beam.sdk.io.gcp.bigquery.StorageApiSinkSchemaUpdateIT) failed

sdks/java/io/google-cloud-platform/build/test-results/integrationTest/TEST-org.apache.beam.sdk.io.gcp.bigquery.StorageApiSinkSchemaUpdateIT.xml [took 5m 58s]

Contributor

liferoad commented Mar 20, 2025 •

edited

Loading

:runners:google-cloud-dataflow-java:worker:test FAILED |  
-- | --
  |   |  
  | org.apache.beam.runners.dataflow.worker.streaming.harness.FanOutStreamingEngineWorkerHarnessTest > testOnNewWorkerMetadata_redistributesBudget FAILED |  
  | java.lang.AssertionError at GrpcCleanupRule.java:201 |  
  |   |  
  | 1159 tests completed, 1 failed, 3 skipped

liferoad merged commit 4450031 into apache:master

22 of 23 checks passed

prodriguezdefino pushed a commit to prodriguezdefino/beam-pabs that referenced this pull request


          Add sdf kafka poll latencies (apache#34275)

fdd696f

* add kafka sdf metrics

* address comments

* address more comments

* address comments

---------

Co-authored-by: Naireen <naireenhussain@google.com>

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

core io java kafka runners