You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "fredia (via GitHub)" <gi...@apache.org> on 2025/03/20 11:52:27 UTC
[PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
fredia opened a new pull request, #26328:
URL: https://github.com/apache/flink/pull/26328
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## 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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006005988
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
Review Comment:
nits:
use-case -> use case
would be -> is
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2022106552
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * 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.");
+ }
+
+ /**
+ * This method is called for each element in the second of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * and go into recovery.
+ */
+ public void processElement2(IN2 value, Context ctx, Collector<OUT> out) throws Exception {
+ throw new IllegalAccessException("This method is replaced by declareProcess2.");
+ }
+
+ /**
+ * Called when a timer set using {@link TimerService} fires.
+ *
+ * @param timestamp The timestamp of the firing timer.
+ * @param ctx An {@link OnTimerContext} that allows querying the timestamp of the firing timer,
+ * querying the {@link TimeDomain} of the firing timer and getting a {@link TimerService}
+ * for registering timers and querying the time. The context is only valid during the
+ * invocation of this method, do not store it.
+ * @param out The collector for returning result values.
+ * @throws Exception This method may throw exceptions. Throwing an exception will cause the
+ * operation to fail and may trigger recovery.
+ */
+ public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception {}
+
+ /**
Review Comment:
@Zakelly `KeyedCoProcessFunction#Context` isn't removed, I added `AsyncKeyedCoProcessOperator#ContextImpl` which inherits from it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2020488907
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperatorWithWatermarkDelay.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.flink.runtime.asyncprocessing.operators.co;
+
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Serializable;
+import java.util.function.Consumer;
+
+/** A {@link KeyedCoProcessOperator} that supports holding back watermarks with a static delay. */
+public class AsyncKeyedCoProcessOperatorWithWatermarkDelay<K, IN1, IN2, OUT>
+ extends AsyncKeyedCoProcessOperator<K, IN1, IN2, OUT> {
+ private static final long serialVersionUID = 1L;
+
+ private final Consumer<Watermark> emitter;
Review Comment:
Please remote the emitter and related initialization code.
##########
flink-runtime/src/main/java/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.
Review Comment:
We should allow returning `null` if there is no need emitting any watermark downstream.
And please update the javadoc of this method explaining this.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * 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.");
+ }
+
+ /**
+ * This method is called for each element in the second of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * and go into recovery.
+ */
+ public void processElement2(IN2 value, Context ctx, Collector<OUT> out) throws Exception {
+ throw new IllegalAccessException("This method is replaced by declareProcess2.");
+ }
+
+ /**
+ * Called when a timer set using {@link TimerService} fires.
+ *
+ * @param timestamp The timestamp of the firing timer.
+ * @param ctx An {@link OnTimerContext} that allows querying the timestamp of the firing timer,
+ * querying the {@link TimeDomain} of the firing timer and getting a {@link TimerService}
+ * for registering timers and querying the time. The context is only valid during the
+ * invocation of this method, do not store it.
+ * @param out The collector for returning result values.
+ * @throws Exception This method may throw exceptions. Throwing an exception will cause the
+ * operation to fail and may trigger recovery.
+ */
+ public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception {}
+
+ /**
Review Comment:
It seems you have removed the `Context` from `KeyedCoProcessFunction` but the javadoc left over. It should be removed.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperator.java:
##########
@@ -0,0 +1,237 @@
+/*
+ * 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.flink.runtime.asyncprocessing.operators.co;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclaredVariable;
+import org.apache.flink.runtime.asyncprocessing.functions.DeclaringAsyncKeyedCoProcessFunction;
+import org.apache.flink.runtime.asyncprocessing.operators.AbstractAsyncStateUdfStreamOperator;
+import org.apache.flink.runtime.asyncprocessing.operators.TimestampedCollectorWithDeclaredVariable;
+import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.runtime.state.VoidNamespaceSerializer;
+import org.apache.flink.streaming.api.SimpleTimerService;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.streaming.api.operators.InternalTimer;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.streaming.api.operators.Triggerable;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing keyed {@link
+ * KeyedCoProcessFunction KeyedCoProcessFunction}.
+ */
+@Internal
+public class AsyncKeyedCoProcessOperator<K, IN1, IN2, OUT>
+ extends AbstractAsyncStateUdfStreamOperator<OUT, KeyedCoProcessFunction<K, IN1, IN2, OUT>>
+ implements TwoInputStreamOperator<IN1, IN2, OUT>, Triggerable<K, VoidNamespace> {
+
+ private static final long serialVersionUID = 1L;
+
+ // Shared timestamp variable for collector, context and onTimerContext.
+ private transient DeclaredVariable<Long> sharedTimestamp;
+
+ private transient TimestampedCollectorWithDeclaredVariable<OUT> collector;
+
+ private transient ContextImpl<K, IN1, IN2, OUT> context;
+
+ private transient OnTimerContextImpl<K, IN1, IN2, OUT> onTimerContext;
+
+ private transient ThrowingConsumer<IN1, Exception> processor1;
+ private transient ThrowingConsumer<IN2, Exception> processor2;
+ private transient ThrowingConsumer<Long, Exception> timerProcessor;
+
+ public AsyncKeyedCoProcessOperator(
+ KeyedCoProcessFunction<K, IN1, IN2, OUT> keyedCoProcessFunction) {
+ super(keyedCoProcessFunction);
+ }
+
+ @Override
+ public void open() throws Exception {
Review Comment:
Better add `@SuppressWarnings({"unchecked", "rawtypes"})` here to suppress the annoying warning.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2024651235
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/ConnectedStreams.java:
##########
@@ -69,12 +71,14 @@ public class ConnectedStreams<IN1, IN2> {
protected final StreamExecutionEnvironment environment;
protected final DataStream<IN1> inputStream1;
protected final DataStream<IN2> inputStream2;
+ protected boolean isEnableAsyncState;
protected ConnectedStreams(
StreamExecutionEnvironment env, DataStream<IN1> input1, DataStream<IN2> input2) {
this.environment = requireNonNull(env);
this.inputStream1 = requireNonNull(input1);
this.inputStream2 = requireNonNull(input2);
+ this.isEnableAsyncState = false;
Review Comment:
The `isEnableAsyncState` should be initialized considering `input1` and `input2`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006044413
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperator.java:
##########
@@ -0,0 +1,237 @@
+/*
+ * 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.flink.runtime.asyncprocessing.operators.co;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclaredVariable;
+import org.apache.flink.runtime.asyncprocessing.functions.DeclaringAsyncKeyedCoProcessFunction;
+import org.apache.flink.runtime.asyncprocessing.operators.AbstractAsyncStateUdfStreamOperator;
+import org.apache.flink.runtime.asyncprocessing.operators.TimestampedCollectorWithDeclaredVariable;
+import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.runtime.state.VoidNamespaceSerializer;
+import org.apache.flink.streaming.api.SimpleTimerService;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.streaming.api.operators.InternalTimer;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.streaming.api.operators.Triggerable;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing keyed {@link
+ * KeyedCoProcessFunction KeyedCoProcessFunction}.
+ */
+@Internal
+public class AsyncKeyedCoProcessOperator<K, IN1, IN2, OUT>
+ extends AbstractAsyncStateUdfStreamOperator<OUT, KeyedCoProcessFunction<K, IN1, IN2, OUT>>
+ implements TwoInputStreamOperator<IN1, IN2, OUT>, Triggerable<K, VoidNamespace> {
+
+ private static final long serialVersionUID = 1L;
+
+ // Shared timestamp variable for collector, context and onTimerContext.
+ private transient DeclaredVariable<Long> sharedTimestamp;
+
+ private transient TimestampedCollectorWithDeclaredVariable<OUT> collector;
+
+ private transient ContextImpl<K, IN1, IN2, OUT> context;
+
+ private transient OnTimerContextImpl<K, IN1, IN2, OUT> onTimerContext;
+
+ private transient ThrowingConsumer<IN1, Exception> processor1;
+ private transient ThrowingConsumer<IN2, Exception> processor2;
+ private transient ThrowingConsumer<Long, Exception> timerProcessor;
+
+ public AsyncKeyedCoProcessOperator(
+ KeyedCoProcessFunction<K, IN1, IN2, OUT> keyedCoProcessFunction) {
+ super(keyedCoProcessFunction);
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ sharedTimestamp =
+ declarationContext.declareVariable(
+ LongSerializer.INSTANCE,
+ "_AsyncCoKeyedProcessOperator$sharedTimestamp",
+ null);
+
+ collector = new TimestampedCollectorWithDeclaredVariable<>(output, sharedTimestamp);
+
+ InternalTimerService<VoidNamespace> internalTimerService =
+ getInternalTimerService("user-timers", VoidNamespaceSerializer.INSTANCE, this);
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2020796756
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/asyncprocess/AsyncProcTimeIntervalJoin.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.flink.table.runtime.operators.join.interval.asyncprocess;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
+import org.apache.flink.table.runtime.operators.join.interval.IntervalJoinFunction;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+
+/** The function to execute processing time interval stream inner-join. */
+public final class AsyncProcTimeIntervalJoin extends AsyncTimeIntervalJoin {
+
+ private static final long serialVersionUID = 9204647938032023101L;
Review Comment:
We could give another random number, different from the one for `ProcTimeIntervalJoin`.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/asyncprocess/AsyncTimeIntervalJoin.java:
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.flink.table.runtime.operators.join.interval.asyncprocess;
+
+import org.apache.flink.api.common.functions.DefaultOpenContext;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.v2.MapState;
+import org.apache.flink.api.common.state.v2.MapStateDescriptor;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.api.common.state.v2.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ListTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
+import org.apache.flink.table.runtime.operators.join.OuterJoinPaddingUtil;
+import org.apache.flink.table.runtime.operators.join.interval.EmitAwareCollector;
+import org.apache.flink.table.runtime.operators.join.interval.IntervalJoinFunction;
+import org.apache.flink.table.runtime.operators.join.interval.TimeIntervalJoinBase;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/** The TimeIntervalJoinBase with async state. */
+public abstract class AsyncTimeIntervalJoin extends TimeIntervalJoinBase {
+ // cache to store rows form the left stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>> leftCache;
+ // cache to store rows from the right stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>> rightCache;
+
+ // state to record the timer on the left stream. 0 means no timer set
+ private transient ValueState<Long> leftTimerState;
+ // state to record the timer on the right stream. 0 means no timer set
+ private transient ValueState<Long> rightTimerState;
+
+ protected AsyncTimeIntervalJoin(
+ FlinkJoinType joinType,
+ long leftLowerBound,
+ long leftUpperBound,
+ long allowedLateness,
+ long minCleanUpInterval,
+ InternalTypeInfo<RowData> leftType,
+ InternalTypeInfo<RowData> rightType,
+ IntervalJoinFunction joinFunc) {
+ super(
+ joinType,
+ leftLowerBound,
+ leftUpperBound,
+ allowedLateness,
+ minCleanUpInterval,
+ leftType,
+ rightType,
+ joinFunc);
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ joinFunction.setRuntimeContext(getRuntimeContext());
+ joinFunction.open(DefaultOpenContext.INSTANCE);
+ joinCollector = new EmitAwareCollector();
+
+ // Initialize the data caches.
+ ListTypeInfo<Tuple2<RowData, Boolean>> leftRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(leftType, BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>> leftMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinLeftCache", BasicTypeInfo.LONG_TYPE_INFO, leftRowListTypeInfo);
+ leftCache = getRuntimeContext().getMapState(leftMapStateDescriptor);
+
+ ListTypeInfo<Tuple2<RowData, Boolean>> rightRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(rightType, BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>> rightMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinRightCache",
+ BasicTypeInfo.LONG_TYPE_INFO,
+ rightRowListTypeInfo);
+ rightCache = getRuntimeContext().getMapState(rightMapStateDescriptor);
+
+ // Initialize the timer states.
+ ValueStateDescriptor<Long> leftValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinLeftTimerState", Long.class);
+ leftTimerState = getRuntimeContext().getState(leftValueStateDescriptor);
+
+ ValueStateDescriptor<Long> rightValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinRightTimerState", Long.class);
+ rightTimerState = getRuntimeContext().getState(rightValueStateDescriptor);
+
+ paddingUtil = new OuterJoinPaddingUtil(leftType.toRowSize(), rightType.toRowSize());
+ }
+
+ @Override
+ public void onTimer(long timestamp, OnTimerContext ctx, Collector<RowData> out)
+ throws Exception {
+ joinFunction.setJoinKey(ctx.getCurrentKey());
+ joinCollector.setInnerCollector(out);
+ updateOperatorTime(ctx);
+ // In the future, we should separate the left and right watermarks. Otherwise, the
+ // registered timer of the faster stream will be delayed, even if the watermarks have
+ // already been emitted by the source.
+ leftTimerState
+ .asyncValue()
+ .thenAccept(
+ leftCleanUpTime -> {
+ if (leftCleanUpTime != null && timestamp == leftCleanUpTime) {
+ rightExpirationTime =
+ calExpirationTime(leftOperatorTime, rightRelativeSize);
+ removeExpiredRows(
+ rightExpirationTime,
+ rightCache,
+ leftTimerState,
+ ctx,
+ false);
+ }
+ });
+
+ rightTimerState
+ .asyncValue()
+ .thenAccept(
+ rightCleanUpTime -> {
+ if (rightCleanUpTime != null && timestamp == rightCleanUpTime) {
+ leftExpirationTime =
+ calExpirationTime(rightOperatorTime, leftRelativeSize);
+ removeExpiredRows(
+ leftExpirationTime, leftCache, rightTimerState, ctx, true);
+ }
+ });
+ }
+
+ @Override
+ protected boolean iterateRightCache(
+ RowData leftRow, long rightQualifiedLowerBound, long rightQualifiedUpperBound)
+ throws Exception {
+ Set<Long> toRemove = new HashSet<>();
+ final AtomicBoolean emitted = new AtomicBoolean(false);
+ CompletableFuture<Boolean> emitResult = new CompletableFuture<>();
+
+ Iterator<Map.Entry<Long, List<Tuple2<RowData, Boolean>>>> rightIterator =
+ rightCache.iterator();
+ rightCache
+ .asyncEntries()
+ .thenCompose(
+ iter ->
+ iter.onNext(
+ entry -> {
+ boolean entryEmitted =
+ tryMatchLeftRowsByTime(
+ leftRow,
+ rightIterator.next(),
+ rightQualifiedLowerBound,
+ rightQualifiedUpperBound,
+ toRemove);
+ if (entryEmitted) {
+ emitted.set(true);
+ }
+ }))
+ .thenAccept(
+ V -> {
+ emitResult.complete(emitted.get());
+ for (Long key : toRemove) {
+ rightCache.asyncRemove(key);
+ }
+ });
+ return emitResult.get();
Review Comment:
We better not blocking `get()` here. It will block the task thread as well as those callbacks for async state requests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2013618187
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * and go into recovery.
+ */
+ public void processElement1(IN1 value, Context ctx, Collector<OUT> out) throws Exception {
Review Comment:
We can't rename it, because `processElement1` is inherited from `KeyedCoProcessFunction`. I added `override` here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006037930
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * 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.");
+ }
+
+ /**
+ * This method is called for each element in the second of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * and go into recovery.
+ */
+ public void processElement2(IN2 value, Context ctx, Collector<OUT> out) throws Exception {
+ throw new IllegalAccessException("This method is replaced by declareProcess2.");
+ }
+
+ /**
+ * Called when a timer set using {@link TimerService} fires.
+ *
+ * @param timestamp The timestamp of the firing timer.
+ * @param ctx An {@link OnTimerContext} that allows querying the timestamp of the firing timer,
+ * querying the {@link TimeDomain} of the firing timer and getting a {@link TimerService}
+ * for registering timers and querying the time. The context is only valid during the
+ * invocation of this method, do not store it.
+ * @param out The collector for returning result values.
+ * @throws Exception This method may throw exceptions. Throwing an exception will cause the
+ * operation to fail and may trigger recovery.
+ */
+ public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception {}
+
+ /**
+ * Information available in an invocation of {@link #processElement1(Object, Context,
+ * Collector)}/ {@link #processElement2(Object, Context, Collector)} or {@link #onTimer(long,
+ * OnTimerContext, Collector)}.
+ */
+
+ /**
+ * Declaring variables before {@link #declareProcess1} and {@link #declareProcess2} and {@link
+ * #declareOnTimer}.
+ */
+ public void declareVariables(DeclarationContext context) {}
+
+ /**
+ * Declare a process for one element from the first of the connected streams.
+ *
+ * <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 context the context that provides useful methods to define named callbacks.
+ * @param ctx A {@link Context} that allows querying the timestamp of the element and getting a
+ * {@link TimerService} for registering timers and querying the time. The context is only
+ * valid during the invocation of this method, do not store it.
+ * @param out The collector for returning result values.
+ * @return the whole processing logic just like {@code processElement}.
+ */
+ public abstract ThrowingConsumer<IN1, Exception> declareProcess1(
+ DeclarationContext context, Context ctx, Collector<OUT> out)
+ throws DeclarationException;
+
+ /**
+ * Declare a process for one element from the second of the connected streams.
+ *
+ * <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 context the context that provides useful methods to define named callbacks.
+ * @param ctx A {@link Context} that allows querying the timestamp of the element and getting a
+ * {@link TimerService} for registering timers and querying the time. The context is only
+ * valid during the invocation of this method, do not store it.
+ * @param out The collector for returning result values.
+ * @return the whole processing logic just like {@code processElement}.
+ */
+ public abstract ThrowingConsumer<IN2, Exception> declareProcess2(
+ DeclarationContext context, Context ctx, Collector<OUT> out)
+ throws DeclarationException;
+
+ /**
+ * Declare a procedure which is called when a timer set using {@link TimerService} fires.
+ *
+ * @param context the context that provides useful methods to define named callbacks.
+ * @param ctx An {@link OnTimerContext} that allows querying the timestamp, the {@link
+ * TimeDomain}, and the key of the firing timer and getting a {@link TimerService} for
+ * registering timers and querying the time. The context is only valid during the invocation
+ * of this method, do not store it.
+ * @param out The processor for processing timestamps.
+ */
+ public ThrowingConsumer<Long, Exception> declareOnTimer(
Review Comment:
nit: declareOnTimer -> declareOnTimerProcessor
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006039802
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperator.java:
##########
@@ -0,0 +1,237 @@
+/*
+ * 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.flink.runtime.asyncprocessing.operators.co;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclaredVariable;
+import org.apache.flink.runtime.asyncprocessing.functions.DeclaringAsyncKeyedCoProcessFunction;
+import org.apache.flink.runtime.asyncprocessing.operators.AbstractAsyncStateUdfStreamOperator;
+import org.apache.flink.runtime.asyncprocessing.operators.TimestampedCollectorWithDeclaredVariable;
+import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.runtime.state.VoidNamespaceSerializer;
+import org.apache.flink.streaming.api.SimpleTimerService;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.streaming.api.operators.InternalTimer;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.streaming.api.operators.Triggerable;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing keyed {@link
+ * KeyedCoProcessFunction KeyedCoProcessFunction}.
+ */
+@Internal
+public class AsyncKeyedCoProcessOperator<K, IN1, IN2, OUT>
+ extends AbstractAsyncStateUdfStreamOperator<OUT, KeyedCoProcessFunction<K, IN1, IN2, OUT>>
+ implements TwoInputStreamOperator<IN1, IN2, OUT>, Triggerable<K, VoidNamespace> {
+
+ private static final long serialVersionUID = 1L;
+
+ // Shared timestamp variable for collector, context and onTimerContext.
+ private transient DeclaredVariable<Long> sharedTimestamp;
Review Comment:
what do we mean by the sharedTimestamp? Is this event time?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2013632788
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * and go into recovery.
+ */
+ public void processElement1(IN1 value, Context ctx, Collector<OUT> out) throws Exception {
Review Comment:
I don't think this is necessary because it is inherited from `KeyedCoProcessFunction `. I add `override` here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2022374399
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/asyncprocess/AsyncTimeIntervalJoin.java:
##########
@@ -0,0 +1,345 @@
+/*
+ * 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.flink.table.runtime.operators.join.interval.asyncprocess;
+
+import org.apache.flink.api.common.functions.DefaultOpenContext;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.v2.MapState;
+import org.apache.flink.api.common.state.v2.MapStateDescriptor;
+import org.apache.flink.api.common.state.v2.StateFuture;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.api.common.state.v2.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ListTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
+import org.apache.flink.table.runtime.operators.join.OuterJoinPaddingUtil;
+import org.apache.flink.table.runtime.operators.join.interval.EmitAwareCollector;
+import org.apache.flink.table.runtime.operators.join.interval.IntervalJoinFunction;
+import org.apache.flink.table.runtime.operators.join.interval.TimeIntervalJoinBase;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/** The TimeIntervalJoinBase with async state. */
+public abstract class AsyncTimeIntervalJoin extends TimeIntervalJoinBase {
+ // cache to store rows form the left stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>> leftCache;
+ // cache to store rows from the right stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>> rightCache;
+
+ // state to record the timer on the left stream. 0 means no timer set
+ private transient ValueState<Long> leftTimerState;
+ // state to record the timer on the right stream. 0 means no timer set
+ private transient ValueState<Long> rightTimerState;
+
+ protected AsyncTimeIntervalJoin(
+ FlinkJoinType joinType,
+ long leftLowerBound,
+ long leftUpperBound,
+ long allowedLateness,
+ long minCleanUpInterval,
+ InternalTypeInfo<RowData> leftType,
+ InternalTypeInfo<RowData> rightType,
+ IntervalJoinFunction joinFunc) {
+ super(
+ joinType,
+ leftLowerBound,
+ leftUpperBound,
+ allowedLateness,
+ minCleanUpInterval,
+ leftType,
+ rightType,
+ joinFunc);
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ joinFunction.setRuntimeContext(getRuntimeContext());
+ joinFunction.open(DefaultOpenContext.INSTANCE);
+ joinCollector = new EmitAwareCollector();
+
+ // Initialize the data caches.
+ ListTypeInfo<Tuple2<RowData, Boolean>> leftRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(leftType, BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>> leftMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinLeftCache", BasicTypeInfo.LONG_TYPE_INFO, leftRowListTypeInfo);
+ leftCache = getRuntimeContext().getMapState(leftMapStateDescriptor);
+
+ ListTypeInfo<Tuple2<RowData, Boolean>> rightRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(rightType, BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>> rightMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinRightCache",
+ BasicTypeInfo.LONG_TYPE_INFO,
+ rightRowListTypeInfo);
+ rightCache = getRuntimeContext().getMapState(rightMapStateDescriptor);
+
+ // Initialize the timer states.
+ ValueStateDescriptor<Long> leftValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinLeftTimerState", Long.class);
+ leftTimerState = getRuntimeContext().getState(leftValueStateDescriptor);
+
+ ValueStateDescriptor<Long> rightValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinRightTimerState", Long.class);
+ rightTimerState = getRuntimeContext().getState(rightValueStateDescriptor);
+
+ paddingUtil = new OuterJoinPaddingUtil(leftType.toRowSize(), rightType.toRowSize());
+ }
+
+ @Override
+ public void onTimer(long timestamp, OnTimerContext ctx, Collector<RowData> out)
+ throws Exception {
+ joinFunction.setJoinKey(ctx.getCurrentKey());
+ joinCollector.setInnerCollector(out);
Review Comment:
the internal state of `joinFunction` and `joinCollector` should be maintained by `DeclaredVariable` since it belongs to the context for async processing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006029603
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * and go into recovery.
+ */
+ public void processElement1(IN1 value, Context ctx, Collector<OUT> out) throws Exception {
Review Comment:
we should label all the method as @public or @exoerimental or the like.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * and go into recovery.
+ */
+ public void processElement1(IN1 value, Context ctx, Collector<OUT> out) throws Exception {
Review Comment:
we should label all the method as @public or @experimental or the like.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #26328:
URL: https://github.com/apache/flink/pull/26328#issuecomment-2740257648
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d3b5ae4f3c63dc46f702d265728514faac954bb8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3b5ae4f3c63dc46f702d265728514faac954bb8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3b5ae4f3c63dc46f702d265728514faac954bb8 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2013624776
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
Review Comment:
Same as the question above, this method is inherited from `KeyedCoProcessFunction`, the naming is consistent with `KeyedCoProcessFunction`. I prefer keeping it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2022380506
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/asyncprocess/AsyncTimeIntervalJoin.java:
##########
@@ -0,0 +1,345 @@
+/*
+ * 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.flink.table.runtime.operators.join.interval.asyncprocess;
+
+import org.apache.flink.api.common.functions.DefaultOpenContext;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.v2.MapState;
+import org.apache.flink.api.common.state.v2.MapStateDescriptor;
+import org.apache.flink.api.common.state.v2.StateFuture;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.api.common.state.v2.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ListTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
+import org.apache.flink.table.runtime.operators.join.OuterJoinPaddingUtil;
+import org.apache.flink.table.runtime.operators.join.interval.EmitAwareCollector;
+import org.apache.flink.table.runtime.operators.join.interval.IntervalJoinFunction;
+import org.apache.flink.table.runtime.operators.join.interval.TimeIntervalJoinBase;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/** The TimeIntervalJoinBase with async state. */
+public abstract class AsyncTimeIntervalJoin extends TimeIntervalJoinBase {
+ // cache to store rows form the left stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>> leftCache;
+ // cache to store rows from the right stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>> rightCache;
+
+ // state to record the timer on the left stream. 0 means no timer set
+ private transient ValueState<Long> leftTimerState;
+ // state to record the timer on the right stream. 0 means no timer set
+ private transient ValueState<Long> rightTimerState;
+
+ protected AsyncTimeIntervalJoin(
+ FlinkJoinType joinType,
+ long leftLowerBound,
+ long leftUpperBound,
+ long allowedLateness,
+ long minCleanUpInterval,
+ InternalTypeInfo<RowData> leftType,
+ InternalTypeInfo<RowData> rightType,
+ IntervalJoinFunction joinFunc) {
+ super(
+ joinType,
+ leftLowerBound,
+ leftUpperBound,
+ allowedLateness,
+ minCleanUpInterval,
+ leftType,
+ rightType,
+ joinFunc);
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ joinFunction.setRuntimeContext(getRuntimeContext());
+ joinFunction.open(DefaultOpenContext.INSTANCE);
+ joinCollector = new EmitAwareCollector();
+
+ // Initialize the data caches.
+ ListTypeInfo<Tuple2<RowData, Boolean>> leftRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(leftType, BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>> leftMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinLeftCache", BasicTypeInfo.LONG_TYPE_INFO, leftRowListTypeInfo);
+ leftCache = getRuntimeContext().getMapState(leftMapStateDescriptor);
+
+ ListTypeInfo<Tuple2<RowData, Boolean>> rightRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(rightType, BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>> rightMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinRightCache",
+ BasicTypeInfo.LONG_TYPE_INFO,
+ rightRowListTypeInfo);
+ rightCache = getRuntimeContext().getMapState(rightMapStateDescriptor);
+
+ // Initialize the timer states.
+ ValueStateDescriptor<Long> leftValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinLeftTimerState", Long.class);
+ leftTimerState = getRuntimeContext().getState(leftValueStateDescriptor);
+
+ ValueStateDescriptor<Long> rightValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinRightTimerState", Long.class);
+ rightTimerState = getRuntimeContext().getState(rightValueStateDescriptor);
+
+ paddingUtil = new OuterJoinPaddingUtil(leftType.toRowSize(), rightType.toRowSize());
+ }
+
+ @Override
+ public void onTimer(long timestamp, OnTimerContext ctx, Collector<RowData> out)
+ throws Exception {
+ joinFunction.setJoinKey(ctx.getCurrentKey());
+ joinCollector.setInnerCollector(out);
+ updateOperatorTime(ctx);
+ // In the future, we should separate the left and right watermarks. Otherwise, the
+ // registered timer of the faster stream will be delayed, even if the watermarks have
+ // already been emitted by the source.
+ leftTimerState
+ .asyncValue()
+ .thenAccept(
+ leftCleanUpTime -> {
+ if (leftCleanUpTime != null && timestamp == leftCleanUpTime) {
+ rightExpirationTime =
+ calExpirationTime(leftOperatorTime, rightRelativeSize);
+ removeExpiredRows(
+ rightExpirationTime,
+ rightCache,
+ leftTimerState,
+ ctx,
+ false);
+ }
+ });
+
+ rightTimerState
+ .asyncValue()
+ .thenAccept(
+ rightCleanUpTime -> {
+ if (rightCleanUpTime != null && timestamp == rightCleanUpTime) {
+ leftExpirationTime =
+ calExpirationTime(rightOperatorTime, leftRelativeSize);
+ removeExpiredRows(
+ leftExpirationTime, leftCache, rightTimerState, ctx, true);
+ }
+ });
+ }
+
+ @Override
+ protected StateFuture<Boolean> iterateRightCache(
+ RowData leftRow, long rightQualifiedLowerBound, long rightQualifiedUpperBound)
+ throws Exception {
+ Set<Long> toRemove = new HashSet<>();
+ final AtomicBoolean emitted = new AtomicBoolean(false);
+ CompletableFuture<Boolean> emitResult = new CompletableFuture<>();
+
+ Iterator<Map.Entry<Long, List<Tuple2<RowData, Boolean>>>> rightIterator =
+ rightCache.iterator();
+ return rightCache
+ .asyncEntries()
+ .thenCompose(
+ iter ->
+ iter.onNext(
+ entry -> {
+ boolean entryEmitted =
+ tryMatchLeftRowsByTime(
+ leftRow,
+ rightIterator.next(),
Review Comment:
```suggestion
entry,
```
And remove the `rightIterator` above
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/asyncprocess/AsyncTimeIntervalJoin.java:
##########
@@ -0,0 +1,345 @@
+/*
+ * 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.flink.table.runtime.operators.join.interval.asyncprocess;
+
+import org.apache.flink.api.common.functions.DefaultOpenContext;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.v2.MapState;
+import org.apache.flink.api.common.state.v2.MapStateDescriptor;
+import org.apache.flink.api.common.state.v2.StateFuture;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.api.common.state.v2.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ListTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
+import org.apache.flink.table.runtime.operators.join.OuterJoinPaddingUtil;
+import org.apache.flink.table.runtime.operators.join.interval.EmitAwareCollector;
+import org.apache.flink.table.runtime.operators.join.interval.IntervalJoinFunction;
+import org.apache.flink.table.runtime.operators.join.interval.TimeIntervalJoinBase;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/** The TimeIntervalJoinBase with async state. */
+public abstract class AsyncTimeIntervalJoin extends TimeIntervalJoinBase {
+ // cache to store rows form the left stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>> leftCache;
+ // cache to store rows from the right stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>> rightCache;
+
+ // state to record the timer on the left stream. 0 means no timer set
+ private transient ValueState<Long> leftTimerState;
+ // state to record the timer on the right stream. 0 means no timer set
+ private transient ValueState<Long> rightTimerState;
+
+ protected AsyncTimeIntervalJoin(
+ FlinkJoinType joinType,
+ long leftLowerBound,
+ long leftUpperBound,
+ long allowedLateness,
+ long minCleanUpInterval,
+ InternalTypeInfo<RowData> leftType,
+ InternalTypeInfo<RowData> rightType,
+ IntervalJoinFunction joinFunc) {
+ super(
+ joinType,
+ leftLowerBound,
+ leftUpperBound,
+ allowedLateness,
+ minCleanUpInterval,
+ leftType,
+ rightType,
+ joinFunc);
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ joinFunction.setRuntimeContext(getRuntimeContext());
+ joinFunction.open(DefaultOpenContext.INSTANCE);
+ joinCollector = new EmitAwareCollector();
+
+ // Initialize the data caches.
+ ListTypeInfo<Tuple2<RowData, Boolean>> leftRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(leftType, BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>> leftMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinLeftCache", BasicTypeInfo.LONG_TYPE_INFO, leftRowListTypeInfo);
+ leftCache = getRuntimeContext().getMapState(leftMapStateDescriptor);
+
+ ListTypeInfo<Tuple2<RowData, Boolean>> rightRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(rightType, BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>> rightMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinRightCache",
+ BasicTypeInfo.LONG_TYPE_INFO,
+ rightRowListTypeInfo);
+ rightCache = getRuntimeContext().getMapState(rightMapStateDescriptor);
+
+ // Initialize the timer states.
+ ValueStateDescriptor<Long> leftValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinLeftTimerState", Long.class);
+ leftTimerState = getRuntimeContext().getState(leftValueStateDescriptor);
+
+ ValueStateDescriptor<Long> rightValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinRightTimerState", Long.class);
+ rightTimerState = getRuntimeContext().getState(rightValueStateDescriptor);
+
+ paddingUtil = new OuterJoinPaddingUtil(leftType.toRowSize(), rightType.toRowSize());
+ }
+
+ @Override
+ public void onTimer(long timestamp, OnTimerContext ctx, Collector<RowData> out)
+ throws Exception {
+ joinFunction.setJoinKey(ctx.getCurrentKey());
+ joinCollector.setInnerCollector(out);
+ updateOperatorTime(ctx);
+ // In the future, we should separate the left and right watermarks. Otherwise, the
+ // registered timer of the faster stream will be delayed, even if the watermarks have
+ // already been emitted by the source.
+ leftTimerState
+ .asyncValue()
+ .thenAccept(
+ leftCleanUpTime -> {
+ if (leftCleanUpTime != null && timestamp == leftCleanUpTime) {
+ rightExpirationTime =
+ calExpirationTime(leftOperatorTime, rightRelativeSize);
+ removeExpiredRows(
+ rightExpirationTime,
+ rightCache,
+ leftTimerState,
+ ctx,
+ false);
+ }
+ });
+
+ rightTimerState
+ .asyncValue()
+ .thenAccept(
+ rightCleanUpTime -> {
+ if (rightCleanUpTime != null && timestamp == rightCleanUpTime) {
+ leftExpirationTime =
+ calExpirationTime(rightOperatorTime, leftRelativeSize);
+ removeExpiredRows(
+ leftExpirationTime, leftCache, rightTimerState, ctx, true);
+ }
+ });
+ }
+
+ @Override
+ protected StateFuture<Boolean> iterateRightCache(
+ RowData leftRow, long rightQualifiedLowerBound, long rightQualifiedUpperBound)
+ throws Exception {
+ Set<Long> toRemove = new HashSet<>();
+ final AtomicBoolean emitted = new AtomicBoolean(false);
+ CompletableFuture<Boolean> emitResult = new CompletableFuture<>();
+
+ Iterator<Map.Entry<Long, List<Tuple2<RowData, Boolean>>>> rightIterator =
+ rightCache.iterator();
+ return rightCache
+ .asyncEntries()
+ .thenCompose(
+ iter ->
+ iter.onNext(
+ entry -> {
+ boolean entryEmitted =
+ tryMatchLeftRowsByTime(
+ leftRow,
+ rightIterator.next(),
+ rightQualifiedLowerBound,
+ rightQualifiedUpperBound,
+ toRemove);
+ if (entryEmitted) {
+ emitted.set(true);
+ }
+ }))
+ .thenApply(
+ V -> {
+ for (Long key : toRemove) {
+ rightCache.asyncRemove(key);
+ }
+ return emitted.get();
+ });
+ }
+
+ @Override
+ protected StateFuture<Boolean> iterateLeftCache(
+ RowData rightRow, long leftQualifiedLowerBound, long leftQualifiedUpperBound)
+ throws Exception {
+ Set<Long> toRemove = new HashSet<>();
+ final AtomicBoolean emitted = new AtomicBoolean(false);
+ Iterator<Map.Entry<Long, List<Tuple2<RowData, Boolean>>>> leftIterator =
+ leftCache.iterator();
+ return leftCache
+ .asyncEntries()
+ .thenCompose(
+ iter ->
+ iter.onNext(
+ entry -> {
+ boolean entryEmitted =
+ tryMatchRightRowsByTime(
+ rightRow,
+ leftIterator.next(),
Review Comment:
```suggestion
entry,
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006052779
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperatorWithWatermarkDelay.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.flink.runtime.asyncprocessing.operators.co;
+
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Serializable;
+import java.util.function.Consumer;
+
+/** A {@link KeyedCoProcessOperator} that supports holding back watermarks with a static delay. */
+public class AsyncKeyedCoProcessOperatorWithWatermarkDelay<K, IN1, IN2, OUT>
Review Comment:
I cant see this class called anywhere. Even from test code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006003805
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
Review Comment:
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,"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2013624776
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
Review Comment:
Same as above question, this method is inherited from `KeyedCoProcessFunction`, the naming is consistent with `KeyedCoProcessFunction`. I prefer keeping it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2020488907
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperatorWithWatermarkDelay.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.flink.runtime.asyncprocessing.operators.co;
+
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Serializable;
+import java.util.function.Consumer;
+
+/** A {@link KeyedCoProcessOperator} that supports holding back watermarks with a static delay. */
+public class AsyncKeyedCoProcessOperatorWithWatermarkDelay<K, IN1, IN2, OUT>
+ extends AsyncKeyedCoProcessOperator<K, IN1, IN2, OUT> {
+ private static final long serialVersionUID = 1L;
+
+ private final Consumer<Watermark> emitter;
Review Comment:
Please remove the emitter and related initialization code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2024470829
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,129 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output stream.
+ *
+ * <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 timers the function can emit yet more elements.
+ *
+ * <p>An example use case for connected streams is the application of a set of rules that 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
+ * apply the previously stored rules to the element and emit a result, and/or register a timer that
+ * will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /** Override this method or use {@link #declareProcess1} instead. */
+ @Override
+ public void processElement1(IN1 value, Context ctx, Collector<OUT> out) throws Exception {
+ throw new IllegalAccessException("This method is replaced by declareProcess1.");
+ }
+
+ /** Override this method or use {@link #declareProcess2} instead. */
Review Comment:
Actually we dont recommend to use this method. How about rewrite this javadoc?
```suggestion
/** Override and finalize this method. Please use {@link #declareProcess2} instead. */
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -387,8 +387,12 @@ 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. Null if there is no need for
+ * following emitting.
*/
- public void postProcessWatermark(Watermark watermark) throws Exception {}
+ public Watermark postProcessWatermark(Watermark watermark) throws Exception {
Review Comment:
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)
##########
flink-runtime/src/main/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() {
Review Comment:
Seems unnecessary?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java:
##########
@@ -367,12 +368,18 @@ private TwoInputTransformation<RowData, RowData, RowData> createProcTimeJoin(
leftTypeInfo,
rightTypeInfo,
joinFunction);
-
+ TwoInputStreamOperator<RowData, RowData, RowData> operator;
+ if (config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_STATE_ENABLED)) {
+ // TODO: add async version procJoinFunc to use AsyncKeyedCoProcessOperator
+ operator = new KeyedCoProcessOperator<>(procJoinFunc);
+ } else {
+ operator = new KeyedCoProcessOperator<>(procJoinFunc);
+ }
Review Comment:
We could revert this part until we start to implement the SQL operator
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/ConnectedStreams.java:
##########
@@ -523,4 +529,14 @@ private <R> SingleOutputStreamOperator<R> doTransform(
return returnStream;
}
+
+ private boolean isEnableAsyncState() {
+ boolean enableAsyncState = false;
+ if ((inputStream1 instanceof KeyedStream) && (inputStream2 instanceof KeyedStream)) {
+ enableAsyncState =
+ ((KeyedStream<?, ?>) inputStream1).isEnableAsyncState()
+ && ((KeyedStream<?, ?>) inputStream2).isEnableAsyncState();
+ }
+ return enableAsyncState;
+ }
Review Comment:
I'd suggest we keep a member `boolean isEnableAsyncState` here. Thus user could perform `enableAsyncState()` on `ConnectedStreams`, WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2022580816
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/asyncprocess/AsyncTimeIntervalJoin.java:
##########
@@ -0,0 +1,345 @@
+/*
+ * 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.flink.table.runtime.operators.join.interval.asyncprocess;
+
+import org.apache.flink.api.common.functions.DefaultOpenContext;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.v2.MapState;
+import org.apache.flink.api.common.state.v2.MapStateDescriptor;
+import org.apache.flink.api.common.state.v2.StateFuture;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.api.common.state.v2.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ListTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
+import org.apache.flink.table.runtime.operators.join.OuterJoinPaddingUtil;
+import org.apache.flink.table.runtime.operators.join.interval.EmitAwareCollector;
+import org.apache.flink.table.runtime.operators.join.interval.IntervalJoinFunction;
+import org.apache.flink.table.runtime.operators.join.interval.TimeIntervalJoinBase;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/** The TimeIntervalJoinBase with async state. */
+public abstract class AsyncTimeIntervalJoin extends TimeIntervalJoinBase {
+ // cache to store rows form the left stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>> leftCache;
+ // cache to store rows from the right stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>> rightCache;
+
+ // state to record the timer on the left stream. 0 means no timer set
+ private transient ValueState<Long> leftTimerState;
+ // state to record the timer on the right stream. 0 means no timer set
+ private transient ValueState<Long> rightTimerState;
+
+ protected AsyncTimeIntervalJoin(
+ FlinkJoinType joinType,
+ long leftLowerBound,
+ long leftUpperBound,
+ long allowedLateness,
+ long minCleanUpInterval,
+ InternalTypeInfo<RowData> leftType,
+ InternalTypeInfo<RowData> rightType,
+ IntervalJoinFunction joinFunc) {
+ super(
+ joinType,
+ leftLowerBound,
+ leftUpperBound,
+ allowedLateness,
+ minCleanUpInterval,
+ leftType,
+ rightType,
+ joinFunc);
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ joinFunction.setRuntimeContext(getRuntimeContext());
+ joinFunction.open(DefaultOpenContext.INSTANCE);
+ joinCollector = new EmitAwareCollector();
+
+ // Initialize the data caches.
+ ListTypeInfo<Tuple2<RowData, Boolean>> leftRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(leftType, BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>> leftMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinLeftCache", BasicTypeInfo.LONG_TYPE_INFO, leftRowListTypeInfo);
+ leftCache = getRuntimeContext().getMapState(leftMapStateDescriptor);
+
+ ListTypeInfo<Tuple2<RowData, Boolean>> rightRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(rightType, BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>> rightMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinRightCache",
+ BasicTypeInfo.LONG_TYPE_INFO,
+ rightRowListTypeInfo);
+ rightCache = getRuntimeContext().getMapState(rightMapStateDescriptor);
+
+ // Initialize the timer states.
+ ValueStateDescriptor<Long> leftValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinLeftTimerState", Long.class);
+ leftTimerState = getRuntimeContext().getState(leftValueStateDescriptor);
+
+ ValueStateDescriptor<Long> rightValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinRightTimerState", Long.class);
+ rightTimerState = getRuntimeContext().getState(rightValueStateDescriptor);
+
+ paddingUtil = new OuterJoinPaddingUtil(leftType.toRowSize(), rightType.toRowSize());
+ }
+
+ @Override
+ public void onTimer(long timestamp, OnTimerContext ctx, Collector<RowData> out)
+ throws Exception {
+ joinFunction.setJoinKey(ctx.getCurrentKey());
+ joinCollector.setInnerCollector(out);
Review Comment:
👍 Good suggestion, I rewrote `AsyncTimeIntervalJoin` with `declareChain`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2013628520
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
Review Comment:
What is a `retriable exception`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2013638927
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperator.java:
##########
@@ -0,0 +1,237 @@
+/*
+ * 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.flink.runtime.asyncprocessing.operators.co;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclaredVariable;
+import org.apache.flink.runtime.asyncprocessing.functions.DeclaringAsyncKeyedCoProcessFunction;
+import org.apache.flink.runtime.asyncprocessing.operators.AbstractAsyncStateUdfStreamOperator;
+import org.apache.flink.runtime.asyncprocessing.operators.TimestampedCollectorWithDeclaredVariable;
+import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.runtime.state.VoidNamespaceSerializer;
+import org.apache.flink.streaming.api.SimpleTimerService;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.streaming.api.operators.InternalTimer;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.streaming.api.operators.Triggerable;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing keyed {@link
+ * KeyedCoProcessFunction KeyedCoProcessFunction}.
+ */
+@Internal
+public class AsyncKeyedCoProcessOperator<K, IN1, IN2, OUT>
+ extends AbstractAsyncStateUdfStreamOperator<OUT, KeyedCoProcessFunction<K, IN1, IN2, OUT>>
+ implements TwoInputStreamOperator<IN1, IN2, OUT>, Triggerable<K, VoidNamespace> {
+
+ private static final long serialVersionUID = 1L;
+
+ // Shared timestamp variable for collector, context and onTimerContext.
+ private transient DeclaredVariable<Long> sharedTimestamp;
Review Comment:
It is used to store one Timestamp. It can be event time or processing time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Implement async state version of KeyedCoProcessOperator [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia merged PR #26328:
URL: https://github.com/apache/flink/pull/26328
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2022111151
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/asyncprocess/AsyncProcTimeIntervalJoin.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.flink.table.runtime.operators.join.interval.asyncprocess;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
+import org.apache.flink.table.runtime.operators.join.interval.IntervalJoinFunction;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+
+/** The function to execute processing time interval stream inner-join. */
+public final class AsyncProcTimeIntervalJoin extends AsyncTimeIntervalJoin {
+
+ private static final long serialVersionUID = 9204647938032023101L;
Review Comment:
Changed all `serialVersionUID` of newly added class to `1L`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2020801368
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/SyncTimeIntervalJoin.java:
##########
@@ -0,0 +1,264 @@
+/*
+ * 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.flink.table.runtime.operators.join.interval;
+
+import org.apache.flink.api.common.functions.DefaultOpenContext;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.MapState;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ListTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
+import org.apache.flink.table.runtime.operators.join.OuterJoinPaddingUtil;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** The TimeIntervalJoinBase with sync state. */
+public abstract class SyncTimeIntervalJoin extends TimeIntervalJoinBase {
Review Comment:
How about remove the `Sync` in name
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2013630598
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * 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.");
+ }
+
+ /**
+ * This method is called for each element in the second of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * and go into recovery.
+ */
+ public void processElement2(IN2 value, Context ctx, Collector<OUT> out) throws Exception {
+ throw new IllegalAccessException("This method is replaced by declareProcess2.");
+ }
+
+ /**
+ * Called when a timer set using {@link TimerService} fires.
+ *
+ * @param timestamp The timestamp of the firing timer.
+ * @param ctx An {@link OnTimerContext} that allows querying the timestamp of the firing timer,
+ * querying the {@link TimeDomain} of the firing timer and getting a {@link TimerService}
+ * for registering timers and querying the time. The context is only valid during the
+ * invocation of this method, do not store it.
+ * @param out The collector for returning result values.
+ * @throws Exception This method may throw exceptions. Throwing an exception will cause the
+ * operation to fail and may trigger recovery.
+ */
+ public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception {}
+
+ /**
Review Comment:
It should be `/**`, because it is for a method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2013650773
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperatorWithWatermarkDelay.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.flink.runtime.asyncprocessing.operators.co;
+
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Serializable;
+import java.util.function.Consumer;
+
+/** A {@link KeyedCoProcessOperator} that supports holding back watermarks with a static delay. */
+public class AsyncKeyedCoProcessOperatorWithWatermarkDelay<K, IN1, IN2, OUT>
Review Comment:
You're right, It will be called at `StreamExecIntervalJoin` in the future after we implement async version `rowJoinFunc`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006027105
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * 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.");
+ }
+
+ /**
+ * This method is called for each element in the second of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * and go into recovery.
+ */
+ public void processElement2(IN2 value, Context ctx, Collector<OUT> out) throws Exception {
+ throw new IllegalAccessException("This method is replaced by declareProcess2.");
+ }
+
+ /**
+ * Called when a timer set using {@link TimerService} fires.
+ *
+ * @param timestamp The timestamp of the firing timer.
+ * @param ctx An {@link OnTimerContext} that allows querying the timestamp of the firing timer,
+ * querying the {@link TimeDomain} of the firing timer and getting a {@link TimerService}
+ * for registering timers and querying the time. The context is only valid during the
+ * invocation of this method, do not store it.
+ * @param out The collector for returning result values.
+ * @throws Exception This method may throw exceptions. Throwing an exception will cause the
+ * operation to fail and may trigger recovery.
+ */
+ public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception {}
+
+ /**
Review Comment:
this javadoc has no associated code. If this is a comment then it should be /* not /**
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006027105
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * 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.");
+ }
+
+ /**
+ * This method is called for each element in the second of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * and go into recovery.
+ */
+ public void processElement2(IN2 value, Context ctx, Collector<OUT> out) throws Exception {
+ throw new IllegalAccessException("This method is replaced by declareProcess2.");
+ }
+
+ /**
+ * Called when a timer set using {@link TimerService} fires.
+ *
+ * @param timestamp The timestamp of the firing timer.
+ * @param ctx An {@link OnTimerContext} that allows querying the timestamp of the firing timer,
+ * querying the {@link TimeDomain} of the firing timer and getting a {@link TimerService}
+ * for registering timers and querying the time. The context is only valid during the
+ * invocation of this method, do not store it.
+ * @param out The collector for returning result values.
+ * @throws Exception This method may throw exceptions. Throwing an exception will cause the
+ * operation to fail and may trigger recovery.
+ */
+ public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception {}
+
+ /**
Review Comment:
this javadoc has no associated code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2005993388
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
Review Comment:
nit: one. -> stream.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006014908
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * 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.");
Review Comment:
what does this exception message mean?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006008196
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
Review Comment:
nits:
remove the word `now`
remove the word `directly`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006017896
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
Review Comment:
why is this parameter called value? Could it be something more specific like streamElement
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Implement async state version of KeyedCoProcessOperator [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on PR #26328:
URL: https://github.com/apache/flink/pull/26328#issuecomment-2774208393
Thanks for reviewing, updated and merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2024546158
##########
flink-runtime/src/main/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() {
Review Comment:
Because `AsyncIntervalJoinOperator` is moved into `asyncprocessing` package, and `AsyncIntervalJoinOperatorTest` still needs these methods.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2013646777
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperator.java:
##########
@@ -0,0 +1,237 @@
+/*
+ * 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.flink.runtime.asyncprocessing.operators.co;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclaredVariable;
+import org.apache.flink.runtime.asyncprocessing.functions.DeclaringAsyncKeyedCoProcessFunction;
+import org.apache.flink.runtime.asyncprocessing.operators.AbstractAsyncStateUdfStreamOperator;
+import org.apache.flink.runtime.asyncprocessing.operators.TimestampedCollectorWithDeclaredVariable;
+import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.runtime.state.VoidNamespaceSerializer;
+import org.apache.flink.streaming.api.SimpleTimerService;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.streaming.api.operators.InternalTimer;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.streaming.api.operators.Triggerable;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing keyed {@link
+ * KeyedCoProcessFunction KeyedCoProcessFunction}.
+ */
+@Internal
+public class AsyncKeyedCoProcessOperator<K, IN1, IN2, OUT>
+ extends AbstractAsyncStateUdfStreamOperator<OUT, KeyedCoProcessFunction<K, IN1, IN2, OUT>>
+ implements TwoInputStreamOperator<IN1, IN2, OUT>, Triggerable<K, VoidNamespace> {
+
+ private static final long serialVersionUID = 1L;
+
+ // Shared timestamp variable for collector, context and onTimerContext.
+ private transient DeclaredVariable<Long> sharedTimestamp;
+
+ private transient TimestampedCollectorWithDeclaredVariable<OUT> collector;
+
+ private transient ContextImpl<K, IN1, IN2, OUT> context;
+
+ private transient OnTimerContextImpl<K, IN1, IN2, OUT> onTimerContext;
+
+ private transient ThrowingConsumer<IN1, Exception> processor1;
+ private transient ThrowingConsumer<IN2, Exception> processor2;
+ private transient ThrowingConsumer<Long, Exception> timerProcessor;
+
+ public AsyncKeyedCoProcessOperator(
+ KeyedCoProcessFunction<K, IN1, IN2, OUT> keyedCoProcessFunction) {
+ super(keyedCoProcessFunction);
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ sharedTimestamp =
+ declarationContext.declareVariable(
+ LongSerializer.INSTANCE,
+ "_AsyncCoKeyedProcessOperator$sharedTimestamp",
+ null);
+
+ collector = new TimestampedCollectorWithDeclaredVariable<>(output, sharedTimestamp);
+
+ InternalTimerService<VoidNamespace> internalTimerService =
+ getInternalTimerService("user-timers", VoidNamespaceSerializer.INSTANCE, this);
+
+ TimerService timerService = new SimpleTimerService(internalTimerService);
+
+ context = new ContextImpl<>(userFunction, timerService, sharedTimestamp);
+ onTimerContext = new OnTimerContextImpl<>(userFunction, timerService, declarationContext);
+ if (userFunction instanceof DeclaringAsyncKeyedCoProcessFunction) {
+ DeclaringAsyncKeyedCoProcessFunction declaringFunction =
+ (DeclaringAsyncKeyedCoProcessFunction) userFunction;
+ declaringFunction.declareVariables(declarationContext);
Review Comment:
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2013644336
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperator.java:
##########
@@ -0,0 +1,237 @@
+/*
+ * 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.flink.runtime.asyncprocessing.operators.co;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclaredVariable;
+import org.apache.flink.runtime.asyncprocessing.functions.DeclaringAsyncKeyedCoProcessFunction;
+import org.apache.flink.runtime.asyncprocessing.operators.AbstractAsyncStateUdfStreamOperator;
+import org.apache.flink.runtime.asyncprocessing.operators.TimestampedCollectorWithDeclaredVariable;
+import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.runtime.state.VoidNamespaceSerializer;
+import org.apache.flink.streaming.api.SimpleTimerService;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.streaming.api.operators.InternalTimer;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.streaming.api.operators.Triggerable;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing keyed {@link
+ * KeyedCoProcessFunction KeyedCoProcessFunction}.
+ */
+@Internal
+public class AsyncKeyedCoProcessOperator<K, IN1, IN2, OUT>
+ extends AbstractAsyncStateUdfStreamOperator<OUT, KeyedCoProcessFunction<K, IN1, IN2, OUT>>
+ implements TwoInputStreamOperator<IN1, IN2, OUT>, Triggerable<K, VoidNamespace> {
+
+ private static final long serialVersionUID = 1L;
+
+ // Shared timestamp variable for collector, context and onTimerContext.
+ private transient DeclaredVariable<Long> sharedTimestamp;
+
+ private transient TimestampedCollectorWithDeclaredVariable<OUT> collector;
+
+ private transient ContextImpl<K, IN1, IN2, OUT> context;
+
+ private transient OnTimerContextImpl<K, IN1, IN2, OUT> onTimerContext;
+
+ private transient ThrowingConsumer<IN1, Exception> processor1;
+ private transient ThrowingConsumer<IN2, Exception> processor2;
+ private transient ThrowingConsumer<Long, Exception> timerProcessor;
+
+ public AsyncKeyedCoProcessOperator(
+ KeyedCoProcessFunction<K, IN1, IN2, OUT> keyedCoProcessFunction) {
+ super(keyedCoProcessFunction);
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ sharedTimestamp =
+ declarationContext.declareVariable(
+ LongSerializer.INSTANCE,
+ "_AsyncCoKeyedProcessOperator$sharedTimestamp",
+ null);
+
+ collector = new TimestampedCollectorWithDeclaredVariable<>(output, sharedTimestamp);
+
+ InternalTimerService<VoidNamespace> internalTimerService =
+ getInternalTimerService("user-timers", VoidNamespaceSerializer.INSTANCE, this);
Review Comment:
The naming "user-timers" in `AsyncKeyedCoProcessOperator` is consistent with `KeyedCoProcessOperator#internalTimerService`. It represents a `timer-service` created by operator.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2013622013
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * 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.");
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2022110782
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/asyncprocess/AsyncTimeIntervalJoin.java:
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.flink.table.runtime.operators.join.interval.asyncprocess;
+
+import org.apache.flink.api.common.functions.DefaultOpenContext;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.v2.MapState;
+import org.apache.flink.api.common.state.v2.MapStateDescriptor;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.api.common.state.v2.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ListTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
+import org.apache.flink.table.runtime.operators.join.OuterJoinPaddingUtil;
+import org.apache.flink.table.runtime.operators.join.interval.EmitAwareCollector;
+import org.apache.flink.table.runtime.operators.join.interval.IntervalJoinFunction;
+import org.apache.flink.table.runtime.operators.join.interval.TimeIntervalJoinBase;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/** The TimeIntervalJoinBase with async state. */
+public abstract class AsyncTimeIntervalJoin extends TimeIntervalJoinBase {
+ // cache to store rows form the left stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>> leftCache;
+ // cache to store rows from the right stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>> rightCache;
+
+ // state to record the timer on the left stream. 0 means no timer set
+ private transient ValueState<Long> leftTimerState;
+ // state to record the timer on the right stream. 0 means no timer set
+ private transient ValueState<Long> rightTimerState;
+
+ protected AsyncTimeIntervalJoin(
+ FlinkJoinType joinType,
+ long leftLowerBound,
+ long leftUpperBound,
+ long allowedLateness,
+ long minCleanUpInterval,
+ InternalTypeInfo<RowData> leftType,
+ InternalTypeInfo<RowData> rightType,
+ IntervalJoinFunction joinFunc) {
+ super(
+ joinType,
+ leftLowerBound,
+ leftUpperBound,
+ allowedLateness,
+ minCleanUpInterval,
+ leftType,
+ rightType,
+ joinFunc);
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ joinFunction.setRuntimeContext(getRuntimeContext());
+ joinFunction.open(DefaultOpenContext.INSTANCE);
+ joinCollector = new EmitAwareCollector();
+
+ // Initialize the data caches.
+ ListTypeInfo<Tuple2<RowData, Boolean>> leftRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(leftType, BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>> leftMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinLeftCache", BasicTypeInfo.LONG_TYPE_INFO, leftRowListTypeInfo);
+ leftCache = getRuntimeContext().getMapState(leftMapStateDescriptor);
+
+ ListTypeInfo<Tuple2<RowData, Boolean>> rightRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(rightType, BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>> rightMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinRightCache",
+ BasicTypeInfo.LONG_TYPE_INFO,
+ rightRowListTypeInfo);
+ rightCache = getRuntimeContext().getMapState(rightMapStateDescriptor);
+
+ // Initialize the timer states.
+ ValueStateDescriptor<Long> leftValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinLeftTimerState", Long.class);
+ leftTimerState = getRuntimeContext().getState(leftValueStateDescriptor);
+
+ ValueStateDescriptor<Long> rightValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinRightTimerState", Long.class);
+ rightTimerState = getRuntimeContext().getState(rightValueStateDescriptor);
+
+ paddingUtil = new OuterJoinPaddingUtil(leftType.toRowSize(), rightType.toRowSize());
+ }
+
+ @Override
+ public void onTimer(long timestamp, OnTimerContext ctx, Collector<RowData> out)
+ throws Exception {
+ joinFunction.setJoinKey(ctx.getCurrentKey());
+ joinCollector.setInnerCollector(out);
+ updateOperatorTime(ctx);
+ // In the future, we should separate the left and right watermarks. Otherwise, the
+ // registered timer of the faster stream will be delayed, even if the watermarks have
+ // already been emitted by the source.
+ leftTimerState
+ .asyncValue()
+ .thenAccept(
+ leftCleanUpTime -> {
+ if (leftCleanUpTime != null && timestamp == leftCleanUpTime) {
+ rightExpirationTime =
+ calExpirationTime(leftOperatorTime, rightRelativeSize);
+ removeExpiredRows(
+ rightExpirationTime,
+ rightCache,
+ leftTimerState,
+ ctx,
+ false);
+ }
+ });
+
+ rightTimerState
+ .asyncValue()
+ .thenAccept(
+ rightCleanUpTime -> {
+ if (rightCleanUpTime != null && timestamp == rightCleanUpTime) {
+ leftExpirationTime =
+ calExpirationTime(rightOperatorTime, leftRelativeSize);
+ removeExpiredRows(
+ leftExpirationTime, leftCache, rightTimerState, ctx, true);
+ }
+ });
+ }
+
+ @Override
+ protected boolean iterateRightCache(
+ RowData leftRow, long rightQualifiedLowerBound, long rightQualifiedUpperBound)
+ throws Exception {
+ Set<Long> toRemove = new HashSet<>();
+ final AtomicBoolean emitted = new AtomicBoolean(false);
+ CompletableFuture<Boolean> emitResult = new CompletableFuture<>();
+
+ Iterator<Map.Entry<Long, List<Tuple2<RowData, Boolean>>>> rightIterator =
+ rightCache.iterator();
+ rightCache
+ .asyncEntries()
+ .thenCompose(
+ iter ->
+ iter.onNext(
+ entry -> {
+ boolean entryEmitted =
+ tryMatchLeftRowsByTime(
+ leftRow,
+ rightIterator.next(),
+ rightQualifiedLowerBound,
+ rightQualifiedUpperBound,
+ toRemove);
+ if (entryEmitted) {
+ emitted.set(true);
+ }
+ }))
+ .thenAccept(
+ V -> {
+ emitResult.complete(emitted.get());
+ for (Long key : toRemove) {
+ rightCache.asyncRemove(key);
+ }
+ });
+ return emitResult.get();
Review Comment:
👍I changed the return value to `StateFuture<Boolean>`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2024547952
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/ConnectedStreams.java:
##########
@@ -523,4 +529,14 @@ private <R> SingleOutputStreamOperator<R> doTransform(
return returnStream;
}
+
+ private boolean isEnableAsyncState() {
+ boolean enableAsyncState = false;
+ if ((inputStream1 instanceof KeyedStream) && (inputStream2 instanceof KeyedStream)) {
+ enableAsyncState =
+ ((KeyedStream<?, ?>) inputStream1).isEnableAsyncState()
+ && ((KeyedStream<?, ?>) inputStream2).isEnableAsyncState();
+ }
+ return enableAsyncState;
+ }
Review Comment:
👍, I added `enableAsyncState()` for `ConnectedStreams`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006013931
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * and go into recovery.
+ */
+ public void processElement1(IN1 value, Context ctx, Collector<OUT> out) throws Exception {
Review Comment:
nit: processElement1 -> processInputStream1Element or the like
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006023868
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
Review Comment:
can it throw a retriable exception?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006021841
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
Review Comment:
nit: maybe a list would be easier to read in the javadoc. A {@link Context} that allows querying of
<ol>
...
</ol>
When you say `querying the time` is this event or processing?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006034494
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * 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.");
+ }
+
+ /**
+ * This method is called for each element in the second of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * and go into recovery.
+ */
+ public void processElement2(IN2 value, Context ctx, Collector<OUT> out) throws Exception {
+ throw new IllegalAccessException("This method is replaced by declareProcess2.");
+ }
+
+ /**
+ * Called when a timer set using {@link TimerService} fires.
+ *
+ * @param timestamp The timestamp of the firing timer.
+ * @param ctx An {@link OnTimerContext} that allows querying the timestamp of the firing timer,
+ * querying the {@link TimeDomain} of the firing timer and getting a {@link TimerService}
+ * for registering timers and querying the time. The context is only valid during the
+ * invocation of this method, do not store it.
+ * @param out The collector for returning result values.
+ * @throws Exception This method may throw exceptions. Throwing an exception will cause the
+ * operation to fail and may trigger recovery.
+ */
+ public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception {}
+
+ /**
+ * Information available in an invocation of {@link #processElement1(Object, Context,
+ * Collector)}/ {@link #processElement2(Object, Context, Collector)} or {@link #onTimer(long,
+ * OnTimerContext, Collector)}.
+ */
+
+ /**
+ * Declaring variables before {@link #declareProcess1} and {@link #declareProcess2} and {@link
+ * #declareOnTimer}.
+ */
+ public void declareVariables(DeclarationContext context) {}
+
+ /**
+ * Declare a process for one element from the first of the connected streams.
+ *
+ * <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 context the context that provides useful methods to define named callbacks.
+ * @param ctx A {@link Context} that allows querying the timestamp of the element and getting a
+ * {@link TimerService} for registering timers and querying the time. The context is only
+ * valid during the invocation of this method, do not store it.
+ * @param out The collector for returning result values.
+ * @return the whole processing logic just like {@code processElement}.
+ */
+ public abstract ThrowingConsumer<IN1, Exception> declareProcess1(
+ DeclarationContext context, Context ctx, Collector<OUT> out)
+ throws DeclarationException;
+
+ /**
+ * Declare a process for one element from the second of the connected streams.
+ *
+ * <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 context the context that provides useful methods to define named callbacks.
+ * @param ctx A {@link Context} that allows querying the timestamp of the element and getting a
+ * {@link TimerService} for registering timers and querying the time. The context is only
+ * valid during the invocation of this method, do not store it.
+ * @param out The collector for returning result values.
+ * @return the whole processing logic just like {@code processElement}.
+ */
+ public abstract ThrowingConsumer<IN2, Exception> declareProcess2(
+ DeclarationContext context, Context ctx, Collector<OUT> out)
+ throws DeclarationException;
+
+ /**
+ * Declare a procedure which is called when a timer set using {@link TimerService} fires.
Review Comment:
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006034494
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/functions/DeclaringAsyncKeyedCoProcessFunction.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.flink.runtime.asyncprocessing.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationException;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+/**
+ * A function that processes elements of two keyed streams and produces a single output one.
+ *
+ * <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.
+ *
+ * <p>An example use-case for connected streams would be the application of a set of rules that
+ * 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
+ * register a timer that will trigger an action in the future.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input.
+ * @param <IN2> Type of the second input.
+ * @param <OUT> Output type.
+ */
+@Internal
+public abstract class DeclaringAsyncKeyedCoProcessFunction<K, IN1, IN2, OUT>
+ extends KeyedCoProcessFunction<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This method is called for each element in the first of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * 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.");
+ }
+
+ /**
+ * This method is called for each element in the second of the connected streams.
+ *
+ * <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
+ * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
+ * {@link TimeDomain} of the firing timer and getting a {@link TimerService} for registering
+ * 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
+ * and go into recovery.
+ */
+ public void processElement2(IN2 value, Context ctx, Collector<OUT> out) throws Exception {
+ throw new IllegalAccessException("This method is replaced by declareProcess2.");
+ }
+
+ /**
+ * Called when a timer set using {@link TimerService} fires.
+ *
+ * @param timestamp The timestamp of the firing timer.
+ * @param ctx An {@link OnTimerContext} that allows querying the timestamp of the firing timer,
+ * querying the {@link TimeDomain} of the firing timer and getting a {@link TimerService}
+ * for registering timers and querying the time. The context is only valid during the
+ * invocation of this method, do not store it.
+ * @param out The collector for returning result values.
+ * @throws Exception This method may throw exceptions. Throwing an exception will cause the
+ * operation to fail and may trigger recovery.
+ */
+ public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception {}
+
+ /**
+ * Information available in an invocation of {@link #processElement1(Object, Context,
+ * Collector)}/ {@link #processElement2(Object, Context, Collector)} or {@link #onTimer(long,
+ * OnTimerContext, Collector)}.
+ */
+
+ /**
+ * Declaring variables before {@link #declareProcess1} and {@link #declareProcess2} and {@link
+ * #declareOnTimer}.
+ */
+ public void declareVariables(DeclarationContext context) {}
+
+ /**
+ * Declare a process for one element from the first of the connected streams.
+ *
+ * <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 context the context that provides useful methods to define named callbacks.
+ * @param ctx A {@link Context} that allows querying the timestamp of the element and getting a
+ * {@link TimerService} for registering timers and querying the time. The context is only
+ * valid during the invocation of this method, do not store it.
+ * @param out The collector for returning result values.
+ * @return the whole processing logic just like {@code processElement}.
+ */
+ public abstract ThrowingConsumer<IN1, Exception> declareProcess1(
+ DeclarationContext context, Context ctx, Collector<OUT> out)
+ throws DeclarationException;
+
+ /**
+ * Declare a process for one element from the second of the connected streams.
+ *
+ * <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 context the context that provides useful methods to define named callbacks.
+ * @param ctx A {@link Context} that allows querying the timestamp of the element and getting a
+ * {@link TimerService} for registering timers and querying the time. The context is only
+ * valid during the invocation of this method, do not store it.
+ * @param out The collector for returning result values.
+ * @return the whole processing logic just like {@code processElement}.
+ */
+ public abstract ThrowingConsumer<IN2, Exception> declareProcess2(
+ DeclarationContext context, Context ctx, Collector<OUT> out)
+ throws DeclarationException;
+
+ /**
+ * Declare a procedure which is called when a timer set using {@link TimerService} fires.
Review Comment:
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37521][runtime] Introduce KeyedCoProcessOperator with async state API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26328:
URL: https://github.com/apache/flink/pull/26328#discussion_r2006050048
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/co/AsyncKeyedCoProcessOperator.java:
##########
@@ -0,0 +1,237 @@
+/*
+ * 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.flink.runtime.asyncprocessing.operators.co;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclarationContext;
+import org.apache.flink.runtime.asyncprocessing.declare.DeclaredVariable;
+import org.apache.flink.runtime.asyncprocessing.functions.DeclaringAsyncKeyedCoProcessFunction;
+import org.apache.flink.runtime.asyncprocessing.operators.AbstractAsyncStateUdfStreamOperator;
+import org.apache.flink.runtime.asyncprocessing.operators.TimestampedCollectorWithDeclaredVariable;
+import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.runtime.state.VoidNamespaceSerializer;
+import org.apache.flink.streaming.api.SimpleTimerService;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.streaming.api.operators.InternalTimer;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.streaming.api.operators.Triggerable;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing keyed {@link
+ * KeyedCoProcessFunction KeyedCoProcessFunction}.
+ */
+@Internal
+public class AsyncKeyedCoProcessOperator<K, IN1, IN2, OUT>
+ extends AbstractAsyncStateUdfStreamOperator<OUT, KeyedCoProcessFunction<K, IN1, IN2, OUT>>
+ implements TwoInputStreamOperator<IN1, IN2, OUT>, Triggerable<K, VoidNamespace> {
+
+ private static final long serialVersionUID = 1L;
+
+ // Shared timestamp variable for collector, context and onTimerContext.
+ private transient DeclaredVariable<Long> sharedTimestamp;
+
+ private transient TimestampedCollectorWithDeclaredVariable<OUT> collector;
+
+ private transient ContextImpl<K, IN1, IN2, OUT> context;
+
+ private transient OnTimerContextImpl<K, IN1, IN2, OUT> onTimerContext;
+
+ private transient ThrowingConsumer<IN1, Exception> processor1;
+ private transient ThrowingConsumer<IN2, Exception> processor2;
+ private transient ThrowingConsumer<Long, Exception> timerProcessor;
+
+ public AsyncKeyedCoProcessOperator(
+ KeyedCoProcessFunction<K, IN1, IN2, OUT> keyedCoProcessFunction) {
+ super(keyedCoProcessFunction);
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ sharedTimestamp =
+ declarationContext.declareVariable(
+ LongSerializer.INSTANCE,
+ "_AsyncCoKeyedProcessOperator$sharedTimestamp",
+ null);
+
+ collector = new TimestampedCollectorWithDeclaredVariable<>(output, sharedTimestamp);
+
+ InternalTimerService<VoidNamespace> internalTimerService =
+ getInternalTimerService("user-timers", VoidNamespaceSerializer.INSTANCE, this);
+
+ TimerService timerService = new SimpleTimerService(internalTimerService);
+
+ context = new ContextImpl<>(userFunction, timerService, sharedTimestamp);
+ onTimerContext = new OnTimerContextImpl<>(userFunction, timerService, declarationContext);
+ if (userFunction instanceof DeclaringAsyncKeyedCoProcessFunction) {
+ DeclaringAsyncKeyedCoProcessFunction declaringFunction =
+ (DeclaringAsyncKeyedCoProcessFunction) userFunction;
+ declaringFunction.declareVariables(declarationContext);
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org