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 3d32be3197..528acc66f2 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,14 +191,14 @@ public final CompletionStage<Void> onMessageAsync(M message, MessageCollector co
             .toArray(CompletableFuture[]::new));
       });
 
-    return result.thenAccept(x -> {
-        WatermarkFunction watermarkFn = getOperatorSpec().getWatermarkFn();
-        if (watermarkFn != null) {
-          // check whether there is new watermark emitted from the user function
-          Long outputWm = watermarkFn.getOutputWatermark();
-          propagateWatermark(outputWm, collector, coordinator);
-        }
-      });
+    WatermarkFunction watermarkFn = getOperatorSpec().getWatermarkFn();
+    if (watermarkFn != null) {
+      // check whether there is new watermark emitted from the user function
+      Long outputWm = watermarkFn.getOutputWatermark();
+      return result.thenCompose(ignored -> propagateWatermark(outputWm, collector, coordinator));
+    }
+
+    return result;
   }
 
   /**