Stop emitting upon truncate in Java PeriodicSequence (apache#25716)

akvelon · Mar 10, 2023 · b6db721 · b6db721
1 parent cfd91ea
commit b6db721
Showing 1 changed file with 7 additions and 1 deletion.
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/PeriodicSequence.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/PeriodicSequence.java
@@ -194,6 +194,12 @@ public WatermarkEstimator<Instant> newWatermarkEstimator(
       return new WatermarkEstimators.Manual(state);
     }
 
+    @TruncateRestriction
+    public RestrictionTracker.TruncateResult<OffsetRange> truncate() {
+      // stop emitting immediately upon drain
+      return null;
+    }
+
     @ProcessElement
     public ProcessContinuation processElement(
         @Element SequenceDefinition srcElement,
@@ -207,7 +213,7 @@ public ProcessContinuation processElement(
 
       boolean claimSuccess = true;
 
-      estimator.setWatermark(Instant.ofEpochMilli(restriction.getFrom()));
+      estimator.setWatermark(Instant.ofEpochMilli(nextOutput));
 
       while (claimSuccess && Instant.ofEpochMilli(nextOutput).isBeforeNow()) {
         claimSuccess = restrictionTracker.tryClaim(nextOutput);