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 2025/01/06 17:40:10 UTC
[PR] Kafka add counters v1 uw2 [beam]
Naireen opened a new pull request, #33503:
URL: https://github.com/apache/beam/pull/33503
Add per worker metadata on metrics propaged over FnApi. This is on top of https://github.com/apache/beam/pull/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](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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1986936703
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingStepMetricsContainer.java:
##########
@@ -402,12 +387,17 @@ Iterable<PerStepNamespaceMetrics> extractPerWorkerMetricUpdates() {
counters.put(k, val);
});
- perWorkerGauges.forEach(
+ gauges.forEach(
(k, v) -> {
- Long val = v.getCumulative().value();
- gauges.put(k, val);
- v.reset();
+ // Check if metric name has the per worker label set
+ if (k.getLabels().containsKey("PER_WORKER_METRIC")
Review Comment:
Oh that makes sense, so we don't have to check each time we process an update.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingStepMetricsContainer.java:
##########
@@ -402,12 +387,17 @@ Iterable<PerStepNamespaceMetrics> extractPerWorkerMetricUpdates() {
counters.put(k, val);
});
- perWorkerGauges.forEach(
+ gauges.forEach(
(k, v) -> {
- Long val = v.getCumulative().value();
- gauges.put(k, val);
- v.reset();
+ // Check if metric name has the per worker label set
+ if (k.getLabels().containsKey("PER_WORKER_METRIC")
Review Comment:
Obsolete after code changes.
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1986937733
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -37,6 +37,7 @@ public interface KafkaMetrics {
void updateBacklogBytes(String topic, int partitionId, long backlog);
+ /*Used to update all metrics in container*/
Review Comment:
Done
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2576322151
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 counters v1 uw2 [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2607090443
Reminder, please take a look at this pr: @kennknowles @damccorm @damondouglas @sjvanrossum
--
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 counters v1 uw2 [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1975424830
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/LabeledMetricNameUtils.java:
##########
@@ -63,8 +65,18 @@ public void addLabel(String key, String value) {
.append(LABEL_DELIMITER);
}
+ /**
+ * Add a metric label KV pair to the metric. This is not concatenated as a part of the name, but
+ * merely for adding attributes to the metric.
+ */
+ public void addMetricLabel(String key, String value) {
+ System.out.println("xxx add metric lable");
Review Comment:
rm
##########
sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/GcpCoreApiSurfaceTest.java:
##########
@@ -40,6 +40,7 @@ public void testGcpCoreApiSurface() throws Exception {
final ApiSurface apiSurface =
ApiSurface.ofPackage(thisPackage, thisClassLoader)
.pruningPattern("org[.]apache[.]beam[.].*Test.*")
+ .pruningPattern("org[.]apache[.]beam[.].*vendor.*")
Review Comment:
is it correct to prune this? It seemsl like we might not want to expose our vendored libs. I'm not sure why it is needed though, since where you add vendor guava deps it seems like they were already there.
##########
model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/metrics.proto:
##########
@@ -447,6 +447,7 @@ message MonitoringInfo {
SPANNER_TABLE_ID = 25 [(label_props) = { name: "SPANNER_TABLE_ID" }];
SPANNER_INSTANCE_ID = 26 [(label_props) = { name: "SPANNER_INSTANCE_ID" }];
SPANNER_QUERY_NAME = 27 [(label_props) = { name: "SPANNER_QUERY_NAME" }];
+ PER_WORKER_METRIC = 28 [(label_props) = { name: "PER_WORKER_METRIC" }];
Review Comment:
How about a comment like:
// Label which if has a "true" value indicates that the metric is intended to be aggregated per-worker.
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2691239705
R: @Abacn - adding you to review deps
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2574095321
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1986936298
##########
runners/google-cloud-dataflow-java/worker/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 =
Review Comment:
I agree, that should be done for all of them, will do that in the followup clean up CL.
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2691771020
Run Python_Dataframes PreCommit 3.9
--
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 counters v1 uw2 [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1962369386
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -146,14 +154,31 @@ private void recordRpcLatencyMetrics() {
}
}
- private void recordBacklogBytes() {
+ private void recordBacklogBytesInternal() {
for (Map.Entry<String, Long> backlogs : perTopicPartitionBacklogs().entrySet()) {
Gauge gauge =
KafkaSinkMetrics.createBacklogGauge(MetricName.named("KafkaSink", backlogs.getKey()));
gauge.set(backlogs.getValue());
}
}
+ /**
+ * This is for recording backlog bytes on the current thread.
+ *
+ * @param topicName topicName
+ * @param partitionId partitionId for the topic Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ * @param backlogBytes backlog for the topic Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ */
+ @Override
+ public void recordBacklogBytes(String topicName, int partitionId, long backlogBytes) {
+ Gauge perPartion =
+ Metrics.gauge(
Review Comment:
ie can we create the gauge here with the getPerWorkerGauge method? and then ensure that those metrics have the per-worker metadata populated appropriately?
##########
runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerImpl.java:
##########
@@ -293,6 +293,13 @@ public MetricUpdates getUpdates() {
.setLabel(MonitoringInfoConstants.Labels.NAME, metricKey.metricName().getName())
.setLabel(MonitoringInfoConstants.Labels.PTRANSFORM, metricKey.stepName());
}
+
+ // Based on namespace, add per worker metrics label to enable separate runner based sink based
+ // processing.
+ if (metricName.getNamespace().equals("BigQuerySink")
Review Comment:
it seems like this woudl be cleaner if it was at the call-site when creating the metric.
That would both make it more obvious there and less based upon magic constant and would allow us to support some kafka metrics per-worker but others aggregated or add per-worker metrics to some namespace that already has metrics.
It seems like maybe we could either add it to MetricKey, or we could plumb it into the cell and then update the function where we generate metadata to base it upon that instead.
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java:
##########
@@ -332,7 +332,6 @@ public long getSplitBacklogBytes() {
if (pBacklog == UnboundedReader.BACKLOG_UNKNOWN) {
return UnboundedReader.BACKLOG_UNKNOWN;
}
- backlogBytes += pBacklog;
Review Comment:
revert?
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2675195329
Run PythonDocker PreCommit 3.10
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2652282901
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2576117704
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2652312127
Run Python_Dataframes PreCommit 3.10
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2655183303
Run PythonDocker PreCommit 3.12
--
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 counters v1 uw2 [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2612773732
stop reviewer notifications
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1972892177
##########
sdks/java/io/kafka/build.gradle:
##########
@@ -52,6 +52,7 @@ dependencies {
provided library.java.jackson_dataformat_csv
permitUnusedDeclared library.java.jackson_dataformat_csv
implementation project(path: ":sdks:java:core", configuration: "shadow")
+ implementation project(path: ":runners:core-java")
Review Comment:
cc @Abacn thoughts?
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2712130773
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1979999342
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -146,14 +154,30 @@ private void recordRpcLatencyMetrics() {
}
}
- private void recordBacklogBytes() {
+ private void recordBacklogBytesInternal() {
for (Map.Entry<String, Long> backlogs : perTopicPartitionBacklogs().entrySet()) {
Gauge gauge =
KafkaSinkMetrics.createBacklogGauge(MetricName.named("KafkaSink", backlogs.getKey()));
gauge.set(backlogs.getValue());
}
}
+ /**
+ * This is for recording backlog bytes on the current thread.
+ *
+ * @param topicName topicName
+ * @param partitionId partitionId for the topic Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ * @param backlogBytes backlog for the topic Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ */
+ @Override
+ public void recordBacklogBytes(String topicName, int partitionId, long backlogBytes) {
Review Comment:
Added comments, one is for updating the metrics container on the current thread, and one is for when its not on the current thread, so you update a map and record values, and once you're on the main processing thread of the dofn, then you update/create gauges with all the observed values (this is from the legacy implementation since metrics are updated in the background).
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -39,6 +40,8 @@ public interface KafkaMetrics {
void updateKafkaMetrics();
+ void recordBacklogBytes(String topic, int partitionId, long backlog);
Review Comment:
We could, where we store all values in a map, and then create gauges afterwards, but we don't need to in this case, we can directly create the gauges. See comments below
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2686948279
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2691319637
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2691766946
Run Java_Flink_Versions 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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2711633428
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2657408735
Run Prism_Python PreCommit 3.12
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2657408393
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2576193521
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.
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2655004578
Run GoPortable 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 counters v1 uw2 [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2711396470
Run Python_Dataframes 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 counters v1 uw2 [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #33503:
URL: https://github.com/apache/beam/pull/33503
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2691767220
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2686618735
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2674992575
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2675753371
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2576375779
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2657581561
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2686193371
Run Spotless 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 counters v1 uw2 [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2691655057
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/33503?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 57.74%. Comparing base [(`c8172fb`)](https://app.codecov.io/gh/apache/beam/commit/c8172fbaadb4c11373baed9d2d48fc8175ed0699?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`658a31e`)](https://app.codecov.io/gh/apache/beam/commit/658a31eeac0f143f17ca8e3523d089ffdc91cbb0?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 8 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #33503 +/- ##
=========================================
Coverage 57.74% 57.74%
Complexity 1474 1474
=========================================
Files 995 995
Lines 157688 157686 -2
Branches 1076 1076
=========================================
Hits 91058 91058
+ Misses 64381 64379 -2
Partials 2249 2249
```
| [Flag](https://app.codecov.io/gh/apache/beam/pull/33503/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [python](https://app.codecov.io/gh/apache/beam/pull/33503/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.28% <ø> (+<0.01%)` | :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/33503?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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1982354211
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -146,14 +154,30 @@ private void recordRpcLatencyMetrics() {
}
}
- private void recordBacklogBytes() {
+ private void recordBacklogBytesInternal() {
for (Map.Entry<String, Long> backlogs : perTopicPartitionBacklogs().entrySet()) {
Gauge gauge =
KafkaSinkMetrics.createBacklogGauge(MetricName.named("KafkaSink", backlogs.getKey()));
gauge.set(backlogs.getValue());
}
}
+ /**
+ * This is for recording backlog bytes on the current thread.
+ *
+ * @param topicName topicName
+ * @param partitionId partitionId for the topic Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ * @param backlogBytes backlog for the topic Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ */
+ @Override
+ public void recordBacklogBytes(String topicName, int partitionId, long backlogBytes) {
Review Comment:
So the metrics are based on the implementation of the IO, so for non-sdf kakfa impl is run on v2, the metrics are on the background thread, so it would be updated accordingly, and it would still have the per worker label. (Since the labels are added for all metrics, so that should still be supported)
For the case of sdf impl on v1, the metrics would be created on the container, but likely would not be exported correctly, since we use stageInfo to do that, which requires per worker metrics, which would not include the sdf Kafka metrics.
Your comment:
>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?
KafkaMetrics does already always set the label, by the `dataflow metric container`, do you mean on v1? for v2, there really isn't a separate `dataflow metric container` container, the runner v2 worker just consumes whatever metrics are included in the [processBundle response here:](https://github.com/apache/beam/blob/c7edbb3dda13807c88a4397574b9b016d9dff19f/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ProcessBundleHandler.java#L492), whos metric container doesn't even have a perWorker counter method (by design to avoid bloating it)
I do agree with your approach, and if I understand correctly, then we’d want to really update this [file]( https://github.com/apache/beam/blob/c7edbb3dda13807c88a4397574b9b016d9dff19f/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingStepMetricsContainer.java#L389) So we no longer have perWorker counter handling (instead use the metric labels to aggregate those metrics separately). Note that this is for the legacy worker path.
This should more cleanly support the two use cases you’ve mentioned, but would be a larger refactor to make things more consistent, since we’d want to port those changes over to BQ as well at some point.
Much of the legacy per worker reporting path is shared, so any changes there would need some BQ IO changes as well to ensure those metrics aren’t broken.
Regardless of that refactoring though, that doesn’t change the fact that we’d still need a way to know if metrics are updated on the current thread or not to ensure it’s added to the metric container correctly. If counters are added not on the current thread, [counters are dropped](https://github.com/apache/beam/blob/c7edbb3dda13807c88a4397574b9b016d9dff19f/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricsEnvironment.java#L144). Its not clear to me how that could be made cleaner. Open to suggestions though.
--
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 counters v1 uw2 [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1981552498
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -146,14 +154,30 @@ private void recordRpcLatencyMetrics() {
}
}
- private void recordBacklogBytes() {
+ private void recordBacklogBytesInternal() {
for (Map.Entry<String, Long> backlogs : perTopicPartitionBacklogs().entrySet()) {
Gauge gauge =
KafkaSinkMetrics.createBacklogGauge(MetricName.named("KafkaSink", backlogs.getKey()));
gauge.set(backlogs.getValue());
}
}
+ /**
+ * This is for recording backlog bytes on the current thread.
+ *
+ * @param topicName topicName
+ * @param partitionId partitionId for the topic Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ * @param backlogBytes backlog for the topic Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ */
+ @Override
+ public void recordBacklogBytes(String topicName, int partitionId, long backlogBytes) {
Review Comment:
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?
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2574095622
Run Python_Coverage 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 counters v1 uw2 [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2576173507
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, 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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2674790826
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2657268372
Reopening this,
This consistently fails though due to a known flakey test: https://github.com/apache/beam/issues/28957
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2652283216
Run Java_PVR_Prism_Loopback 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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2691766332
Run Java_Amazon-Web-Services2_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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2691322706
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2691563125
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 counters v1 uw2 [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1970452713
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosImpl.java:
##########
@@ -1068,5 +1069,10 @@ public String getNamespace() {
public String getName() {
return name;
}
+
+ @Override
+ public HashMap<String, String> getLabels() {
+ return new HashMap<>();
Review Comment:
ImmutableMap.of() is a singleton
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/LabeledMetricNameUtils.java:
##########
@@ -42,6 +43,7 @@ public class LabeledMetricNameUtils {
*/
public static class MetricNameBuilder {
private final StringBuilder labeledNameBuilder;
+ private HashMap<String, String> metricLabels = new HashMap<String, String>();
Review Comment:
How about ImmutableMap builder? and then build to immutable map below?
##########
sdks/java/io/kafka/build.gradle:
##########
@@ -52,6 +52,7 @@ dependencies {
provided library.java.jackson_dataformat_csv
permitUnusedDeclared library.java.jackson_dataformat_csv
implementation project(path: ":sdks:java:core", configuration: "shadow")
+ implementation project(path: ":runners:core-java")
Review Comment:
I would appreciate someone more familiar with desired deps structure to review this.
It might be preferrable to move the constants to somewhere in core out of runners
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/LabeledMetricNameUtils.java:
##########
@@ -63,8 +65,12 @@ public void addLabel(String key, String value) {
.append(LABEL_DELIMITER);
}
+ public void addMetricLabel(String key, String value) {
Review Comment:
needs a comment, it is a little confusing on how this is different from addLabel
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricName.java:
##########
@@ -48,12 +53,24 @@ public String toString() {
public static MetricName named(String namespace, String name) {
checkArgument(!Strings.isNullOrEmpty(namespace), "Metric namespace must be non-empty");
checkArgument(!Strings.isNullOrEmpty(name), "Metric name must be non-empty");
- return new AutoValue_MetricName(namespace, name);
+ return new AutoValue_MetricName(namespace, name, new HashMap<String, String>());
+ }
+
+ public static MetricName named(String namespace, String name, HashMap<String, String> labels) {
Review Comment:
change the map type to immutablemap throughout (change to ImmutableMap.of() for the empty ones)
Nice to have immutable to avoid possible changes to the labels. And it is optimized for small maps
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2698986989
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 counters v1 uw2 [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2588295206
Would you mind adding some more context on what this change is doing?
> This is on top of https://github.com/apache/beam/pull/33408
2 questions:
1) Does this mean that this is blocked on that PR or just related?
2) Is it better to have the same person/people review both PRs?
--
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 counters v1 uw2 [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r2014965191
##########
sdks/java/io/kafka/build.gradle:
##########
@@ -52,6 +52,7 @@ dependencies {
provided library.java.jackson_dataformat_csv
permitUnusedDeclared library.java.jackson_dataformat_csv
implementation project(path: ":sdks:java:core", configuration: "shadow")
+ implementation project(path: ":runners:core-java")
Review Comment:
Missed that. Yes kafka io should not depends on runners. This does break internal import
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2575981102
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2576113607
tests failing due to https://github.com/apache/beam/issues/28957
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2574163322
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2655004736
Run Go 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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2675195727
Run Prism_Python PreCommit 3.12
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2675753913
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2590815310
> Would you mind adding some more context on what this change is doing?
>
> > This is on top of #33408
>
> 2 questions:
>
> 1. Does this mean that this is blocked on that PR or just related?
> 2. Is it better to have the same person/people review both PRs?
1) Yes, this is blocked on that PR
2) I agree, it would be best to have the same reviewers as #33408
I moved this to a draft for now, will rebase this and update to review once this is ready again.
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2695361640
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2698985977
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 counters v1 uw2 [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1984730676
##########
runners/google-cloud-dataflow-java/worker/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 =
Review Comment:
nit: I don't think this (or counters) needs to be Concurrent. How about using ImmutableMapBuilder and then build when passing to convert?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingStepMetricsContainer.java:
##########
@@ -402,12 +387,17 @@ Iterable<PerStepNamespaceMetrics> extractPerWorkerMetricUpdates() {
counters.put(k, val);
});
- perWorkerGauges.forEach(
+ gauges.forEach(
(k, v) -> {
- Long val = v.getCumulative().value();
- gauges.put(k, val);
- v.reset();
+ // Check if metric name has the per worker label set
+ if (k.getLabels().containsKey("PER_WORKER_METRIC")
+ && k.getLabels().get("PER_WORKER_METRIC").equals("true")) {
Review Comment:
nit: just lookup label once
@Nullable String value = k.getLables().get("PER_WORKER_METRIC");
if (value != null && value.equals("true"))
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -37,6 +37,7 @@ public interface KafkaMetrics {
void updateBacklogBytes(String topic, int partitionId, long backlog);
+ /*Used to update all metrics in container*/
Review Comment:
how about
Flushes the buffered metrics to the current metric container for this thread.
could we rename it to flushBufferedMetrics?
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaSinkMetrics.java:
##########
@@ -97,7 +86,15 @@ public static Gauge createBacklogGauge(String topic, int partitionId) {
* @return Counter.
*/
public static Gauge createBacklogGauge(MetricName name) {
- return new DelegatingGauge(name, false, true);
+ // Use label to differenciate between the type of gauge metric is created
+ // TODO(34195):
+ if (name.getLabels().containsKey(MonitoringInfoConstants.Labels.PER_WORKER_METRIC)
Review Comment:
maybe we could add a method for this in MonitoringInfoConstant? woudl reduce duplication of this lookup/comparison logic:
boolean isPerWorkerMetric(Map labels)
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingStepMetricsContainer.java:
##########
@@ -402,12 +387,17 @@ Iterable<PerStepNamespaceMetrics> extractPerWorkerMetricUpdates() {
counters.put(k, val);
});
- perWorkerGauges.forEach(
+ gauges.forEach(
(k, v) -> {
- Long val = v.getCumulative().value();
- gauges.put(k, val);
- v.reset();
+ // Check if metric name has the per worker label set
+ if (k.getLabels().containsKey("PER_WORKER_METRIC")
Review Comment:
use the PER_WORKER_METRIC contant
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaSinkMetrics.java:
##########
@@ -97,7 +86,15 @@ public static Gauge createBacklogGauge(String topic, int partitionId) {
* @return Counter.
*/
public static Gauge createBacklogGauge(MetricName name) {
- return new DelegatingGauge(name, false, true);
+ // Use label to differenciate between the type of gauge metric is created
+ // TODO(34195):
+ if (name.getLabels().containsKey(MonitoringInfoConstants.Labels.PER_WORKER_METRIC)
+ && name.getLabels().get(MonitoringInfoConstants.Labels.PER_WORKER_METRIC).equals("true")) {
+ return new DelegatingGauge(name, false);
+ } else {
+ // Currently KafkaSink metrics only supports aggregated per worker metrics.
Review Comment:
should we make it a Preconditions.checkState(isPerWorkerMetric))?
seems perhaps better to fail so that if more metrics are added it's not confusing why they don't do anything
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingStepMetricsContainer.java:
##########
@@ -402,12 +387,17 @@ Iterable<PerStepNamespaceMetrics> extractPerWorkerMetricUpdates() {
counters.put(k, val);
});
- perWorkerGauges.forEach(
+ gauges.forEach(
(k, v) -> {
- Long val = v.getCumulative().value();
- gauges.put(k, val);
- v.reset();
+ // Check if metric name has the per worker label set
+ if (k.getLabels().containsKey("PER_WORKER_METRIC")
Review Comment:
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.
--
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 counters v1 uw2 [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1979108960
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -39,6 +40,8 @@ public interface KafkaMetrics {
void updateKafkaMetrics();
+ void recordBacklogBytes(String topic, int partitionId, long backlog);
Review Comment:
can we just modify already public updateBacklogBytes?
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -146,14 +154,30 @@ private void recordRpcLatencyMetrics() {
}
}
- private void recordBacklogBytes() {
+ private void recordBacklogBytesInternal() {
for (Map.Entry<String, Long> backlogs : perTopicPartitionBacklogs().entrySet()) {
Gauge gauge =
KafkaSinkMetrics.createBacklogGauge(MetricName.named("KafkaSink", backlogs.getKey()));
gauge.set(backlogs.getValue());
}
}
+ /**
+ * This is for recording backlog bytes on the current thread.
+ *
+ * @param topicName topicName
+ * @param partitionId partitionId for the topic Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ * @param backlogBytes backlog for the topic Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ */
+ @Override
+ public void recordBacklogBytes(String topicName, int partitionId, long backlogBytes) {
Review Comment:
can we do this in recordBacklogBytesInternal instead?
It's not clear from comments/code why both of these are needed or how they differ
--
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 counters v1 uw2 [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2711396108
Run Java_PVR_Prism_Loopback 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 counters v1 uw2 [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1986951687
##########
runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MonitoringInfoConstants.java:
##########
@@ -212,4 +214,12 @@ static String extractUrn(MonitoringInfoSpecs.Enum value) {
private static String extractLabel(MonitoringInfo.MonitoringInfoLabels value) {
return value.getValueDescriptor().getOptions().getExtension(labelProps).getName();
}
+
+ public static boolean isPerWorkerMetric(MetricName metricName) {
+ @Nullable String value = metricName.getLabels().get("PER_WORKER_METRIC");
Review Comment:
use the constant
--
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 counters v1 uw2 [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2612776319
Stopping reviewer notifications for this pull request: requested by reviewer. 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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2574095055
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 counters v1 uw2 [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2711395558
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2711633041
Run Python_Dataframes PreCommit 3.10
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2712195839
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2712259604
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2655003988
Run Python_Dataframes PreCommit 3.10
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2655004411
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2698741185
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1964918863
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -146,14 +154,31 @@ private void recordRpcLatencyMetrics() {
}
}
- private void recordBacklogBytes() {
+ private void recordBacklogBytesInternal() {
for (Map.Entry<String, Long> backlogs : perTopicPartitionBacklogs().entrySet()) {
Gauge gauge =
KafkaSinkMetrics.createBacklogGauge(MetricName.named("KafkaSink", backlogs.getKey()));
gauge.set(backlogs.getValue());
}
}
+ /**
+ * This is for recording backlog bytes on the current thread.
+ *
+ * @param topicName topicName
+ * @param partitionId partitionId for the topic Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ * @param backlogBytes backlog for the topic Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ */
+ @Override
+ public void recordBacklogBytes(String topicName, int partitionId, long backlogBytes) {
+ Gauge perPartion =
+ Metrics.gauge(
Review Comment:
We create it as a regular gauge, and then add per worker metadata to differenciate between the two (with the metric labels). If we add a new api here, it would then be added to any other derived metrics, (and then we'd likely want per worker other sorts of metrics.) Keeping the original set (gauge, hist, counter, etc) and then adding metadata for runner aggregation might be a cleaner approach?
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java:
##########
@@ -332,7 +332,6 @@ public long getSplitBacklogBytes() {
if (pBacklog == UnboundedReader.BACKLOG_UNKNOWN) {
return UnboundedReader.BACKLOG_UNKNOWN;
}
- backlogBytes += pBacklog;
Review Comment:
Thanks for catching!
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2674790145
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2675851775
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1984058553
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -146,14 +154,30 @@ private void recordRpcLatencyMetrics() {
}
}
- private void recordBacklogBytes() {
+ private void recordBacklogBytesInternal() {
for (Map.Entry<String, Long> backlogs : perTopicPartitionBacklogs().entrySet()) {
Gauge gauge =
KafkaSinkMetrics.createBacklogGauge(MetricName.named("KafkaSink", backlogs.getKey()));
gauge.set(backlogs.getValue());
}
}
+ /**
+ * This is for recording backlog bytes on the current thread.
+ *
+ * @param topicName topicName
+ * @param partitionId partitionId for the topic Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ * @param backlogBytes backlog for the topic Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ */
+ @Override
+ public void recordBacklogBytes(String topicName, int partitionId, long backlogBytes) {
Review Comment:
Discussed offline.
I refactored the per worker gauge metrics out, so the path is now more unified. Theres no worry about breaking BQ per worker metrics, since that uses only counters and histograms.
Next steps would be to do the same refactoring for histograms and counters, and then add in the kafka poll metrics for the portable runner (and refactor as needed)
--
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 counters v1 uw2 [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2576195344
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).
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2576196011
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2574094697
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2575827855
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2655183628
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2657582052
R @scwhittle
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2686467378
Run PythonDocker PreCommit 3.12
--
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 counters v1 uw2 [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1977517724
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricName.java:
##########
@@ -48,12 +53,26 @@ public String toString() {
public static MetricName named(String namespace, String name) {
checkArgument(!Strings.isNullOrEmpty(namespace), "Metric namespace must be non-empty");
checkArgument(!Strings.isNullOrEmpty(name), "Metric name must be non-empty");
- return new AutoValue_MetricName(namespace, name);
+ return new AutoValue_MetricName(namespace, name, ImmutableMap.of());
+ }
+
+ public static MetricName named(
+ String namespace, String name, ImmutableMap<String, String> labels) {
Review Comment:
a possible fix for the vendor issue would be to change this back to map and do
ImmutableMap.copyOf when passing to AutoValue.
This might be nicer api since it makes tests easier etc. If callers use ImmutableMap there is no performance overhead for the copy.
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1975807380
##########
sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/GcpCoreApiSurfaceTest.java:
##########
@@ -40,6 +40,7 @@ public void testGcpCoreApiSurface() throws Exception {
final ApiSurface apiSurface =
ApiSurface.ofPackage(thisPackage, thisClassLoader)
.pruningPattern("org[.]apache[.]beam[.].*Test.*")
+ .pruningPattern("org[.]apache[.]beam[.].*vendor.*")
Review Comment:
I think its because of the changes to metricName (a public method), we've added ImmutableMaps (which is coming from our vendored deps) as a part of the interface.
--
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2691764917
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2691767861
Run Java_PVR_Prism_Loopback 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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2691774350
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 counters v1 uw2 [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1964919555
##########
runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerImpl.java:
##########
@@ -293,6 +293,13 @@ public MetricUpdates getUpdates() {
.setLabel(MonitoringInfoConstants.Labels.NAME, metricKey.metricName().getName())
.setLabel(MonitoringInfoConstants.Labels.PTRANSFORM, metricKey.stepName());
}
+
+ // Based on namespace, add per worker metrics label to enable separate runner based sink based
+ // processing.
+ if (metricName.getNamespace().equals("BigQuerySink")
Review Comment:
Done. Removed BQ sink support, that can be done later. I've added it to MetricName
--
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 counters v1 uw2 [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #33503:
URL: https://github.com/apache/beam/pull/33503#issuecomment-2710251114
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