Address comments and spotless fixes

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

@@ -25,14 +25,34 @@
 public class DelegatingCounter implements Metric, Counter, Serializable {
   private final MetricName name;
   private final boolean processWideContainer;
+  private final boolean perWorkerCounter;
 
+  /**
+   * Create a {@code DelegatingCounter} with {@code perWorkerCounter} and {@code processWideContainer} set to false.
+   * @param name Metric name for this metric.
+   */
   public DelegatingCounter(MetricName name) {
-    this(name, false);
+    this(name, false, false);
   }
 
+  /**
+   * Create a {@code DelegatingCounter} with {@code perWorkerCounter} set to false.
+   * @param name Metric name for this metric.
+   * @param processWideContainer Whether this Counter is stored in the ProcessWide container or the current thread's container.
+   */
   public DelegatingCounter(MetricName name, boolean processWideContainer) {
+    this(name, processWideContainer, false);    
+  }
+
+  /**
+   * @param name Metric name for this metric.
+   * @param processWideContainer Whether this Counter is stored in the ProcessWide container or the current thread's container.
+   * @param perWorkerCounter Whether this Counter refers to a perWorker metric or not.
+   */
+  public DelegatingCounter(MetricName name, boolean processWideContainer, boolean perWorkerCounter) {
     this.name = name;
     this.processWideContainer = processWideContainer;
+    this.perWorkerCounter = perWorkerCounter;
   }
 
   /** Increment the counter. */
@@ -48,7 +68,12 @@ public void inc(long n) {
         this.processWideContainer
             ? MetricsEnvironment.getProcessWideContainer()
             : MetricsEnvironment.getCurrentContainer();
-    if (container != null) {
+    if (container == null) {
+      return;
+    }
+    if (perWorkerCounter) {
+      container.getPerWorkerCounter(name).inc(n);
+    } else {
       container.getCounter(name).inc(n);
     }
   }

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

@@ -27,21 +27,46 @@ public class DelegatingHistogram implements Metric, Histogram, Serializable {
   private final MetricName name;
   private final HistogramData.BucketType bucketType;
   private final boolean processWideContainer;
+  private final boolean perWorkerHistogram;
 
+  /**
+   * Create a {@code DelegatingHistogram} with {@code perWorkerHistogram} set to false.
+   * @param name Metric name for this metric.
+   * @param bucketType Histogram bucketing strategy.
+   * @param processWideContainer Whether this Counter is stored in the ProcessWide container or the current thread's container.
+   */
   public DelegatingHistogram(
       MetricName name, HistogramData.BucketType bucketType, boolean processWideContainer) {
-    this.name = name;
-    this.bucketType = bucketType;
-    this.processWideContainer = processWideContainer;
+    this(name, bucketType, processWideContainer, false);
   }
 
+
+  /**
+   * @param name Metric name for this metric.
+   * @param bucketType Histogram bucketing strategy.
+   * @param processWideContainer Whether this Counter is stored in the ProcessWide container or the current thread's container.
+   * @param perWorkerHistogram Whether this Histogram refers to a perWorker metric or not.
+   */
+  public DelegatingHistogram(
+    MetricName name, HistogramData.BucketType bucketType, boolean processWideContainer, boolean perWorkerHistogram) {
+      this.name = name;
+      this.bucketType = bucketType;
+      this.processWideContainer = processWideContainer;
+      this.perWorkerHistogram = perWorkerHistogram;
+    }
+
   @Override
   public void update(double value) {
     MetricsContainer container =
         processWideContainer
             ? MetricsEnvironment.getProcessWideContainer()
             : MetricsEnvironment.getCurrentContainer();
-    if (container != null) {
+    if (container == null) {
+      return;
+    }
+    if (perWorkerHistogram) {
+      container.getPerWorkerHistogram(name, bucketType).update(value);
+    } else {
       container.getHistogram(name, bucketType).update(value);
     }
   }