another round of comments

apache · Jan 21, 2025 · 11137f9 · 11137f9
1 parent f02a194
commit 11137f9
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 19 deletions.
diff --git a/...rg/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverterTest.java b/...rg/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverterTest.java
@@ -83,7 +83,7 @@ public double getAccumulatedBucketSize(int endIndex) {
   }
 
   @Test
-  public void testConvert_successfulyConvertCounters() {
+  public void testConvert_successfullyConvertCounters() {
     String step = "testStepName";
     Map<MetricName, LockFreeHistogram.Snapshot> emptyHistograms = new HashMap<>();
     Map<MetricName, Long> counters = new HashMap<MetricName, Long>();
@@ -374,7 +374,7 @@ public void testConvert_convertCountersAndHistograms() {
   }
 
   @Test
-  public void testConvert_successfulyConvertGauges() {
+  public void testConvert_successfullyConvertGauges() {
     String step = "testStepName";
     Map<MetricName, LockFreeHistogram.Snapshot> emptyHistograms = new HashMap<>();
     Map<MetricName, Long> counters = new HashMap<MetricName, Long>();

diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java
@@ -19,9 +19,9 @@
 
 import com.google.auto.value.AutoValue;
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.beam.sdk.metrics.Gauge;
 import org.apache.beam.sdk.metrics.Histogram;
@@ -74,31 +74,29 @@ abstract class KafkaMetricsImpl implements KafkaMetrics {
 
     private static final Logger LOG = LoggerFactory.getLogger(KafkaMetricsImpl.class);
 
-    static ConcurrentHashMap<String, Histogram> latencyHistograms =
-        new ConcurrentHashMap<String, Histogram>();
+    private static final Map<String, Histogram> LATENCY_HISTOGRAMS =
+        new HashMap<String, Histogram>();
 
-    abstract ConcurrentHashMap<String, ConcurrentLinkedQueue<Duration>> perTopicRpcLatencies();
+    abstract HashMap<String, ArrayList<Duration>> perTopicRpcLatencies();
 
-    static ConcurrentHashMap<String, Gauge> backlogGauges = new ConcurrentHashMap<String, Gauge>();
-
-    abstract ConcurrentHashMap<String, Long> perTopicPartitionBacklogs();
+    abstract HashMap<String, Long> perTopicPartitionBacklogs();
 
     abstract AtomicBoolean isWritable();
 
     public static KafkaMetricsImpl create() {
       return new AutoValue_KafkaMetrics_KafkaMetricsImpl(
-          new ConcurrentHashMap<String, ConcurrentLinkedQueue<Duration>>(),
-          new ConcurrentHashMap<String, Long>(),
+          new HashMap<String, ArrayList<Duration>>(),
+          new HashMap<String, Long>(),
           new AtomicBoolean(true));
     }
 
     /** Record the rpc status and latency of a successful Kafka poll RPC call. */
     @Override
     public void updateSuccessfulRpcMetrics(String topic, Duration elapsedTime) {
       if (isWritable().get()) {
-        ConcurrentLinkedQueue<Duration> latencies = perTopicRpcLatencies().get(topic);
+        ArrayList<Duration> latencies = perTopicRpcLatencies().get(topic);
         if (latencies == null) {
-          latencies = new ConcurrentLinkedQueue<Duration>();
+          latencies = new ArrayList<Duration>();
           latencies.add(elapsedTime);
           perTopicRpcLatencies().putIfAbsent(topic, latencies);
         } else {
@@ -122,18 +120,18 @@ public void updateBacklogBytes(String topicName, int partitionId, long backlog)
 
     /** Record rpc latency histogram metrics for all recorded topics. */
     private void recordRpcLatencyMetrics() {
-      for (Map.Entry<String, ConcurrentLinkedQueue<Duration>> topicLatencies :
+      for (Map.Entry<String, ArrayList<Duration>> topicLatencies :
           perTopicRpcLatencies().entrySet()) {
         Histogram topicHistogram;
-        if (latencyHistograms.containsKey(topicLatencies.getKey())) {
-          topicHistogram = latencyHistograms.get(topicLatencies.getKey());
+        if (LATENCY_HISTOGRAMS.containsKey(topicLatencies.getKey())) {
+          topicHistogram = LATENCY_HISTOGRAMS.get(topicLatencies.getKey());
         } else {
           topicHistogram =
               KafkaSinkMetrics.createRPCLatencyHistogram(
                   KafkaSinkMetrics.RpcMethod.POLL, topicLatencies.getKey());
-          latencyHistograms.put(topicLatencies.getKey(), topicHistogram);
+          LATENCY_HISTOGRAMS.put(topicLatencies.getKey(), topicHistogram);
         }
-        // update all the latencies
+        // Update all the latencies
         for (Duration d : topicLatencies.getValue()) {
           Preconditions.checkArgumentNotNull(topicHistogram);
           topicHistogram.update(d.toMillis());