Address a data race on numMessages in DirectStreamObserver.java

apache · Jul 2, 2024 · 42a9d3b · 42a9d3b
1 parent 8a4a250
commit 42a9d3b
Showing 1 changed file with 7 additions and 4 deletions.
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/fn/stream/DirectStreamObserver.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/fn/stream/DirectStreamObserver.java
@@ -53,17 +53,20 @@ public final class DirectStreamObserver<T> implements StreamObserver<T> {
   private final int maxMessagesBeforeCheck;
 
   private final Object lock = new Object();
-  private int numMessages = -1;
+  private int numMessages;
 
   public DirectStreamObserver(Phaser phaser, CallStreamObserver<T> outboundObserver) {
     this(phaser, outboundObserver, DEFAULT_MAX_MESSAGES_BEFORE_CHECK);
   }
 
   DirectStreamObserver(
       Phaser phaser, CallStreamObserver<T> outboundObserver, int maxMessagesBeforeCheck) {
-    this.phaser = phaser;
-    this.outboundObserver = outboundObserver;
-    this.maxMessagesBeforeCheck = maxMessagesBeforeCheck;
+    synchronized (this.lock) {
+      this.phaser = phaser;
+      this.outboundObserver = outboundObserver;
+      this.maxMessagesBeforeCheck = maxMessagesBeforeCheck;
+      this.numMessages = -1;
+    }
   }
 
   @Override