You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2024/05/22 04:04:36 UTC
(flink) 01/11: [hotfix] Temporarily revert the change that makes the v2 operator inherit from the async operator
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4d7aebaed1d4ca693eefa49ecb3bef87e87d0b5c
Author: Weijie Guo <re...@163.com>
AuthorDate: Wed May 15 16:59:27 2024 +0800
[hotfix] Temporarily revert the change that makes the v2 operator inherit from the async operator
This revert FLINK-35156.
The async operator currently has imperfect support for state and timer, and needs to wait until all the related PRs are merged before it can be used as the base class of v2 operators
---
.../org/apache/flink/datastream/impl/operators/ProcessOperator.java | 4 ++--
.../datastream/impl/operators/TwoInputBroadcastProcessOperator.java | 4 ++--
.../impl/operators/TwoInputNonBroadcastProcessOperator.java | 4 ++--
.../flink/datastream/impl/operators/TwoOutputProcessOperator.java | 4 ++--
4 files changed, 8 insertions(+), 8 deletions(-)
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/ProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/ProcessOperator.java
index fb773e271e1..6346079cc21 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/ProcessOperator.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/ProcessOperator.java
@@ -23,15 +23,15 @@ import org.apache.flink.datastream.impl.common.OutputCollector;
import org.apache.flink.datastream.impl.common.TimestampCollector;
import org.apache.flink.datastream.impl.context.DefaultNonPartitionedContext;
import org.apache.flink.datastream.impl.context.DefaultRuntimeContext;
+import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-import org.apache.flink.streaming.runtime.operators.asyncprocessing.AbstractAsyncStateUdfStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
/** Operator for {@link OneInputStreamProcessFunction}. */
public class ProcessOperator<IN, OUT>
- extends AbstractAsyncStateUdfStreamOperator<OUT, OneInputStreamProcessFunction<IN, OUT>>
+ extends AbstractUdfStreamOperator<OUT, OneInputStreamProcessFunction<IN, OUT>>
implements OneInputStreamOperator<IN, OUT>, BoundedOneInput {
protected transient DefaultRuntimeContext context;
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputBroadcastProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputBroadcastProcessOperator.java
index 0ba5b986fe5..5c0faf05bab 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputBroadcastProcessOperator.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputBroadcastProcessOperator.java
@@ -23,17 +23,17 @@ import org.apache.flink.datastream.impl.common.OutputCollector;
import org.apache.flink.datastream.impl.common.TimestampCollector;
import org.apache.flink.datastream.impl.context.DefaultNonPartitionedContext;
import org.apache.flink.datastream.impl.context.DefaultRuntimeContext;
+import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedMultiInput;
import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
-import org.apache.flink.streaming.runtime.operators.asyncprocessing.AbstractAsyncStateUdfStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import static org.apache.flink.util.Preconditions.checkState;
/** Operator for {@link TwoInputBroadcastStreamProcessFunction}. */
public class TwoInputBroadcastProcessOperator<IN1, IN2, OUT>
- extends AbstractAsyncStateUdfStreamOperator<
+ extends AbstractUdfStreamOperator<
OUT, TwoInputBroadcastStreamProcessFunction<IN1, IN2, OUT>>
implements TwoInputStreamOperator<IN1, IN2, OUT>, BoundedMultiInput {
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputNonBroadcastProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputNonBroadcastProcessOperator.java
index e4f02a10dcb..60d88682c2d 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputNonBroadcastProcessOperator.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputNonBroadcastProcessOperator.java
@@ -23,17 +23,17 @@ import org.apache.flink.datastream.impl.common.OutputCollector;
import org.apache.flink.datastream.impl.common.TimestampCollector;
import org.apache.flink.datastream.impl.context.DefaultNonPartitionedContext;
import org.apache.flink.datastream.impl.context.DefaultRuntimeContext;
+import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedMultiInput;
import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
-import org.apache.flink.streaming.runtime.operators.asyncprocessing.AbstractAsyncStateUdfStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import static org.apache.flink.util.Preconditions.checkState;
/** Operator for {@link TwoInputNonBroadcastStreamProcessFunction}. */
public class TwoInputNonBroadcastProcessOperator<IN1, IN2, OUT>
- extends AbstractAsyncStateUdfStreamOperator<
+ extends AbstractUdfStreamOperator<
OUT, TwoInputNonBroadcastStreamProcessFunction<IN1, IN2, OUT>>
implements TwoInputStreamOperator<IN1, IN2, OUT>, BoundedMultiInput {
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoOutputProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoOutputProcessOperator.java
index 91ef5230b3a..2eb7dee587e 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoOutputProcessOperator.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoOutputProcessOperator.java
@@ -24,11 +24,11 @@ import org.apache.flink.datastream.impl.common.OutputCollector;
import org.apache.flink.datastream.impl.common.TimestampCollector;
import org.apache.flink.datastream.impl.context.DefaultRuntimeContext;
import org.apache.flink.datastream.impl.context.DefaultTwoOutputNonPartitionedContext;
+import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.Output;
-import org.apache.flink.streaming.runtime.operators.asyncprocessing.AbstractAsyncStateUdfStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.util.OutputTag;
@@ -38,7 +38,7 @@ import org.apache.flink.util.OutputTag;
* <p>We support the second output via flink side-output mechanism.
*/
public class TwoOutputProcessOperator<IN, OUT_MAIN, OUT_SIDE>
- extends AbstractAsyncStateUdfStreamOperator<
+ extends AbstractUdfStreamOperator<
OUT_MAIN, TwoOutputStreamProcessFunction<IN, OUT_MAIN, OUT_SIDE>>
implements OneInputStreamOperator<IN, OUT_MAIN>, BoundedOneInput {
protected transient TimestampCollector<OUT_MAIN> mainCollector;