diff --git a/samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java b/samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java
index 8d4ae213d7..3d32be3197 100644
--- a/samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java
+++ b/samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java
@@ -191,7 +191,7 @@ public final CompletionStage<Void> onMessageAsync(M message, MessageCollector co
             .toArray(CompletableFuture[]::new));
       });
 
-    result.thenAccept(x -> {
+    return result.thenAccept(x -> {
         WatermarkFunction watermarkFn = getOperatorSpec().getWatermarkFn();
         if (watermarkFn != null) {
           // check whether there is new watermark emitted from the user function
@@ -199,8 +199,6 @@ public final CompletionStage<Void> onMessageAsync(M message, MessageCollector co
           propagateWatermark(outputWm, collector, coordinator);
         }
       });
-
-    return result;
   }
 
   /**
@@ -415,7 +413,7 @@ private CompletionStage<Void> onWatermark(long watermark, MessageCollector colle
                 .toArray(CompletableFuture[]::new));
       }
 
-      watermarkFuture.thenCompose(res -> propagateWatermark(outputWm, collector, coordinator));
+      watermarkFuture = watermarkFuture.thenCompose(res -> propagateWatermark(outputWm, collector, coordinator));
     }
 
     return watermarkFuture;