Add support to record multiple values at once in Histogram metric (ap…

…ache#30717)

* Add support to record multiple values at once in Histogram metric

* Address comments

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

@@ -18,6 +18,7 @@
 package org.apache.beam.sdk.metrics;
 
 import java.io.Serializable;
+import java.util.Optional;
 import org.apache.beam.sdk.annotations.Internal;
 import org.apache.beam.sdk.util.HistogramData;
 
@@ -60,22 +61,31 @@ public DelegatingHistogram(
     this.perWorkerHistogram = perWorkerHistogram;
   }
 
-  @Override
-  public void update(double value) {
+  private Optional<Histogram> getHistogram() {
     MetricsContainer container =
         processWideContainer
             ? MetricsEnvironment.getProcessWideContainer()
             : MetricsEnvironment.getCurrentContainer();
     if (container == null) {
-      return;
+      return Optional.empty();
     }
     if (perWorkerHistogram) {
-      container.getPerWorkerHistogram(name, bucketType).update(value);
+      return Optional.of(container.getPerWorkerHistogram(name, bucketType));
     } else {
-      container.getHistogram(name, bucketType).update(value);
+      return Optional.of(container.getHistogram(name, bucketType));
     }
   }
 
+  @Override
+  public void update(double value) {
+    getHistogram().ifPresent(histogram -> histogram.update(value));
+  }
+
+  @Override
+  public void update(double... values) {
+    getHistogram().ifPresent(histogram -> histogram.update(values));
+  }
+
   @Override
   public MetricName getName() {
     return name;

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

@@ -21,4 +21,11 @@
 public interface Histogram extends Metric {
   /** Add an observation to this histogram. */
   void update(double value);
+
+  /** Add observations to this histogram. */
+  default void update(double... values) {
+    for (double value : values) {
+      update(value);
+    }
+  }
 }

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

@@ -31,6 +31,9 @@ private NoOpHistogram() {}
   @Override
   public void update(double value) {}
 
+  @Override
+  public void update(double... value) {}
+
   @Override
   public MetricName getName() {
     return name;

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

@@ -156,6 +156,16 @@ public void updateSuccessfulAndFailedRows(int totalRows, int failedRows) {
       }
     }
 
+    /** Record rpc latency histogram metrics. */
+    private void recordRpcLatencyMetrics() {
+      Histogram latencyHistogram =
+          BigQuerySinkMetrics.createRPCLatencyHistogram(
+              BigQuerySinkMetrics.RpcMethod.STREAMING_INSERTS);
+      double[] rpcLatencies =
+          rpcLatencies().stream().mapToDouble(duration -> duration.toMillis()).toArray();
+      latencyHistogram.update(rpcLatencies);
+    }
+
     /**
      * Export all metrics recorded in this instance to the underlying {@code perWorkerMetrics}
      * containers. This function will only report metrics once per instance. Subsequent calls to
@@ -223,12 +233,7 @@ public void updateStreamingInsertsMetrics(@Nullable TableReference tableRef) {
             .inc(successfulRowsCount().longValue());
       }
 
-      Histogram latencyHistogram =
-          BigQuerySinkMetrics.createRPCLatencyHistogram(
-              BigQuerySinkMetrics.RpcMethod.STREAMING_INSERTS);
-      for (Duration latency : rpcLatencies()) {
-        latencyHistogram.update(latency.toMillis());
-      }
+      recordRpcLatencyMetrics();
     }
   }
 }

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

@@ -39,7 +39,7 @@ public void testNoOpStreamingInsertsMetrics() throws Exception {
 
     StreamingInsertsMetrics results =
         StreamingInsertsMetrics.NoOpStreamingInsertsMetrics.getInstance();
-    results.updateRetriedRowsWithStatus("INTERNAL", 0);
+    results.updateRetriedRowsWithStatus("INTERNAL", 10);
     Instant t1 = Instant.now();
     results.updateSuccessfulRpcMetrics(t1, t1.plus(Duration.ofMillis(10)));
     TableReference ref = new TableReference().setTableId("t").setDatasetId("d");