Skip to content

[Dataflow] Fix thread safety of HotKey logger #38816

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
scwhittle merged 2 commits into from
Jun 5, 2026
Merged

[Dataflow] Fix thread safety of HotKey logger #38816

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7e0f07b
Fix thread safety of HotKey logger
arunpandianp Jun 4, 2026
d53696f
spotless
arunpandianp Jun 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions ...aflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/HotKeyLogger.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.beam.runners.dataflow.worker;

import com.google.api.client.util.Clock;
import javax.annotation.concurrent.GuardedBy;

@gemini-code-assist gemini-code-assist Bot Jun 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Since we can use AtomicLong to achieve lock-free thread safety, the GuardedBy annotation is no longer needed. We can replace it with an import for AtomicLong.

Suggested change
import javax.annotation.concurrent.GuardedBy;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.beam.runners.dataflow.util.TimeUtil;
import org.joda.time.Duration;
import org.slf4j.Logger;
Expand All @@ -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;
Comment on lines +37 to 38

@gemini-code-assist gemini-code-assist Bot Jun 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Using synchronized(this) on a public/protected class is generally discouraged as it exposes the lock to external callers, which can lead to deadlocks or unexpected contention. Using AtomicLong provides a lock-free, thread-safe alternative.

Suggested change
@GuardedBy("this")
private long prevHotKeyDetectionLogMs = 0;
private final AtomicLong prevHotKeyDetectionLogMs = new AtomicLong(0);


/** Throttles logging the detection to every loggingPeriod */
Expand Down Expand Up @@ -83,10 +85,12 @@ public void logHotKeyDetection(String userStepName, Duration hotKeyAge, Object h
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;
}
}
Loading