another round of comments

apache · Jan 24, 2025 · fc42673 · fc42673
1 parent f02a194
commit fc42673
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 11 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
@@ -74,12 +74,10 @@ abstract class KafkaMetricsImpl implements KafkaMetrics {
 
     private static final Logger LOG = LoggerFactory.getLogger(KafkaMetricsImpl.class);
 
-    static ConcurrentHashMap<String, Histogram> latencyHistograms =
+    private static final Map<String, Histogram> LATENCY_HISTOGRAMS =
         new ConcurrentHashMap<String, Histogram>();
 
-    abstract ConcurrentHashMap<String, ConcurrentLinkedQueue<Duration>> perTopicRpcLatencies();
-
-    static ConcurrentHashMap<String, Gauge> backlogGauges = new ConcurrentHashMap<String, Gauge>();
+    abstract ConcurrentHashMap<String, ConcurrentLinkedQueue<Duration>> perTopicRpcLatencies();;
 
     abstract ConcurrentHashMap<String, Long> perTopicPartitionBacklogs();
 
@@ -92,7 +90,14 @@ public static KafkaMetricsImpl create() {
           new AtomicBoolean(true));
     }
 
-    /** Record the rpc status and latency of a successful Kafka poll RPC call. */
+    /**
+     * Record the rpc status and latency of a successful Kafka poll RPC call.
+     *
+     * <p>TODO: It's possible that `isWritable().get()` is called before it's set to false in
+     * another thread, allowing an extraneous measurement to slip in, so perTopicRpcLatencies()
+     * isn't necessarily thread safe. One way to address this would be to add syncrhoized blocks to
+     * ensure that there is only one thread ever reading/modifying the perTopicRpcLatencies() map.
+     */
     @Override
     public void updateSuccessfulRpcMetrics(String topic, Duration elapsedTime) {
       if (isWritable().get()) {
@@ -125,15 +130,15 @@ private void recordRpcLatencyMetrics() {
       for (Map.Entry<String, ConcurrentLinkedQueue<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());