You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "Naireen (via GitHub)" <gi...@apache.org> on 2024/12/17 23:00:26 UTC
[PR] Kafka add gauge v1 [beam]
Naireen opened a new pull request, #33408:
URL: https://github.com/apache/beam/pull/33408
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](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33408:
URL: https://github.com/apache/beam/pull/33408#discussion_r1893029140
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverter.java:
##########
@@ -86,6 +87,38 @@ private static Optional<MetricValue> convertCounterToMetricValue(
.setValueInt64(value));
}
+ /**
+ * @param metricName The {@link MetricName} that represents this counter.
+ * @param value The counter value.
+ * @return If the conversion succeeds, {@code MetricValue} that represents this counter. Otherwise
+ * returns an empty optional
+ */
+ private static Optional<MetricValue> convertGaugeToMetricValue(
+ MetricName metricName,
+ Long value,
+ Map<MetricName, LabeledMetricNameUtils.ParsedMetricName> parsedPerWorkerMetricsCache) {
+
+ if ((!metricName.getNamespace().equals(BigQuerySinkMetrics.METRICS_NAMESPACE)
Review Comment:
Correct, if its not in that namespace, just skip the conversion. Since we do that for all the metrics here, I moved the check out from the gauge and other metric converters.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2551947462
R: @johnjcasey for the sdk portion of it.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2609007901
Run Java PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2550298140
R: @sjvanrossum for the kafka io part, thanks in advance!
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33408:
URL: https://github.com/apache/beam/pull/33408#discussion_r1920529274
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -67,15 +74,22 @@ abstract class KafkaMetricsImpl implements KafkaMetrics {
private static final Logger LOG = LoggerFactory.getLogger(KafkaMetricsImpl.class);
- static HashMap<String, Histogram> latencyHistograms = new HashMap<String, Histogram>();
+ static ConcurrentHashMap<String, Histogram> latencyHistograms =
+ new ConcurrentHashMap<String, Histogram>();
+
+ abstract ConcurrentHashMap<String, ConcurrentLinkedQueue<Duration>> perTopicRpcLatencies();
- abstract HashMap<String, ConcurrentLinkedQueue<Duration>> perTopicRpcLatencies();
+ static ConcurrentHashMap<String, Gauge> backlogGauges = new ConcurrentHashMap<String, Gauge>();
Review Comment:
Removed.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2555843478
Run Java_GCP_IO_Direct PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "sjvanrossum (via GitHub)" <gi...@apache.org>.
sjvanrossum commented on code in PR #33408:
URL: https://github.com/apache/beam/pull/33408#discussion_r1906156715
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverterTest.java:
##########
@@ -345,4 +372,76 @@ public void testConvert_convertCountersAndHistograms() {
parsedMetricNames,
IsMapContaining.hasEntry(histogramMetricName, parsedHistogramMetricName));
}
+
+ @Test
+ public void testConvert_successfulyConvertGauges() {
+ String step = "testStepName";
+ Map<MetricName, LockFreeHistogram.Snapshot> emptyHistograms = new HashMap<>();
+ Map<MetricName, Long> counters = new HashMap<MetricName, Long>();
+ Map<MetricName, Long> gauges = new HashMap<MetricName, Long>();
+ Map<MetricName, LabeledMetricNameUtils.ParsedMetricName> parsedMetricNames = new HashMap<>();
+
+ MetricName KafkaMetric1 = MetricName.named("KafkaSink", "metric1");
+ MetricName KafkaMetric2 = MetricName.named("KafkaSink", "metric2*label1:val1;label2:val2;");
+ MetricName KafkaMetric3 = MetricName.named("KafkaSink", "metric3");
+
+ gauges.put(KafkaMetric1, 5L);
+ gauges.put(KafkaMetric2, 10L);
+ gauges.put(KafkaMetric3, 0L);
+
+ Collection<PerStepNamespaceMetrics> conversionResult =
+ MetricsToPerStepNamespaceMetricsConverter.convert(
+ step, counters, gauges, emptyHistograms, parsedMetricNames);
+
+ DataflowGaugeValue gauge_value1 = new DataflowGaugeValue();
+ gauge_value1.setValue(5L);
+
+ DataflowGaugeValue gauge_value2 = new DataflowGaugeValue();
+ gauge_value2.setValue(10L);
+
+ DataflowGaugeValue gauge_value3 = new DataflowGaugeValue();
+ gauge_value3.setValue(0L); // zero valued
Review Comment:
These variable names should be in lower camel case.
##########
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();
Review Comment:
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.
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/NoOpGauge.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.metrics;
+
+/**
+ * A no-op implementation of Gauge. This class exists to provide a default if an implementation of
+ * MetricsContainer does not override a Gauge getter.
+ */
+public class NoOpGauge implements Gauge {
+
+ private static final NoOpGauge singleton = new NoOpGauge();
+ private static final MetricName name = MetricName.named(NoOpGauge.class, "singleton");
Review Comment:
Static field names should be in upper snake case.
##########
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";
Review Comment:
Nit: `ESTIAMTED_BACKLOG_SIZE` -> `ESTIMATED_BACKLOG_SIZE`
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -93,6 +107,21 @@ public void updateSuccessfulRpcMetrics(String topic, Duration elapsedTime) {
}
}
+ /**
+ * @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.
+ */
Review Comment:
" Only" -> ". Only"?
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaSinkMetrics.java:
##########
@@ -71,6 +75,48 @@ public static Histogram createRPCLatencyHistogram(RpcMethod method, String topic
return new DelegatingHistogram(metricName, buckets, false, true);
}
+ /**
+ * Creates an Gauge metric to record per partition backlog. Metric will have name:
Review Comment:
Nit: `an Gauge` -> `a {@link Gauge}`, ". Metric will have name:" -> " with the name:"
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaSinkMetrics.java:
##########
@@ -71,6 +75,48 @@ public static Histogram createRPCLatencyHistogram(RpcMethod method, String topic
return new DelegatingHistogram(metricName, buckets, false, true);
}
+ /**
+ * Creates an Gauge metric to record per partition backlog. Metric will have name:
+ *
+ * <p>'EstimatedBacklogSize*topic_name:{topic};partitionId:{partitionId};'
+ *
+ * @param topic Kafka topic associated with this metric.
+ * @param partitionId partition id associated with this metric.
+ * @return Counter.
+ */
+ public static Gauge createBacklogGauge(String topic, int partitionId) {
+ return new DelegatingGauge(getMetricGaugeName(topic, partitionId), false, true);
+ }
+
+ /**
+ * Creates an Gauge metric to record per partition backlog. Metric will have name:
Review Comment:
Nit: an Gauge -> a {@link Gauge}, ". Metric will have name:" -> " with the name:"
##########
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() {
Review Comment:
Looks like this can be merged with `reportBacklog` (potentially rename that method to `reportBacklogMetrics` `updateBacklogMetrics`).
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java:
##########
@@ -302,13 +303,11 @@ public Instant getCurrentTimestamp() throws NoSuchElementException {
@Override
public long getSplitBacklogBytes() {
long backlogBytes = 0;
-
for (PartitionState<K, V> p : partitionStates) {
long pBacklog = p.approxBacklogInBytes();
if (pBacklog == UnboundedReader.BACKLOG_UNKNOWN) {
return UnboundedReader.BACKLOG_UNKNOWN;
}
- backlogBytes += pBacklog;
}
return backlogBytes;
Review Comment:
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`.
##########
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();
Review Comment:
Slightly unrelated, but why doesn't `perTopicRpcLatencies` use a gauge or sum as the value type?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2555650554
Run Java_Pulsar_IO_Direct PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2555809753
Run Java PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2610580992
Run Java PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33408:
URL: https://github.com/apache/beam/pull/33408#discussion_r1909343906
##########
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";
Review Comment:
Done
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaSinkMetrics.java:
##########
@@ -71,6 +75,48 @@ public static Histogram createRPCLatencyHistogram(RpcMethod method, String topic
return new DelegatingHistogram(metricName, buckets, false, true);
}
+ /**
+ * Creates an Gauge metric to record per partition backlog. Metric will have name:
Review Comment:
Thanks, updated the working for the histogram metric above as well to be consistent.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "johnjcasey (via GitHub)" <gi...@apache.org>.
johnjcasey merged PR #33408:
URL: https://github.com/apache/beam/pull/33408
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33408:
URL: https://github.com/apache/beam/pull/33408#discussion_r1920584519
##########
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);
Review Comment:
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.)
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2550299120
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`
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2555760755
Run Java PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2555837400
Run Java PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2599025616
Run Java_IOs_Direct PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2581404740
Flink test is failing, likely due to https://github.com/apache/beam/pull/33510,
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2549940587
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/33408?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 58.95%. Comparing base [(`8da9bbe`)](https://app.codecov.io/gh/apache/beam/commit/8da9bbe484f86bb1da48989ddb3d1655fcc82a13?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`b7b7e2a`)](https://app.codecov.io/gh/apache/beam/commit/b7b7e2a27060aeb2142a409a5e4536571309be66?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 190 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ 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](https://app.codecov.io/gh/apache/beam/pull/33408/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [java](https://app.codecov.io/gh/apache/beam/pull/33408/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.33% <ø> (+0.43%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/beam/pull/33408?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2551819436
Run Java PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33408:
URL: https://github.com/apache/beam/pull/33408#discussion_r1909343596
##########
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();
Review Comment:
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/KafkaMetrics.java:
##########
@@ -93,6 +107,21 @@ public void updateSuccessfulRpcMetrics(String topic, Duration elapsedTime) {
}
}
+ /**
+ * @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.
+ */
Review Comment:
Removed, thanks for catching that.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2605563960
Run Java PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "sjvanrossum (via GitHub)" <gi...@apache.org>.
sjvanrossum commented on code in PR #33408:
URL: https://github.com/apache/beam/pull/33408#discussion_r1920233993
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -67,15 +74,22 @@ abstract class KafkaMetricsImpl implements KafkaMetrics {
private static final Logger LOG = LoggerFactory.getLogger(KafkaMetricsImpl.class);
- static HashMap<String, Histogram> latencyHistograms = new HashMap<String, Histogram>();
+ static ConcurrentHashMap<String, Histogram> latencyHistograms =
+ new ConcurrentHashMap<String, Histogram>();
Review Comment:
```suggestion
private static final Map<String, Histogram> LATENCY_HISTOGRAMS =
new ConcurrentHashMap<>();
```
##########
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);
Review Comment:
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.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverterTest.java:
##########
@@ -345,4 +372,76 @@ public void testConvert_convertCountersAndHistograms() {
parsedMetricNames,
IsMapContaining.hasEntry(histogramMetricName, parsedHistogramMetricName));
}
+
+ @Test
+ public void testConvert_successfulyConvertGauges() {
Review Comment:
```suggestion
public void testConvert_successfullyConvertGauges() {
```
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -67,15 +74,22 @@ abstract class KafkaMetricsImpl implements KafkaMetrics {
private static final Logger LOG = LoggerFactory.getLogger(KafkaMetricsImpl.class);
- static HashMap<String, Histogram> latencyHistograms = new HashMap<String, Histogram>();
+ static ConcurrentHashMap<String, Histogram> latencyHistograms =
+ new ConcurrentHashMap<String, Histogram>();
+
+ abstract ConcurrentHashMap<String, ConcurrentLinkedQueue<Duration>> perTopicRpcLatencies();
- abstract HashMap<String, ConcurrentLinkedQueue<Duration>> perTopicRpcLatencies();
+ static ConcurrentHashMap<String, Gauge> backlogGauges = new ConcurrentHashMap<String, Gauge>();
Review Comment:
Hmmm... I searched for `backlogGauges`, but I'm not finding any uses of this it.
Is this still needed?
##########
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() {
Review Comment:
SGTM
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -67,15 +74,22 @@ abstract class KafkaMetricsImpl implements KafkaMetrics {
private static final Logger LOG = LoggerFactory.getLogger(KafkaMetricsImpl.class);
- static HashMap<String, Histogram> latencyHistograms = new HashMap<String, Histogram>();
+ static ConcurrentHashMap<String, Histogram> latencyHistograms =
+ new ConcurrentHashMap<String, Histogram>();
+
+ abstract ConcurrentHashMap<String, ConcurrentLinkedQueue<Duration>> perTopicRpcLatencies();
- abstract HashMap<String, ConcurrentLinkedQueue<Duration>> perTopicRpcLatencies();
+ static ConcurrentHashMap<String, Gauge> backlogGauges = new ConcurrentHashMap<String, Gauge>();
Review Comment:
```suggestion
private static final Map<String, Gauge> BACKLOG_GAUGES = new ConcurrentHashMap<>();
```
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "johnjcasey (via GitHub)" <gi...@apache.org>.
johnjcasey commented on code in PR #33408:
URL: https://github.com/apache/beam/pull/33408#discussion_r1892970875
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverter.java:
##########
@@ -86,6 +87,38 @@ private static Optional<MetricValue> convertCounterToMetricValue(
.setValueInt64(value));
}
+ /**
+ * @param metricName The {@link MetricName} that represents this counter.
+ * @param value The counter value.
+ * @return If the conversion succeeds, {@code MetricValue} that represents this counter. Otherwise
+ * returns an empty optional
+ */
+ private static Optional<MetricValue> convertGaugeToMetricValue(
+ MetricName metricName,
+ Long value,
+ Map<MetricName, LabeledMetricNameUtils.ParsedMetricName> parsedPerWorkerMetricsCache) {
+
+ if ((!metricName.getNamespace().equals(BigQuerySinkMetrics.METRICS_NAMESPACE)
Review Comment:
If so, I'd convert it to be an or condition, not a pair of nots and an and
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverter.java:
##########
@@ -86,6 +87,38 @@ private static Optional<MetricValue> convertCounterToMetricValue(
.setValueInt64(value));
}
+ /**
+ * @param metricName The {@link MetricName} that represents this counter.
+ * @param value The counter value.
+ * @return If the conversion succeeds, {@code MetricValue} that represents this counter. Otherwise
+ * returns an empty optional
+ */
+ private static Optional<MetricValue> convertGaugeToMetricValue(
+ MetricName metricName,
+ Long value,
+ Map<MetricName, LabeledMetricNameUtils.ParsedMetricName> parsedPerWorkerMetricsCache) {
+
+ if ((!metricName.getNamespace().equals(BigQuerySinkMetrics.METRICS_NAMESPACE)
Review Comment:
We probably shouldn't do this check here, and separate out the check from the conversion itself. To be clear, this is checking that the namespace is either BQ or Kafka, right?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverter.java:
##########
@@ -86,6 +87,38 @@ private static Optional<MetricValue> convertCounterToMetricValue(
.setValueInt64(value));
}
+ /**
+ * @param metricName The {@link MetricName} that represents this counter.
+ * @param value The counter value.
+ * @return If the conversion succeeds, {@code MetricValue} that represents this counter. Otherwise
+ * returns an empty optional
+ */
+ private static Optional<MetricValue> convertGaugeToMetricValue(
+ MetricName metricName,
+ Long value,
+ Map<MetricName, LabeledMetricNameUtils.ParsedMetricName> parsedPerWorkerMetricsCache) {
+
+ if ((!metricName.getNamespace().equals(BigQuerySinkMetrics.METRICS_NAMESPACE)
Review Comment:
or check that the namespace is in the set of supported namespaces
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2555837615
Run Java_GCP_IO_Direct PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2555896319
Run Java PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2555761134
Run Java_GCP_IO_Direct PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33408:
URL: https://github.com/apache/beam/pull/33408#discussion_r1909344437
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java:
##########
@@ -302,13 +303,11 @@ public Instant getCurrentTimestamp() throws NoSuchElementException {
@Override
public long getSplitBacklogBytes() {
long backlogBytes = 0;
-
for (PartitionState<K, V> p : partitionStates) {
long pBacklog = p.approxBacklogInBytes();
if (pBacklog == UnboundedReader.BACKLOG_UNKNOWN) {
return UnboundedReader.BACKLOG_UNKNOWN;
}
- backlogBytes += pBacklog;
}
return backlogBytes;
Review Comment:
yes I agree, this was deleted accidentally. Reverted that.
##########
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() {
Review Comment:
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).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2610547027
Run Java_IOs_Direct PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2606535231
Run Java_Hadoop_IO_Direct PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka add gauge v1 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33408:
URL: https://github.com/apache/beam/pull/33408#issuecomment-2607862243
Run Java PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org