address comments

Author: Naireen

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingStepMetricsContainer.java

@@ -386,9 +386,9 @@ private void deleteStaleCounters(
   @VisibleForTesting
   Iterable<PerStepNamespaceMetrics> extractPerWorkerMetricUpdates() {
     ImmutableMap.Builder<MetricName, Long> counters = new ImmutableMap.Builder<MetricName, Long>();
-    ImmutableMap.Builder<MetricName, Long> per_worker_gauges =
+    ImmutableMap.Builder<MetricName, Long> perWorkerGaugeUpdates =
         new ImmutableMap.Builder<MetricName, Long>();
-    ImmutableMap.Builder<MetricName, LockFreeHistogram.Snapshot> per_worker_histograms =
+    ImmutableMap.Builder<MetricName, LockFreeHistogram.Snapshot> perWorkerHistogramUpdates =
         new ImmutableMap.Builder<MetricName, LockFreeHistogram.Snapshot>();
     HashSet<MetricName> currentZeroValuedCounters = new HashSet<MetricName>();
     perWorkerCounters.forEach(
@@ -404,22 +404,22 @@ Iterable<PerStepNamespaceMetrics> extractPerWorkerMetricUpdates() {
     perWorkerGauges.forEach(
         (k, v) -> {
           Long val = v.getCumulative().value();
-          per_worker_gauges.put(k, val);
+          perWorkerGaugeUpdates.put(k, val);
           v.reset();
         });
 
     perWorkerHistograms.forEach(
         (k, v) -> {
-          v.getSnapshotAndReset().ifPresent(snapshot -> per_worker_histograms.put(k, snapshot));
+          v.getSnapshotAndReset().ifPresent(snapshot -> perWorkerHistogramUpdates.put(k, snapshot));
         });
 
     deleteStaleCounters(currentZeroValuedCounters, Instant.now(clock));
 
     return MetricsToPerStepNamespaceMetricsConverter.convert(
         stepName,
         counters.build(),
-        per_worker_gauges.build(),
-        per_worker_histograms.build(),
+        perWorkerGaugeUpdates.build(),
+        perWorkerHistogramUpdates.build(),
         parsedPerWorkerMetricsCache);
   }
 

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

@@ -50,7 +50,6 @@ private Optional<Histogram> getHistogram() {
     if (container == null) {
       return Optional.empty();
     }
-    // LOG.info("Xxx get histogram");
     return Optional.of(container.getHistogram(name, bucketType));
   }
 

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySinkMetrics.java

@@ -114,8 +114,8 @@ static Histogram createRPCLatencyHistogram(RpcMethod method) {
     LabeledMetricNameUtils.MetricNameBuilder nameBuilder =
         LabeledMetricNameUtils.MetricNameBuilder.baseNameBuilder(RPC_LATENCY);
     nameBuilder.addLabel(RPC_METHOD, method.toString());
-    MetricName metricName = nameBuilder.build(METRICS_NAMESPACE);
     nameBuilder.addMetricLabel(MonitoringInfoConstants.Labels.PER_WORKER_METRIC, "true");
+    MetricName metricName = nameBuilder.build(METRICS_NAMESPACE);
 
     // Create Exponential histogram buckets with the following parameters:
     // 0 scale, resulting in bucket widths with a size 2 growth factor.

sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySinkMetricsTest.java

@@ -41,6 +41,7 @@
 import org.apache.beam.sdk.metrics.MetricsEnvironment;
 import org.apache.beam.sdk.util.HistogramData;
 import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
 import org.hamcrest.collection.IsMapContaining;
 import org.junit.Test;
@@ -203,7 +204,10 @@ public void testReportSuccessfulRpcMetrics() throws Exception {
     MetricName counterNameDisabledDeletes =
         MetricName.named("BigQuerySink", "RpcRequestsCount*rpc_method:APPEND_ROWS;rpc_status:OK;");
     MetricName histogramName =
-        MetricName.named("BigQuerySink", "RpcLatency*rpc_method:APPEND_ROWS;");
+        MetricName.named(
+            "BigQuerySink",
+            "RpcLatency*rpc_method:APPEND_ROWS;",
+            ImmutableMap.of("PER_WORKER_METRIC", "true"));
     HistogramData.BucketType bucketType = HistogramData.ExponentialBuckets.of(0, 17);
     testContainer.assertPerWorkerCounterValue(counterNameDisabledDeletes, 1L);
     testContainer.assertPerWorkerHistogramValues(histogramName, bucketType, 3.0);
@@ -243,7 +247,10 @@ public void testReportFailedRPCMetrics_KnownGrpcError() throws Exception {
         MetricName.named(
             "BigQuerySink", "RpcRequestsCount*rpc_method:APPEND_ROWS;rpc_status:NOT_FOUND;");
     MetricName histogramName =
-        MetricName.named("BigQuerySink", "RpcLatency*rpc_method:APPEND_ROWS;");
+        MetricName.named(
+            "BigQuerySink",
+            "RpcLatency*rpc_method:APPEND_ROWS;",
+            ImmutableMap.of("PER_WORKER_METRIC", "true"));
     HistogramData.BucketType bucketType = HistogramData.ExponentialBuckets.of(0, 17);
     testContainer.assertPerWorkerCounterValue(counterNameDisabledDeletes, 1L);
     testContainer.assertPerWorkerHistogramValues(histogramName, bucketType, 5.0);
@@ -282,7 +289,10 @@ public void testReportFailedRPCMetrics_UnknownGrpcError() throws Exception {
         MetricName.named(
             "BigQuerySink", "RpcRequestsCount*rpc_method:APPEND_ROWS;rpc_status:UNKNOWN;");
     MetricName histogramName =
-        MetricName.named("BigQuerySink", "RpcLatency*rpc_method:APPEND_ROWS;");
+        MetricName.named(
+            "BigQuerySink",
+            "RpcLatency*rpc_method:APPEND_ROWS;",
+            ImmutableMap.of("PER_WORKER_METRIC", "true"));
     HistogramData.BucketType bucketType = HistogramData.ExponentialBuckets.of(0, 17);
     testContainer.assertPerWorkerCounterValue(counterNameDisabledDeletes, 1L);
     testContainer.assertPerWorkerHistogramValues(histogramName, bucketType, 15.0);

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

@@ -71,8 +71,8 @@ public static Histogram createRPCLatencyHistogram(RpcMethod method, String topic
     nameBuilder.addLabel(RPC_METHOD, method.toString());
     nameBuilder.addLabel(TOPIC_LABEL, topic);
 
-    MetricName metricName = nameBuilder.build(METRICS_NAMESPACE);
     nameBuilder.addMetricLabel(MonitoringInfoConstants.Labels.PER_WORKER_METRIC, "true");
+    MetricName metricName = nameBuilder.build(METRICS_NAMESPACE);
 
     HistogramData.BucketType buckets = HistogramData.ExponentialBuckets.of(1, 17);
     return new DelegatingHistogram(metricName, buckets, false);

sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaMetricsTest.java

@@ -121,7 +121,10 @@ public void testKafkaRPCLatencyMetrics() throws Exception {
     results.flushBufferedMetrics();
     // RpcLatency*rpc_method:POLL;topic_name:test-topic
     MetricName histogramName =
-        MetricName.named("KafkaSink", "RpcLatency*rpc_method:POLL;topic_name:test-topic;");
+        MetricName.named(
+            "KafkaSink",
+            "RpcLatency*rpc_method:POLL;topic_name:test-topic;",
+            ImmutableMap.of("PER_WORKER_METRIC", "true"));
     HistogramData.BucketType bucketType = HistogramData.ExponentialBuckets.of(1, 17);
 
     assertThat(testContainer.histograms.size(), equalTo(1));

sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaSinkMetricsTest.java

@@ -38,7 +38,10 @@ public void testCreatingHistogram() throws Exception {
         KafkaSinkMetrics.createRPCLatencyHistogram(KafkaSinkMetrics.RpcMethod.POLL, "topic1");
 
     MetricName histogramName =
-        MetricName.named("KafkaSink", "RpcLatency*rpc_method:POLL;topic_name:topic1;");
+        MetricName.named(
+            "KafkaSink",
+            "RpcLatency*rpc_method:POLL;topic_name:topic1;",
+            ImmutableMap.of("PER_WORKER_METRIC", "true"));
     assertThat(histogram.getName(), equalTo(histogramName));
   }