Fix continuation of OrFinally trigger

Since the continuation is specified to fire as-soon-as-reasonable, the simple translation was firing the or-finally too early.

----Release Notes----
[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=107423899

sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/OrFinallyTrigger.java

@@ -77,8 +77,11 @@ public Instant getWatermarkThatGuaranteesFiring(W window) {
 
   @Override
   public Trigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
-    return new OrFinallyTrigger<W>(
-        continuationTriggers.get(ACTUAL),
-        (Trigger.OnceTrigger<W>) continuationTriggers.get(UNTIL));
+    // Use OrFinallyTrigger instead of AfterFirst because the continuation of ACTUAL
+    // may not be a OnceTrigger.
+    return Repeatedly.forever(
+        new OrFinallyTrigger<W>(
+            continuationTriggers.get(ACTUAL),
+            (Trigger.OnceTrigger<W>) continuationTriggers.get(UNTIL)));
   }
 }

sdk/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/OrFinallyTriggerTest.java

@@ -321,10 +321,12 @@ public void testContinuation() throws Exception {
     Trigger<IntervalWindow> aOrFinallyB = triggerA.orFinally(triggerB);
     Trigger<IntervalWindow> bOrFinallyA = triggerB.orFinally(triggerA);
     assertEquals(
-        triggerA.getContinuationTrigger().orFinally(triggerB.getContinuationTrigger()),
+        Repeatedly.forever(
+            triggerA.getContinuationTrigger().orFinally(triggerB.getContinuationTrigger())),
         aOrFinallyB.getContinuationTrigger());
     assertEquals(
-        triggerB.getContinuationTrigger().orFinally(triggerA.getContinuationTrigger()),
+        Repeatedly.forever(
+            triggerB.getContinuationTrigger().orFinally(triggerA.getContinuationTrigger())),
         bOrFinallyA.getContinuationTrigger());
   }
 }