(beam) branch master updated: [Dataflow] Fix thread safety of HotKey logger (#38816)

[scwhittle]

This is an automated email from the ASF dual-hosted git repository.

scwhittle pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new aaf677b16c3 [Dataflow] Fix thread safety of HotKey logger (#38816)
aaf677b16c3 is described below

commit aaf677b16c361c230064322949e236a17c531d34
Author: Arun Pandian <pandi...@google.com>
AuthorDate: Fri Jun 5 00:45:49 2026 -0700

    [Dataflow] Fix thread safety of HotKey logger (#38816)
---
 .../org/apache/beam/runners/dataflow/worker/HotKeyLogger.java  | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git 
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/HotKeyLogger.java
 
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/HotKeyLogger.java
index 00d93890d9a..c449f9d6c82 100644
--- 
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/HotKeyLogger.java
+++ 
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/HotKeyLogger.java
@@ -18,6 +18,7 @@
 package org.apache.beam.runners.dataflow.worker;
 
 import com.google.api.client.util.Clock;
+import javax.annotation.concurrent.GuardedBy;
 import org.apache.beam.runners.dataflow.util.TimeUtil;
 import org.joda.time.Duration;
 import org.slf4j.Logger;
@@ -33,6 +34,7 @@ public class HotKeyLogger {
    * The previous time the HotKeyDetection was logged. This is used to 
throttle logging to every 5
    * minutes.
    */
+  @GuardedBy("this")
   private long prevHotKeyDetectionLogMs = 0;
 
   /** Throttles logging the detection to every loggingPeriod */
@@ -83,10 +85,12 @@ public class HotKeyLogger {
   protected boolean isThrottled() {
     // Throttle logging the HotKeyDetection to every 5 minutes.
     long nowMs = clock.currentTimeMillis();
-    if (nowMs - prevHotKeyDetectionLogMs < loggingPeriod.getMillis()) {
-      return true;
+    synchronized (this) {
+      if (nowMs - prevHotKeyDetectionLogMs < loggingPeriod.getMillis()) {
+        return true;
+      }
+      prevHotKeyDetectionLogMs = nowMs;
     }
-    prevHotKeyDetectionLogMs = nowMs;
     return false;
   }
 }