[BEAM-8543] Dataflow streaming timers are not strictly time ordered when set earlier mid-bundle

CHANGES.md

@@ -101,6 +101,7 @@
 * Upgrade Sphinx to 3.0.3 for building PyDoc.
 * Added a PTransform for image annotation using Google Cloud AI image processing service
 ([BEAM-9646](https://issues.apache.org/jira/browse/BEAM-9646))
+* Dataflow streaming timers are not strictly time ordered when set earlier mid-bundle ([BEAM-8543](https://issues.apache.org/jira/browse/BEAM-8543)).
 
 ## Breaking Changes
 

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java

@@ -24,10 +24,12 @@
 import com.google.api.services.dataflow.model.SideInputInfo;
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.PriorityQueue;
 import java.util.Set;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicLong;
@@ -522,6 +524,7 @@ public void start(
 
       this.cachedFiredTimers = null;
       this.cachedFiredUserTimers = null;
+      this.toBeFiredTimersOrdered.clear();
     }
 
     public void flushState() {
@@ -564,10 +567,13 @@ public <W extends BoundedWindow> TimerData getNextFiredTimer(Coder<W> windowCode
     // Lazily initialized
     private Iterator<TimerData> cachedFiredUserTimers = null;
 
+    private PriorityQueue<TimerData> toBeFiredTimersOrdered =
+        new PriorityQueue<>(Comparator.comparing(TimerData::getTimestamp));
+
     public <W extends BoundedWindow> TimerData getNextFiredUserTimer(Coder<W> windowCoder) {
       if (cachedFiredUserTimers == null) {
         cachedFiredUserTimers =
-            FluentIterable.<Timer>from(StreamingModeExecutionContext.this.getFiredTimers())
+            FluentIterable.from(StreamingModeExecutionContext.this.getFiredTimers())
                 .filter(
                     timer ->
                         WindmillTimerInternals.isUserTimer(timer)
@@ -577,12 +583,21 @@ public <W extends BoundedWindow> TimerData getNextFiredUserTimer(Coder<W> window
                         WindmillTimerInternals.windmillTimerToTimerData(
                             WindmillNamespacePrefix.USER_NAMESPACE_PREFIX, timer, windowCoder))
                 .iterator();
+
+        cachedFiredUserTimers.forEachRemaining(toBeFiredTimersOrdered::add);
+      }
+
+      Instant currentInputWatermark = userTimerInternals.currentInputWatermarkTime();
+      if (userTimerInternals.hasTimerBefore(currentInputWatermark)) {
+        while (!toBeFiredTimersOrdered.isEmpty()) {
+          userTimerInternals.setTimer(toBeFiredTimersOrdered.poll());
+        }
       }
 
-      if (!cachedFiredUserTimers.hasNext()) {
+      if (toBeFiredTimersOrdered.isEmpty()) {
         return null;
       }
-      TimerData nextTimer = cachedFiredUserTimers.next();
+      TimerData nextTimer = toBeFiredTimersOrdered.poll();
       // User timers must be explicitly deleted when delivered, to release the implied hold
       userTimerInternals.deleteTimer(nextTimer);
       return nextTimer;

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillTimerInternals.java

@@ -227,6 +227,18 @@ public void persistTo(Windmill.WorkItemCommitRequest.Builder outputBuilder) {
     timers.clear();
   }
 
+  public boolean hasTimerBefore(Instant time) {
+    for (Cell<String, StateNamespace, Boolean> cell : timerStillPresent.cellSet()) {
+      TimerData timerData = timers.get(cell.getRowKey(), cell.getColumnKey());
+      if (cell.getValue()) {
+        if (timerData.getTimestamp().isBefore(time)) {
+          return true;
+        }
+      }
+    }
+    return false;
+  }
+
   private boolean needsWatermarkHold(TimerData timerData) {
     // If it is a user timer or a system timer with outputTimestamp different than timestamp
     return WindmillNamespacePrefix.USER_NAMESPACE_PREFIX.equals(prefix)

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java

@@ -139,6 +139,7 @@
 import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions;
 import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
 import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Iterables;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Iterators;
 import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
 import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Sets;
 import org.hamcrest.Matchers;
@@ -3938,10 +3939,11 @@ public void testEventTimeTimerOrdering() throws Exception {
       ValidatesRunner.class,
       UsesTimersInParDo.class,
       UsesStatefulParDo.class,
+      UsesUnboundedPCollections.class,
       UsesStrictTimerOrdering.class
     })
     public void testEventTimeTimerOrderingWithCreate() throws Exception {
-      final int numTestElements = 100;
+      final int numTestElements = 5;
       final Instant now = new Instant(1500000000000L);
 
       List<TimestampedValue<KV<String, String>>> elements = new ArrayList<>();
@@ -4009,6 +4011,7 @@ public void onTimer(
 
               List<TimestampedValue<String>> flush = new ArrayList<>();
               Instant flushTime = context.timestamp();
+              int bagSize = Iterators.size(bagState.read().iterator());
               for (TimestampedValue<String> val : bagState.read()) {
                 if (!val.getTimestamp().isAfter(flushTime)) {
                   flush.add(val);
@@ -4018,7 +4021,7 @@ public void onTimer(
               context.output(
                   Joiner.on(":").join(flush.stream().map(TimestampedValue::getValue).iterator()));
               Instant newMinStamp = flushTime.plus(1);
-              if (flush.size() < numTestElements) {
+              if (flush.size() < numTestElements && bagSize > 0) {
                 timer.set(newMinStamp);
               }
             }
@@ -4040,7 +4043,8 @@ public void onTimer(
             }
           };
 
-      PCollection<String> output = pipeline.apply(transform).apply(ParDo.of(fn));
+      PCollection<String> output =
+          pipeline.apply(transform).setIsBoundedInternal(IsBounded.UNBOUNDED).apply(ParDo.of(fn));
       List<String> expected =
           IntStream.rangeClosed(0, numTestElements)
               .mapToObj(expandFn(numTestElements))