[FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API #26328

fredia · 2025-03-20T11:52:25Z

What is the purpose of the change

This PR introduces AsyncKeyedCoProcessOperator

Brief change log

Add AsyncKeyedCoProcessOperator
Add AsyncKeyedCoProcessOperatorWithWatermarkDelay
Add DeclaringAsyncKeyedCoProcessFunction
Move AsyncIntervalJoinOperator to o.a.f.runtime.asyncprocessing.operators.co package.

Verifying this change

Add AsyncKeyedCoProcessOperatorTest

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2025-03-20T12:13:36Z

CI report:

9e10849 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

davidradl · 2025-03-20T16:10:29Z

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java

+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.


nit: one. -> stream.

davidradl · 2025-03-20T16:15:42Z

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java

+ * <p>The function will be called for every element in the input streams and can produce zero or
+ * more output elements. Contrary to the {@link CoFlatMapFunction}, this function can also query the
+ * time (both event and processing) and set timers, through the provided {@link Context}. When
+ * reacting to the firing of set timers the function can emit yet more elements.


nit. in the previous sentence I assume that this function can set timers. In this sentence set timers is not so clear. Maybe "When reacting to the firing of timers," or "When reacting to the firing of timers that have been set,"

davidradl · 2025-03-20T16:16:42Z

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java

+ * time (both event and processing) and set timers, through the provided {@link Context}. When
+ * reacting to the firing of set timers the function can emit yet more elements.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that


nits:
use-case -> use case
would be -> is

davidradl · 2025-03-20T16:18:00Z

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java

+ * change over time ({@code stream A}) to the elements contained in another stream (stream {@code
+ * B}). The rules contained in {@code stream A} can be stored in the state and wait for new elements
+ * to arrive on {@code stream B}. Upon reception of a new element on {@code stream B}, the function
+ * can now apply the previously stored rules to the element and directly emit a result, and/or


nits:
remove the word now
remove the word directly

davidradl · 2025-03-20T16:20:02Z

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java

+     * @throws Exception The function may throw exceptions which cause the streaming program to fail
+     *     and go into recovery.
+     */
+    public void processElement1(IN1 value, Context ctx, Collector<OUT> out) throws Exception {


nit: processElement1 -> processInputStream1Element or the like

We can't rename it, because processElement1 is inherited from KeyedCoProcessFunction. I added override here.

davidradl · 2025-03-20T16:20:36Z

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java

+     *     and go into recovery.
+     */
+    public void processElement1(IN1 value, Context ctx, Collector<OUT> out) throws Exception {
+        throw new IllegalAccessException("This method is replaced by declareProcess1.");


what does this exception message mean?

This method is inherited from KeyedCoProcessFunction, but we want to use declareProcess1 to replace it.
This method should not be called as expected, so an exception thrown to prevent misuse.

davidradl · 2025-03-20T16:22:06Z

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java

+     * <p>This function can output zero or more elements using the {@link Collector} parameter and
+     * also update internal state or set timers using the {@link Context} parameter.
+     *
+     * @param value The stream element


why is this parameter called value? Could it be something more specific like streamElement

Same as the question above, this method is inherited from KeyedCoProcessFunction, the naming is consistent with KeyedCoProcessFunction. I prefer keeping it.

davidradl · 2025-03-20T16:23:57Z

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java

+     * also update internal state or set timers using the {@link Context} parameter.
+     *
+     * @param value The stream element
+     * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the


nit: maybe a list would be easier to read in the javadoc. A {@link Context} that allows querying of

...

When you say querying the time is this event or processing?

davidradl · 2025-03-20T16:25:05Z

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java

+     *     timers and querying the time. The context is only valid during the invocation of this
+     *     method, do not store it.
+     * @param out The collector to emit resulting elements to
+     * @throws Exception The function may throw exceptions which cause the streaming program to fail


can it throw a retriable exception?

What is a retriable exception?

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java

davidradl · 2025-03-20T16:28:15Z

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java

+     * @throws Exception The function may throw exceptions which cause the streaming program to fail
+     *     and go into recovery.
+     */
+    public void processElement1(IN1 value, Context ctx, Collector<OUT> out) throws Exception {


we should label all the method as @public or @experimental or the like.

I don't think this is necessary because it is inherited from KeyedCoProcessFunction . I add override here.

davidradl · 2025-03-20T16:31:17Z

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java

+            throws DeclarationException;
+
+    /**
+     * Declare a procedure which is called when a timer set using {@link TimerService} fires.


the words say it is declaring a procedure but the output parameter says it is a processor. I would expect these words to be the same. The method name should align with what we are declaring. maybe declareStream2Processor

davidradl · 2025-03-20T16:33:08Z

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java

+     *     of this method, do not store it.
+     * @param out The processor for processing timestamps.
+     */
+    public ThrowingConsumer<Long, Exception> declareOnTimer(


nit: declareOnTimer -> declareOnTimerProcessor

davidradl · 2025-03-20T16:34:20Z

.../java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperator.java

+    private static final long serialVersionUID = 1L;
+
+    // Shared timestamp variable for collector, context and onTimerContext.
+    private transient DeclaredVariable<Long> sharedTimestamp;


what do we mean by the sharedTimestamp? Is this event time?

It is used to store one Timestamp. It can be event time or processing time.

davidradl · 2025-03-20T16:37:11Z

.../java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperator.java

+        collector = new TimestampedCollectorWithDeclaredVariable<>(output, sharedTimestamp);
+
+        InternalTimerService<VoidNamespace> internalTimerService =
+                getInternalTimerService("user-timers", VoidNamespaceSerializer.INSTANCE, this);


nit: maybe change to a constant and defined in the same place as other named InternalTimerServices. I am not sure what does user mean this context.

The naming "user-timers" in AsyncKeyedCoProcessOperator is consistent with KeyedCoProcessOperator#internalTimerService. It represents a timer-service created by operator.

davidradl · 2025-03-20T16:40:33Z

.../java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperator.java

+        if (userFunction instanceof DeclaringAsyncKeyedCoProcessFunction) {
+            DeclaringAsyncKeyedCoProcessFunction declaringFunction =
+                    (DeclaringAsyncKeyedCoProcessFunction) userFunction;
+            declaringFunction.declareVariables(declarationContext);


I am curious, is there a description of what these declarations are and why we are declaring things for async?
I assume we are describing the variables, functions and timers that will be realised when they run asynchronously.

You can find the description about declarations at https://cwiki.apache.org/confluence/display/FLINK/FLIP-455%3A+Declare+async+state+processing+and+checkpoint+the+in-flight+requests

davidradl · 2025-03-20T16:42:01Z

...link/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperatorWithWatermarkDelay.java

+import java.util.function.Consumer;
+
+/** A {@link KeyedCoProcessOperator} that supports holding back watermarks with a static delay. */
+public class AsyncKeyedCoProcessOperatorWithWatermarkDelay<K, IN1, IN2, OUT>


I cant see this class called anywhere. Even from test code.

You're right, It will be called at StreamExecIntervalJoin in the future after we implement async version rowJoinFunc.

…ate API

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java

.../java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperator.java

...link/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperatorWithWatermarkDelay.java

Zakelly · 2025-03-31T06:54:43Z

...ava/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java

@@ -387,8 +387,11 @@ public Watermark preProcessWatermark(Watermark watermark) throws Exception {
     * perform async state here. Only some synchronous logic is suggested.
     *
     * @param watermark the advanced watermark.
+     * @return the watermark that should be emitted to downstream.


We should allow returning null if there is no need emitting any watermark downstream.

And please update the javadoc of this method explaining this.

...ache/flink/table/runtime/operators/join/interval/asyncprocess/AsyncProcTimeIntervalJoin.java

Zakelly · 2025-03-31T10:33:47Z

...g/apache/flink/table/runtime/operators/join/interval/asyncprocess/AsyncTimeIntervalJoin.java

+                                rightCache.asyncRemove(key);
+                            }
+                        });
+        return emitResult.get();


We better not blocking get() here. It will block the task thread as well as those callbacks for async state requests.

👍I changed the return value to StateFuture<Boolean>.

...c/main/java/org/apache/flink/table/runtime/operators/join/interval/SyncTimeIntervalJoin.java

Zakelly · 2025-04-01T08:15:05Z

...g/apache/flink/table/runtime/operators/join/interval/asyncprocess/AsyncTimeIntervalJoin.java

+        joinFunction.setJoinKey(ctx.getCurrentKey());
+        joinCollector.setInnerCollector(out);


the internal state of joinFunction and joinCollector should be maintained by DeclaredVariable since it belongs to the context for async processing.

👍 Good suggestion, I rewrote AsyncTimeIntervalJoin with declareChain.

...g/apache/flink/table/runtime/operators/join/interval/asyncprocess/AsyncTimeIntervalJoin.java

Zakelly · 2025-04-02T09:39:37Z

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java

+        throw new IllegalAccessException("This method is replaced by declareProcess1.");
+    }
+
+    /** Override this method or use {@link #declareProcess2} instead. */


Actually we dont recommend to use this method. How about rewrite this javadoc?

Suggested change

/** Override this method or use {@link #declareProcess2} instead. */

/** Override and finalize this method. Please use {@link #declareProcess2} instead. */

Zakelly · 2025-04-02T09:42:32Z

...ava/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java

     */
-    public void postProcessWatermark(Watermark watermark) throws Exception {}
+    public Watermark postProcessWatermark(Watermark watermark) throws Exception {


How about change the javadoc of this method to :

A hook that will be invoked after finishing advancing the watermark and right before the watermark being emitting downstream. Here is a chance for customization of the emitting watermark. ....(and following description)

Zakelly · 2025-04-02T09:44:32Z

...in/java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncIntervalJoinOperator.java

@@ -417,12 +419,12 @@ public <X> void output(OutputTag<X> outputTag, X value) {
    }

    @VisibleForTesting
-    MapState<Long, List<IntervalJoinOperator.BufferEntry<T1>>> getLeftBuffer() {
+    public MapState<Long, List<IntervalJoinOperator.BufferEntry<T1>>> getLeftBuffer() {


Seems unnecessary?

Because AsyncIntervalJoinOperator is moved into asyncprocessing package, and AsyncIntervalJoinOperatorTest still needs these methods.

flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/ConnectedStreams.java

.../main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java

flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/ConnectedStreams.java

Zakelly

Thanks for the update! LGTM. Would you please update the description of this PR and remove the unrelated change?

flinkbot added the component=<none> label Mar 20, 2025

davidradl reviewed Mar 20, 2025

View reviewed changes

...org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java Outdated Show resolved Hide resolved

davidradl reviewed Mar 20, 2025

View reviewed changes

[FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async st…

ac618f8

…ate API

fredia force-pushed the twoinds branch 2 times, most recently from 3640756 to 7ddeb5d Compare March 31, 2025 06:15

Zakelly reviewed Mar 31, 2025

View reviewed changes

Zakelly requested changes Mar 31, 2025

View reviewed changes

Zakelly reviewed Mar 31, 2025

View reviewed changes

...c/main/java/org/apache/flink/table/runtime/operators/join/interval/SyncTimeIntervalJoin.java Outdated Show resolved Hide resolved

fredia force-pushed the twoinds branch 2 times, most recently from 3351542 to 1c0dca7 Compare April 1, 2025 06:55

Zakelly reviewed Apr 1, 2025

View reviewed changes

...g/apache/flink/table/runtime/operators/join/interval/asyncprocess/AsyncTimeIntervalJoin.java Outdated Show resolved Hide resolved

...g/apache/flink/table/runtime/operators/join/interval/asyncprocess/AsyncTimeIntervalJoin.java Outdated Show resolved Hide resolved

fredia force-pushed the twoinds branch from 1c0dca7 to 9d8f866 Compare April 1, 2025 10:18

fredia force-pushed the twoinds branch from 9d8f866 to d9ba20b Compare April 2, 2025 08:58

address comments

05c3eb0

fredia force-pushed the twoinds branch from d9ba20b to 05c3eb0 Compare April 2, 2025 09:37

Zakelly reviewed Apr 2, 2025

View reviewed changes

address comments

3c16c28

Zakelly reviewed Apr 2, 2025

View reviewed changes

flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/ConnectedStreams.java Outdated Show resolved Hide resolved

address comment

9e10849

Zakelly approved these changes Apr 2, 2025

View reviewed changes

		joinFunction.setJoinKey(ctx.getCurrentKey());
		joinCollector.setInnerCollector(out);

	/** Override this method or use {@link #declareProcess2} instead. */
	/** Override and finalize this method. Please use {@link #declareProcess2} instead. */

[FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API #26328

Are you sure you want to change the base?

[FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API #26328

Conversation

fredia commented Mar 20, 2025

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Mar 20, 2025 • edited Loading

CI report:

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fredia Mar 26, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davidradl Mar 20, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davidradl Mar 20, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Zakelly left a comment

Choose a reason for hiding this comment

flinkbot commented Mar 20, 2025 •

edited

Loading

fredia Mar 26, 2025 •

edited

Loading

davidradl Mar 20, 2025 •

edited

Loading

davidradl Mar 20, 2025 •

edited

Loading