Add support for global sequence processing to the "ordered" extension in Java SDK

CHANGES.md

@@ -68,6 +68,7 @@
 
 * Dataflow worker can install packages from Google Artifact Registry Python repositories (Python) ([#32123](https://github.com/apache/beam/issues/32123)).
 * Added support for Zstd codec in SerializableAvroCodecFactory (Java) ([#32349](https://github.com/apache/beam/issues/32349))
+* Added support for processing events which use a global sequence to "ordered" extension (Java)
 * X feature added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 
 ## Breaking Changes

sdks/java/extensions/ordered/build.gradle

@@ -28,6 +28,12 @@ dependencies {
     implementation library.java.vendored_guava_32_1_2_jre
     testImplementation library.java.junit
     testImplementation library.java.hamcrest
+    testImplementation library.java.slf4j_jdk14
     testImplementation project(path: ':sdks:java:core')
+    testImplementation 'junit:junit:4.13.1'
+    testImplementation project(path: ':runners:google-cloud-dataflow-java')
     testRuntimeOnly project(path: ":runners:direct-java", configuration: "shadow")
+    testImplementation project(path: ":runners:google-cloud-dataflow-java")
+    testImplementation project(path: ":sdks:java:extensions:google-cloud-platform-core")
+    testImplementation project(path: ":sdks:java:io:google-cloud-platform")
 }

sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/ContiguousSequenceRange.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.extensions.ordered;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.InstantCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.checkerframework.checker.initialization.qual.Initialized;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.UnknownKeyFor;
+import org.joda.time.Instant;
+
+/** A range of contiguous event sequences and the latest timestamp of the events in the range. */
+@AutoValue
+public abstract class ContiguousSequenceRange {
+  public static final ContiguousSequenceRange EMPTY =
+      ContiguousSequenceRange.of(
+          Long.MIN_VALUE, Long.MIN_VALUE, Instant.ofEpochMilli(Long.MIN_VALUE));
+
+  /** @return inclusive starting sequence */
+  public abstract long getStart();
+
+  /** @return inclusive end sequence */
+  public abstract long getEnd();
+
+  /** @return latest timestamp of all events in the range */
+  public abstract Instant getTimestamp();
+
+  public static ContiguousSequenceRange of(long start, long end, Instant timestamp) {
+    return new AutoValue_ContiguousSequenceRange(start, end, timestamp);
+  }
+
+  static class CompletedSequenceRangeCoder extends CustomCoder<ContiguousSequenceRange> {
+
+    private static final CompletedSequenceRangeCoder INSTANCE = new CompletedSequenceRangeCoder();
+
+    static CompletedSequenceRangeCoder of() {
+      return INSTANCE;
+    }
+
+    private CompletedSequenceRangeCoder() {}
+
+    @Override
+    public void encode(
+        ContiguousSequenceRange value, @UnknownKeyFor @NonNull @Initialized OutputStream outStream)
+        throws @UnknownKeyFor @NonNull @Initialized CoderException, @UnknownKeyFor @NonNull
+            @Initialized IOException {
+      VarLongCoder.of().encode(value.getStart(), outStream);
+      VarLongCoder.of().encode(value.getEnd(), outStream);
+      InstantCoder.of().encode(value.getTimestamp(), outStream);
+    }
+
+    @Override
+    public ContiguousSequenceRange decode(@UnknownKeyFor @NonNull @Initialized InputStream inStream)
+        throws @UnknownKeyFor @NonNull @Initialized CoderException, @UnknownKeyFor @NonNull
+            @Initialized IOException {
+      long start = VarLongCoder.of().decode(inStream);
+      long end = VarLongCoder.of().decode(inStream);
+      Instant timestamp = InstantCoder.of().decode(inStream);
+      return ContiguousSequenceRange.of(start, end, timestamp);
+    }
+  }
+}

sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/EventExaminer.java

@@ -31,7 +31,8 @@ public interface EventExaminer<EventT, StateT extends MutableState<EventT, ?>>
     extends Serializable {
 
   /**
-   * Is this event the first expected event for the given key and window?
+   * Is this event the first expected event for the given key and window if the per key sequence is
+   * used? In case of global sequence it determines the first global sequence event.
    *
    * @param sequenceNumber the sequence number of the event as defined by the key of the input
    *     PCollection to {@link OrderedEventProcessor}
@@ -41,8 +42,8 @@ public interface EventExaminer<EventT, StateT extends MutableState<EventT, ?>>
   boolean isInitialEvent(long sequenceNumber, EventT event);
 
   /**
-   * If the event was the first event in the sequence, create the state to hold the required data
-   * needed for processing. This data will be persisted.
+   * If the event was the first event for a given key, create the state to hold the required data
+   * needed for processing. This data will be persisted in a Beam state.
    *
    * @param event the first event in the sequence.
    * @return the state to persist.
@@ -53,6 +54,8 @@ public interface EventExaminer<EventT, StateT extends MutableState<EventT, ?>>
   /**
    * Is this event the last expected event for a given key and window?
    *
+   * <p>Note, this method is not used yet with global sequences.
+   *
    * @param sequenceNumber of the event
    * @param event being processed
    * @return true if the last event. There are cases where it's impossible to know whether it's the

sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/GlobalSequenceTracker.java

@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.extensions.ordered;
+
+import org.apache.beam.sdk.extensions.ordered.ContiguousSequenceRange.CompletedSequenceRangeCoder;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.windowing.AfterFirst;
+import org.apache.beam.sdk.transforms.windowing.AfterPane;
+import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
+import org.apache.beam.sdk.transforms.windowing.Repeatedly;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.transforms.windowing.WindowFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.TimestampedValue;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Duration;
+
+/**
+ * PTransform to produce the side input of the maximum contiguous range of sequence numbers.
+ *
+ * @param <EventKeyT> type of event key
+ * @param <EventT> type of event
+ * @param <ResultT> type of processing result
+ * @param <StateT> type of state
+ */
+class GlobalSequenceTracker<
+        EventKeyT, EventT, ResultT, StateT extends MutableState<EventT, ResultT>>
+    extends PTransform<
+        PCollection<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>>,
+        PCollectionView<ContiguousSequenceRange>> {
+
+  private final Combine.GloballyAsSingletonView<
+          TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+      sideInputProducer;
+  private final @Nullable Duration frequencyOfGeneration;
+
+  public GlobalSequenceTracker(
+      Combine.GloballyAsSingletonView<
+              TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+          sideInputProducer) {
+    this.sideInputProducer = sideInputProducer;
+    this.frequencyOfGeneration = null;
+  }
+
+  public GlobalSequenceTracker(
+      Combine.GloballyAsSingletonView<
+              TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+          sideInputProducer,
+      Duration globalSequenceGenerationFrequency) {
+    this.sideInputProducer = sideInputProducer;
+    this.frequencyOfGeneration = globalSequenceGenerationFrequency;
+  }
+
+  @Override
+  public PCollectionView<ContiguousSequenceRange> expand(
+      PCollection<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>> input) {
+    input
+        .getPipeline()
+        .getCoderRegistry()
+        .registerCoderForClass(ContiguousSequenceRange.class, CompletedSequenceRangeCoder.of());
+
+    if (frequencyOfGeneration != null) {
+      // This branch will only be executed in case of streaming pipelines.
+      // For batch pipelines the side input should only be computed once.
+      input =
+          input.apply(
+              "Triggering Setup",
+              // Reproduce the windowing of the input PCollection, but change the triggering
+              // in order to create a slowing changing side input
+              Window.<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>>into(
+                      (WindowFn<? super TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ?>)
+                          input.getWindowingStrategy().getWindowFn())
+                  .accumulatingFiredPanes()
+                  // TODO: verify that we don't need to have the lateness parameterized
+                  .withAllowedLateness(Duration.ZERO)
+                  .triggering(
+                      Repeatedly.forever(
+                          AfterFirst.of(
+                              AfterPane.elementCountAtLeast(1),
+                              AfterProcessingTime.pastFirstElementInPane()
+                                  .plusDelayOf(frequencyOfGeneration)))));
+    }
+    return input.apply("Create Side Input", sideInputProducer);
+  }
+}