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 2025/01/18 10:30:42 UTC
(flink) 02/02: [FLINK-37136][runtime] Implement Window extension for DataStream V2
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 f2460c889dd2199ba59f12c9457a0d2a44d6f915
Author: Xu Huang <hu...@gmail.com>
AuthorDate: Sat Jan 18 02:33:52 2025 +0800
[FLINK-37136][runtime] Implement Window extension for DataStream V2
---
.../java/org/apache/flink/util/TaggedUnion.java | 76 ++
.../impl/builtin/BuiltinWindowFuncs.java | 94 +++
.../TwoInputNonBroadcastJoinProcessOperator.java | 2 +-
.../context/DefaultOneInputWindowContext.java | 139 ++++
.../context/DefaultTwoInputWindowContext.java | 155 +++++
.../extension/window/context/WindowStateStore.java | 264 +++++++
.../window/context/WindowTriggerContext.java | 190 +++++
...nternalOneInputWindowStreamProcessFunction.java | 109 +++
...nternalTwoInputWindowStreamProcessFunction.java | 120 ++++
...ternalTwoOutputWindowStreamProcessFunction.java | 114 +++
.../window/operators/MergingWindowSet.java | 246 +++++++
.../operators/OneInputWindowProcessOperator.java | 532 ++++++++++++++
.../TwoInputNonBroadcastWindowProcessOperator.java | 763 +++++++++++++++++++++
.../operators/TwoOutputWindowProcessOperator.java | 549 +++++++++++++++
.../impl/extension/window/utils/WindowUtils.java | 230 +++++++
...Operator.java => BaseKeyedProcessOperator.java} | 55 +-
...eKeyedTwoInputNonBroadcastProcessOperator.java} | 57 +-
...java => BaseKeyedTwoOutputProcessOperator.java} | 66 +-
.../impl/operators/KeyedProcessOperator.java | 56 +-
.../KeyedTwoInputNonBroadcastProcessOperator.java | 63 +-
.../operators/KeyedTwoOutputProcessOperator.java | 72 +-
.../datastream/impl/stream/GlobalStreamImpl.java | 111 ++-
.../impl/stream/KeyedPartitionStreamImpl.java | 141 +++-
.../impl/stream/NonKeyedPartitionStreamImpl.java | 111 ++-
.../flink/datastream/impl/utils/StreamUtils.java | 224 ++++++
.../AbstractAsyncStateStreamOperator.java | 2 +-
.../streaming/api/datastream/CoGroupedStreams.java | 54 +-
.../streaming/api/datastream/JoinedStreams.java | 2 +-
.../api/operators/AbstractStreamOperator.java | 2 +-
.../api/datastream/UnionSerializerTest.java | 2 +-
.../api/datastream/UnionSerializerUpgradeTest.java | 2 +-
.../datastream/extension/window/WindowITCase.java | 620 +++++++++++++++++
32 files changed, 4750 insertions(+), 473 deletions(-)
diff --git a/flink-core-api/src/main/java/org/apache/flink/util/TaggedUnion.java b/flink-core-api/src/main/java/org/apache/flink/util/TaggedUnion.java
new file mode 100644
index 00000000000..2d24d0f9bb9
--- /dev/null
+++ b/flink-core-api/src/main/java/org/apache/flink/util/TaggedUnion.java
@@ -0,0 +1,76 @@
+/*
+ * 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.util;
+
+import org.apache.flink.annotation.Internal;
+
+import java.util.Objects;
+
+/**
+ * Utility class for implementing CoGroupedStream in DataStream V1, as well as two-input window
+ * operations in DataStream V2.
+ */
+@Internal
+public class TaggedUnion<T1, T2> {
+ private final T1 one;
+ private final T2 two;
+
+ private TaggedUnion(T1 one, T2 two) {
+ this.one = one;
+ this.two = two;
+ }
+
+ public boolean isOne() {
+ return one != null;
+ }
+
+ public boolean isTwo() {
+ return two != null;
+ }
+
+ public T1 getOne() {
+ return one;
+ }
+
+ public T2 getTwo() {
+ return two;
+ }
+
+ public static <T1, T2> TaggedUnion<T1, T2> one(T1 one) {
+ return new TaggedUnion<>(one, null);
+ }
+
+ public static <T1, T2> TaggedUnion<T1, T2> two(T2 two) {
+ return new TaggedUnion<>(null, two);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this) {
+ return true;
+ }
+
+ if (!(obj instanceof TaggedUnion)) {
+ return false;
+ }
+
+ TaggedUnion<?, ?> other = (TaggedUnion<?, ?>) obj;
+ return Objects.equals(one, other.one) && Objects.equals(two, other.two);
+ }
+}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/builtin/BuiltinWindowFuncs.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/builtin/BuiltinWindowFuncs.java
new file mode 100644
index 00000000000..48bf5f0e704
--- /dev/null
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/builtin/BuiltinWindowFuncs.java
@@ -0,0 +1,94 @@
+/*
+ * 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.datastream.impl.builtin;
+
+import org.apache.flink.datastream.api.builtin.BuiltinFuncs;
+import org.apache.flink.datastream.api.extension.window.function.OneInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.api.extension.window.function.TwoInputNonBroadcastWindowStreamProcessFunction;
+import org.apache.flink.datastream.api.extension.window.function.TwoOutputWindowStreamProcessFunction;
+import org.apache.flink.datastream.api.extension.window.strategy.WindowStrategy;
+import org.apache.flink.datastream.api.function.OneInputStreamProcessFunction;
+import org.apache.flink.datastream.api.function.TwoInputNonBroadcastStreamProcessFunction;
+import org.apache.flink.datastream.api.function.TwoOutputStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.function.InternalOneInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.function.InternalTwoInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.function.InternalTwoOutputWindowStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.utils.WindowUtils;
+import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
+import org.apache.flink.streaming.api.windowing.windows.Window;
+import org.apache.flink.util.TaggedUnion;
+
+/** Window-related implementations in {@link BuiltinFuncs}. */
+public class BuiltinWindowFuncs {
+
+ /**
+ * Wrap the WindowStrategy and OneInputWindowStreamProcessFunction within a ProcessFunction to
+ * perform the window operation.
+ */
+ public static <IN, OUT, W extends Window> OneInputStreamProcessFunction<IN, OUT> window(
+ WindowStrategy windowStrategy,
+ OneInputWindowStreamProcessFunction<IN, OUT> windowProcessFunction) {
+ WindowAssigner<IN, W> windowAssigner =
+ (WindowAssigner<IN, W>) WindowUtils.createWindowAssigner(windowStrategy);
+ return new InternalOneInputWindowStreamProcessFunction<>(
+ windowProcessFunction,
+ windowAssigner,
+ windowAssigner.getDefaultTrigger(),
+ WindowUtils.getAllowedLateness(windowStrategy),
+ windowStrategy);
+ }
+
+ /**
+ * Wrap the WindowStrategy and TwoInputWindowStreamProcessFunction within a ProcessFunction to
+ * perform the window operation.
+ */
+ public static <IN1, IN2, OUT, W extends Window>
+ TwoInputNonBroadcastStreamProcessFunction<IN1, IN2, OUT> window(
+ WindowStrategy windowStrategy,
+ TwoInputNonBroadcastWindowStreamProcessFunction<IN1, IN2, OUT>
+ windowProcessFunction) {
+ WindowAssigner<TaggedUnion<IN1, IN2>, W> windowAssigner =
+ (WindowAssigner<TaggedUnion<IN1, IN2>, W>)
+ WindowUtils.createWindowAssigner(windowStrategy);
+ return new InternalTwoInputWindowStreamProcessFunction<>(
+ windowProcessFunction,
+ windowAssigner,
+ windowAssigner.getDefaultTrigger(),
+ WindowUtils.getAllowedLateness(windowStrategy),
+ windowStrategy);
+ }
+
+ /**
+ * Wrap the WindowStrategy and TwoOutputWindowStreamProcessFunction within a ProcessFunction to
+ * perform the window operation.
+ */
+ public static <IN, OUT1, OUT2, W extends Window>
+ TwoOutputStreamProcessFunction<IN, OUT1, OUT2> window(
+ WindowStrategy windowStrategy,
+ TwoOutputWindowStreamProcessFunction<IN, OUT1, OUT2> windowProcessFunction) {
+ WindowAssigner<IN, W> windowAssigner =
+ (WindowAssigner<IN, W>) WindowUtils.createWindowAssigner(windowStrategy);
+ return new InternalTwoOutputWindowStreamProcessFunction<>(
+ windowProcessFunction,
+ windowAssigner,
+ windowAssigner.getDefaultTrigger(),
+ WindowUtils.getAllowedLateness(windowStrategy),
+ windowStrategy);
+ }
+}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/join/operators/TwoInputNonBroadcastJoinProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/join/operators/TwoInputNonBroadcastJoinProcessOperator.java
index 135ca15b938..ae33b2bfc54 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/join/operators/TwoInputNonBroadcastJoinProcessOperator.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/join/operators/TwoInputNonBroadcastJoinProcessOperator.java
@@ -19,13 +19,13 @@
package org.apache.flink.datastream.impl.extension.join.operators;
import org.apache.flink.api.common.state.v2.ListState;
+import org.apache.flink.api.common.state.v2.ListStateDescriptor;
import org.apache.flink.datastream.api.extension.join.JoinType;
import org.apache.flink.datastream.api.function.TwoInputNonBroadcastStreamProcessFunction;
import org.apache.flink.datastream.api.stream.KeyedPartitionStream;
import org.apache.flink.datastream.impl.operators.KeyedTwoInputNonBroadcastProcessOperator;
import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.runtime.state.VoidNamespaceSerializer;
-import org.apache.flink.runtime.state.v2.ListStateDescriptor;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import static org.apache.flink.util.Preconditions.checkArgument;
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/context/DefaultOneInputWindowContext.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/context/DefaultOneInputWindowContext.java
new file mode 100644
index 00000000000..0891b02958d
--- /dev/null
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/context/DefaultOneInputWindowContext.java
@@ -0,0 +1,139 @@
+/*
+ * 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.datastream.impl.extension.window.context;
+
+import org.apache.flink.api.common.state.AggregatingStateDeclaration;
+import org.apache.flink.api.common.state.ListStateDeclaration;
+import org.apache.flink.api.common.state.MapStateDeclaration;
+import org.apache.flink.api.common.state.ReducingStateDeclaration;
+import org.apache.flink.api.common.state.ValueStateDeclaration;
+import org.apache.flink.api.common.state.v2.AggregatingState;
+import org.apache.flink.api.common.state.v2.AppendingState;
+import org.apache.flink.api.common.state.v2.ListState;
+import org.apache.flink.api.common.state.v2.MapState;
+import org.apache.flink.api.common.state.v2.ReducingState;
+import org.apache.flink.api.common.state.v2.StateIterator;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.datastream.api.extension.window.context.OneInputWindowContext;
+import org.apache.flink.datastream.api.extension.window.function.WindowProcessFunction;
+import org.apache.flink.runtime.asyncprocessing.operators.AbstractAsyncStateStreamOperator;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+import org.apache.flink.streaming.api.windowing.windows.Window;
+
+import javax.annotation.Nullable;
+
+import java.util.Optional;
+
+/**
+ * Default implementation of the {@link OneInputWindowContext}.
+ *
+ * @param <K> Type of the window key.
+ * @param <IN> Type of the input elements.
+ * @param <W> Type of the window.
+ */
+public class DefaultOneInputWindowContext<K, IN, W extends Window>
+ implements OneInputWindowContext<IN> {
+
+ /**
+ * The current processing window. An instance should be set every time before accessing
+ * window-related attributes, data, and state.
+ */
+ @Nullable private W window;
+
+ /** Use to retrieve state associated with windows. */
+ private final WindowStateStore<K, W> windowStateStore;
+
+ /** The state utilized for storing window data. */
+ private final AppendingState<IN, StateIterator<IN>, Iterable<IN>> windowState;
+
+ public DefaultOneInputWindowContext(
+ @Nullable W window,
+ AppendingState<IN, StateIterator<IN>, Iterable<IN>> windowState,
+ WindowProcessFunction windowProcessFunction,
+ AbstractAsyncStateStreamOperator<?> operator,
+ TypeSerializer<W> windowSerializer,
+ boolean isMergingWindow) {
+ this.window = window;
+ this.windowState = windowState;
+ this.windowStateStore =
+ new WindowStateStore<>(
+ windowProcessFunction, operator, windowSerializer, isMergingWindow);
+ }
+
+ public void setWindow(W window) {
+ this.window = window;
+ }
+
+ @Override
+ public long getStartTime() {
+ if (window instanceof TimeWindow) {
+ return ((TimeWindow) window).getStart();
+ }
+ return -1;
+ }
+
+ @Override
+ public long getEndTime() {
+ if (window instanceof TimeWindow) {
+ return ((TimeWindow) window).getEnd();
+ }
+ return -1;
+ }
+
+ @Override
+ public <T> Optional<ListState<T>> getWindowState(ListStateDeclaration<T> stateDeclaration)
+ throws Exception {
+ return windowStateStore.getWindowState(stateDeclaration, window);
+ }
+
+ @Override
+ public <KEY, V> Optional<MapState<KEY, V>> getWindowState(
+ MapStateDeclaration<KEY, V> stateDeclaration) throws Exception {
+ return windowStateStore.getWindowState(stateDeclaration, window);
+ }
+
+ @Override
+ public <T> Optional<ValueState<T>> getWindowState(ValueStateDeclaration<T> stateDeclaration)
+ throws Exception {
+ return windowStateStore.getWindowState(stateDeclaration, window);
+ }
+
+ @Override
+ public <T> Optional<ReducingState<T>> getWindowState(
+ ReducingStateDeclaration<T> stateDeclaration) throws Exception {
+ return windowStateStore.getWindowState(stateDeclaration, window);
+ }
+
+ @Override
+ public <T, ACC, OUT> Optional<AggregatingState<T, OUT>> getWindowState(
+ AggregatingStateDeclaration<T, ACC, OUT> stateDeclaration) throws Exception {
+ return windowStateStore.getWindowState(stateDeclaration, window);
+ }
+
+ @Override
+ public void putRecord(IN record) {
+ windowState.add(record);
+ }
+
+ @Override
+ public Iterable<IN> getAllRecords() {
+ return windowState.get();
+ }
+}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/context/DefaultTwoInputWindowContext.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/context/DefaultTwoInputWindowContext.java
new file mode 100644
index 00000000000..a6bc616a5aa
--- /dev/null
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/context/DefaultTwoInputWindowContext.java
@@ -0,0 +1,155 @@
+/*
+ * 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.datastream.impl.extension.window.context;
+
+import org.apache.flink.api.common.state.AggregatingStateDeclaration;
+import org.apache.flink.api.common.state.ListStateDeclaration;
+import org.apache.flink.api.common.state.MapStateDeclaration;
+import org.apache.flink.api.common.state.ReducingStateDeclaration;
+import org.apache.flink.api.common.state.ValueStateDeclaration;
+import org.apache.flink.api.common.state.v2.AggregatingState;
+import org.apache.flink.api.common.state.v2.AppendingState;
+import org.apache.flink.api.common.state.v2.ListState;
+import org.apache.flink.api.common.state.v2.MapState;
+import org.apache.flink.api.common.state.v2.ReducingState;
+import org.apache.flink.api.common.state.v2.StateIterator;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.datastream.api.extension.window.context.TwoInputWindowContext;
+import org.apache.flink.datastream.api.extension.window.function.WindowProcessFunction;
+import org.apache.flink.runtime.asyncprocessing.operators.AbstractAsyncStateStreamOperator;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+import org.apache.flink.streaming.api.windowing.windows.Window;
+
+import javax.annotation.Nullable;
+
+import java.util.Optional;
+
+/**
+ * Default implementation of the {@link TwoInputWindowContext}.
+ *
+ * @param <K> Type of the key.
+ * @param <IN1> Type of the first input elements.
+ * @param <IN2> Type of the second input elements.
+ * @param <W> Type of the window.
+ */
+public class DefaultTwoInputWindowContext<K, IN1, IN2, W extends Window>
+ implements TwoInputWindowContext<IN1, IN2> {
+
+ /**
+ * The current processing window. An instance should be set every time before accessing
+ * window-related attributes, data, and state.
+ */
+ @Nullable private W window;
+
+ /** Use to retrieve state associated with windows. */
+ private final WindowStateStore<K, W> windowStateStore;
+
+ /** The state utilized for storing window received input data from first input. */
+ private final AppendingState<IN1, StateIterator<IN1>, Iterable<IN1>> leftWindowState;
+
+ /** The state utilized for storing window received input data from second input. */
+ private final AppendingState<IN2, StateIterator<IN2>, Iterable<IN2>> rightWindowState;
+
+ public DefaultTwoInputWindowContext(
+ W window,
+ AppendingState<IN1, StateIterator<IN1>, Iterable<IN1>> leftWindowState,
+ AppendingState<IN2, StateIterator<IN2>, Iterable<IN2>> rightWindowState,
+ WindowProcessFunction windowProcessFunction,
+ AbstractAsyncStateStreamOperator<?> operator,
+ TypeSerializer<W> windowSerializer,
+ boolean isMergingWindow) {
+ this.window = window;
+ this.leftWindowState = leftWindowState;
+ this.rightWindowState = rightWindowState;
+ this.windowStateStore =
+ new WindowStateStore<>(
+ windowProcessFunction, operator, windowSerializer, isMergingWindow);
+ }
+
+ public void setWindow(W window) {
+ this.window = window;
+ }
+
+ @Override
+ public long getStartTime() {
+ if (window instanceof TimeWindow) {
+ return ((TimeWindow) window).getStart();
+ }
+ return -1;
+ }
+
+ @Override
+ public long getEndTime() {
+ if (window instanceof TimeWindow) {
+ return ((TimeWindow) window).getEnd();
+ }
+ return -1;
+ }
+
+ @Override
+ public <T> Optional<ListState<T>> getWindowState(ListStateDeclaration<T> stateDeclaration)
+ throws Exception {
+ return windowStateStore.getWindowState(stateDeclaration, window);
+ }
+
+ @Override
+ public <KEY, V> Optional<MapState<KEY, V>> getWindowState(
+ MapStateDeclaration<KEY, V> stateDeclaration) throws Exception {
+ return windowStateStore.getWindowState(stateDeclaration, window);
+ }
+
+ @Override
+ public <T> Optional<ValueState<T>> getWindowState(ValueStateDeclaration<T> stateDeclaration)
+ throws Exception {
+ return windowStateStore.getWindowState(stateDeclaration, window);
+ }
+
+ @Override
+ public <T> Optional<ReducingState<T>> getWindowState(
+ ReducingStateDeclaration<T> stateDeclaration) throws Exception {
+ return windowStateStore.getWindowState(stateDeclaration, window);
+ }
+
+ @Override
+ public <T, ACC, OUT> Optional<AggregatingState<T, OUT>> getWindowState(
+ AggregatingStateDeclaration<T, ACC, OUT> stateDeclaration) throws Exception {
+ return windowStateStore.getWindowState(stateDeclaration, window);
+ }
+
+ @Override
+ public void putRecord1(IN1 record) {
+ leftWindowState.add(record);
+ }
+
+ @Override
+ public Iterable<IN1> getAllRecords1() {
+ return leftWindowState.get();
+ }
+
+ @Override
+ public void putRecord2(IN2 record) {
+ rightWindowState.add(record);
+ }
+
+ @Override
+ public Iterable<IN2> getAllRecords2() {
+ return rightWindowState.get();
+ }
+}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/context/WindowStateStore.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/context/WindowStateStore.java
new file mode 100644
index 00000000000..695ad743c85
--- /dev/null
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/context/WindowStateStore.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.datastream.impl.extension.window.context;
+
+import org.apache.flink.api.common.state.AggregatingStateDeclaration;
+import org.apache.flink.api.common.state.ListStateDeclaration;
+import org.apache.flink.api.common.state.MapStateDeclaration;
+import org.apache.flink.api.common.state.ReducingStateDeclaration;
+import org.apache.flink.api.common.state.StateDeclaration;
+import org.apache.flink.api.common.state.ValueStateDeclaration;
+import org.apache.flink.api.common.state.v2.AggregatingState;
+import org.apache.flink.api.common.state.v2.AggregatingStateDescriptor;
+import org.apache.flink.api.common.state.v2.ListState;
+import org.apache.flink.api.common.state.v2.ListStateDescriptor;
+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.ReducingState;
+import org.apache.flink.api.common.state.v2.ReducingStateDescriptor;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.api.common.state.v2.ValueStateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.datastream.api.extension.window.context.WindowContext;
+import org.apache.flink.datastream.api.extension.window.function.WindowProcessFunction;
+import org.apache.flink.runtime.asyncprocessing.operators.AbstractAsyncStateStreamOperator;
+import org.apache.flink.runtime.state.v2.adaptor.AggregatingStateAdaptor;
+import org.apache.flink.runtime.state.v2.adaptor.ListStateAdaptor;
+import org.apache.flink.runtime.state.v2.adaptor.MapStateAdaptor;
+import org.apache.flink.runtime.state.v2.adaptor.ReducingStateAdaptor;
+import org.apache.flink.runtime.state.v2.adaptor.ValueStateAdaptor;
+import org.apache.flink.streaming.api.windowing.windows.Window;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * This class provides methods to store and retrieve state associated with windows in {@link
+ * WindowContext}.
+ *
+ * @param <K> Type of the window key.
+ * @param <W> Type of the window.
+ */
+public class WindowStateStore<K, W extends Window> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WindowStateStore.class);
+
+ /**
+ * The {@link StateDeclaration}s that have been declared by the {@link
+ * WindowProcessFunction#useWindowStates()}.
+ */
+ private final Set<StateDeclaration> windowStateDeclarations;
+
+ /** The operator to which the window belongs, used for creating and retrieving window state. */
+ private final AbstractAsyncStateStreamOperator<?> operator;
+
+ private final TypeSerializer<W> windowSerializer;
+
+ /** Whether the window is a merging window. */
+ private final boolean isMergingWindow;
+
+ public WindowStateStore(
+ WindowProcessFunction windowProcessFunction,
+ AbstractAsyncStateStreamOperator<?> operator,
+ TypeSerializer<W> windowSerializer,
+ boolean isMergingWindow) {
+ this.windowStateDeclarations = windowProcessFunction.useWindowStates();
+ this.operator = operator;
+ this.windowSerializer = windowSerializer;
+ this.isMergingWindow = isMergingWindow;
+ }
+
+ private boolean isStateDeclared(StateDeclaration stateDeclaration) {
+ if (!windowStateDeclarations.contains(stateDeclaration)) {
+ LOG.warn(
+ "Fail to get window state for "
+ + stateDeclaration.getName()
+ + ", please declare the used state in the `WindowProcessFunction#useWindowStates` method first.");
+ return false;
+ }
+ return true;
+ }
+
+ private boolean stateRedistributionModeIsNotNone(StateDeclaration stateDeclaration) {
+ StateDeclaration.RedistributionMode redistributionMode =
+ stateDeclaration.getRedistributionMode();
+ return redistributionMode != StateDeclaration.RedistributionMode.NONE;
+ }
+
+ /** Retrieve window state of list type. */
+ public <T> Optional<ListState<T>> getWindowState(
+ ListStateDeclaration<T> stateDeclaration, W namespace) throws Exception {
+ checkState(
+ !isMergingWindow,
+ "Retrieving the window state is not permitted when using merging windows, such as session windows.");
+
+ if (!isStateDeclared(stateDeclaration)) {
+ return Optional.empty();
+ }
+
+ if (stateRedistributionModeIsNotNone(stateDeclaration)) {
+ throw new UnsupportedOperationException(
+ "RedistributionMode "
+ + stateDeclaration.getRedistributionMode().name()
+ + " is not supported for window state.");
+ }
+
+ ListStateDescriptor<T> stateDescriptor =
+ new ListStateDescriptor<>(
+ stateDeclaration.getName(),
+ TypeExtractor.createTypeInfo(
+ stateDeclaration.getTypeDescriptor().getTypeClass()));
+
+ ListStateAdaptor<K, W, T> state =
+ operator.getOrCreateKeyedState(namespace, windowSerializer, stateDescriptor);
+ state.setCurrentNamespace(namespace);
+ return Optional.of(state);
+ }
+
+ /** Retrieve window state of map type. */
+ public <KEY, V> Optional<MapState<KEY, V>> getWindowState(
+ MapStateDeclaration<KEY, V> stateDeclaration, W namespace) throws Exception {
+ checkState(
+ !isMergingWindow,
+ "Retrieving the window state is not permitted when using merging windows, such as session windows.");
+
+ if (!isStateDeclared(stateDeclaration)) {
+ return Optional.empty();
+ }
+
+ if (stateRedistributionModeIsNotNone(stateDeclaration)) {
+ throw new UnsupportedOperationException(
+ "RedistributionMode "
+ + stateDeclaration.getRedistributionMode().name()
+ + " is not supported for window state.");
+ }
+
+ MapStateDescriptor<KEY, V> stateDescriptor =
+ new MapStateDescriptor<>(
+ stateDeclaration.getName(),
+ TypeExtractor.createTypeInfo(
+ stateDeclaration.getKeyTypeDescriptor().getTypeClass()),
+ TypeExtractor.createTypeInfo(
+ stateDeclaration.getValueTypeDescriptor().getTypeClass()));
+
+ MapStateAdaptor<K, W, KEY, V> state =
+ operator.getOrCreateKeyedState(namespace, windowSerializer, stateDescriptor);
+ state.setCurrentNamespace(namespace);
+ return Optional.of(state);
+ }
+
+ /** Retrieve window state of value type. */
+ public <T> Optional<ValueState<T>> getWindowState(
+ ValueStateDeclaration<T> stateDeclaration, W namespace) throws Exception {
+ checkState(
+ !isMergingWindow,
+ "Retrieving the window state is not permitted when using merging windows, such as session windows.");
+
+ if (!isStateDeclared(stateDeclaration)) {
+ return Optional.empty();
+ }
+
+ if (stateRedistributionModeIsNotNone(stateDeclaration)) {
+ throw new UnsupportedOperationException(
+ "RedistributionMode "
+ + stateDeclaration.getRedistributionMode().name()
+ + " is not supported for window state.");
+ }
+
+ ValueStateDescriptor<T> stateDescriptor =
+ new ValueStateDescriptor<>(
+ stateDeclaration.getName(),
+ TypeExtractor.createTypeInfo(
+ stateDeclaration.getTypeDescriptor().getTypeClass()));
+
+ ValueStateAdaptor<K, W, T> state =
+ operator.getOrCreateKeyedState(namespace, windowSerializer, stateDescriptor);
+ state.setCurrentNamespace(namespace);
+ return Optional.of(state);
+ }
+
+ /** Retrieve window state of ReducingState type. */
+ public <T> Optional<ReducingState<T>> getWindowState(
+ ReducingStateDeclaration<T> stateDeclaration, W namespace) throws Exception {
+ checkState(
+ !isMergingWindow,
+ "Retrieving the window state is not permitted when using merging windows, such as session windows.");
+
+ if (!isStateDeclared(stateDeclaration)) {
+ return Optional.empty();
+ }
+
+ if (stateRedistributionModeIsNotNone(stateDeclaration)) {
+ throw new UnsupportedOperationException(
+ "RedistributionMode "
+ + stateDeclaration.getRedistributionMode().name()
+ + " is not supported for window state.");
+ }
+
+ ReducingStateDescriptor<T> stateDescriptor =
+ new ReducingStateDescriptor<>(
+ stateDeclaration.getName(),
+ stateDeclaration.getReduceFunction(),
+ TypeExtractor.createTypeInfo(
+ stateDeclaration.getTypeDescriptor().getTypeClass()));
+
+ ReducingStateAdaptor<K, W, T> state =
+ operator.getOrCreateKeyedState(namespace, windowSerializer, stateDescriptor);
+ state.setCurrentNamespace(namespace);
+ return Optional.of(state);
+ }
+
+ /** Retrieve window state of AggregatingState type. */
+ public <T, ACC, OUT> Optional<AggregatingState<T, OUT>> getWindowState(
+ AggregatingStateDeclaration<T, ACC, OUT> stateDeclaration, W namespace)
+ throws Exception {
+ checkState(
+ !isMergingWindow,
+ "Retrieving the window state is not permitted when using merging windows, such as session windows.");
+
+ if (!isStateDeclared(stateDeclaration)) {
+ return Optional.empty();
+ }
+
+ if (stateRedistributionModeIsNotNone(stateDeclaration)) {
+ throw new UnsupportedOperationException(
+ "RedistributionMode "
+ + stateDeclaration.getRedistributionMode().name()
+ + " is not supported for window state.");
+ }
+
+ AggregatingStateDescriptor<T, ACC, OUT> stateDescriptor =
+ new AggregatingStateDescriptor<>(
+ stateDeclaration.getName(),
+ stateDeclaration.getAggregateFunction(),
+ TypeExtractor.createTypeInfo(
+ stateDeclaration.getTypeDescriptor().getTypeClass()));
+
+ AggregatingStateAdaptor<K, W, T, ACC, OUT> state =
+ operator.getOrCreateKeyedState(namespace, windowSerializer, stateDescriptor);
+ state.setCurrentNamespace(namespace);
+ return Optional.of(state);
+ }
+}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/context/WindowTriggerContext.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/context/WindowTriggerContext.java
new file mode 100644
index 00000000000..b90b1c967a0
--- /dev/null
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/context/WindowTriggerContext.java
@@ -0,0 +1,190 @@
+/*
+ * 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.datastream.impl.extension.window.context;
+
+import org.apache.flink.api.common.state.MergingState;
+import org.apache.flink.api.common.state.State;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
+import org.apache.flink.streaming.api.windowing.windows.Window;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.util.Collection;
+
+/**
+ * {@code WindowTriggerContext} is a utility for handling {@code Trigger} invocations. It can be
+ * reused by setting the {@code key} and {@code window} fields.
+ *
+ * @param <K> Type of the key.
+ * @param <IN> Type of the input value.
+ * @param <W> Type of the window.
+ */
+public class WindowTriggerContext<K, IN, W extends Window> implements Trigger.OnMergeContext {
+
+ /** Current processing key. */
+ private K key;
+
+ /** Current processing window. */
+ private W window;
+
+ /** The operator to which the window belongs, used for creating and retrieving window state. */
+ private final AbstractStreamOperator<?> operator;
+
+ /** The timer service of {@code operator}. */
+ private final InternalTimerService<W> internalTimerService;
+
+ private final TypeSerializer<W> windowSerializer;
+
+ private final Trigger<? super IN, ? super W> trigger;
+
+ private Collection<W> mergedWindows;
+
+ public WindowTriggerContext(
+ K key,
+ W window,
+ AbstractStreamOperator<?> operator,
+ InternalTimerService<W> internalTimerService,
+ Trigger<? super IN, ? super W> trigger,
+ TypeSerializer<W> windowSerializer) {
+ this.key = key;
+ this.window = window;
+ this.operator = operator;
+ this.internalTimerService = internalTimerService;
+ this.trigger = trigger;
+ this.windowSerializer = windowSerializer;
+ }
+
+ @Override
+ public MetricGroup getMetricGroup() {
+ return operator.getMetricGroup();
+ }
+
+ /** Get current event time. */
+ public long getCurrentWatermark() {
+ return internalTimerService.currentWatermark();
+ }
+
+ public <S extends State> S getPartitionedState(
+ org.apache.flink.api.common.state.StateDescriptor<S, ?> stateDescriptor) {
+ try {
+ return operator.getPartitionedState(window, windowSerializer, stateDescriptor);
+ } catch (Exception e) {
+ throw new RuntimeException("Could not retrieve state", e);
+ }
+ }
+
+ @Override
+ public <S extends MergingState<?, ?>> void mergePartitionedState(
+ org.apache.flink.api.common.state.StateDescriptor<S, ?> stateDescriptor) {
+ if (mergedWindows != null && !mergedWindows.isEmpty()) {
+ try {
+ S rawState =
+ operator.getKeyedStateBackend()
+ .getOrCreateKeyedState(windowSerializer, stateDescriptor);
+
+ if (rawState
+ instanceof org.apache.flink.runtime.state.internal.InternalMergingState) {
+ @SuppressWarnings("unchecked")
+ org.apache.flink.runtime.state.internal.InternalMergingState<K, W, ?, ?, ?>
+ mergingState =
+ (org.apache.flink.runtime.state.internal.InternalMergingState<
+ K, W, ?, ?, ?>)
+ rawState;
+ mergingState.mergeNamespaces(window, mergedWindows);
+ } else {
+ throw new IllegalArgumentException(
+ "The given state descriptor does not refer to a mergeable state (MergingState)");
+ }
+ } catch (Exception e) {
+ throw new RuntimeException("Error while merging state.", e);
+ }
+ }
+ }
+
+ @Override
+ public long getCurrentProcessingTime() {
+ return internalTimerService.currentProcessingTime();
+ }
+
+ @Override
+ public void registerProcessingTimeTimer(long time) {
+ internalTimerService.registerProcessingTimeTimer(window, time);
+ }
+
+ @Override
+ public void registerEventTimeTimer(long time) {
+ internalTimerService.registerEventTimeTimer(window, time);
+ }
+
+ @Override
+ public void deleteProcessingTimeTimer(long time) {
+ internalTimerService.deleteProcessingTimeTimer(window, time);
+ }
+
+ @Override
+ public void deleteEventTimeTimer(long time) {
+ internalTimerService.deleteEventTimeTimer(window, time);
+ }
+
+ public TriggerResult onElement(StreamRecord<IN> element) throws Exception {
+ return trigger.onElement(element.getValue(), element.getTimestamp(), window, this);
+ }
+
+ public TriggerResult onProcessingTime(long time) throws Exception {
+ return trigger.onProcessingTime(time, window, this);
+ }
+
+ public TriggerResult onEventTime(long time) throws Exception {
+ return trigger.onEventTime(time, window, this);
+ }
+
+ public void onMerge(Collection<W> mergedWindows) throws Exception {
+ this.mergedWindows = mergedWindows;
+ trigger.onMerge(window, this);
+ }
+
+ public void clear() throws Exception {
+ trigger.clear(window, this);
+ }
+
+ @Override
+ public String toString() {
+ return "WindowTriggerContext{" + "key=" + key + ", window=" + window + '}';
+ }
+
+ public void setKey(K key) {
+ this.key = key;
+ }
+
+ public void setWindow(W window) {
+ this.window = window;
+ }
+
+ public K getKey() {
+ return key;
+ }
+
+ public W getWindow() {
+ return window;
+ }
+}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/function/InternalOneInputWindowStreamProcessFunction.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/function/InternalOneInputWindowStreamProcessFunction.java
new file mode 100644
index 00000000000..b5370941612
--- /dev/null
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/function/InternalOneInputWindowStreamProcessFunction.java
@@ -0,0 +1,109 @@
+/*
+ * 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.datastream.impl.extension.window.function;
+
+import org.apache.flink.api.common.state.StateDeclaration;
+import org.apache.flink.datastream.api.common.Collector;
+import org.apache.flink.datastream.api.context.PartitionedContext;
+import org.apache.flink.datastream.api.extension.window.function.OneInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.api.extension.window.function.WindowProcessFunction;
+import org.apache.flink.datastream.api.extension.window.strategy.WindowStrategy;
+import org.apache.flink.datastream.api.function.OneInputStreamProcessFunction;
+import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.windows.Window;
+
+import java.util.Set;
+
+/**
+ * A class that wrap a {@link OneInputWindowStreamProcessFunction} to process function. This will be
+ * translated to a window operator instead of vanilla process operator.
+ *
+ * @param <IN> Type of the input elements.
+ * @param <OUT> Type of the output elements.
+ * @param <W> Type of the window.
+ */
+public class InternalOneInputWindowStreamProcessFunction<IN, OUT, W extends Window>
+ implements OneInputStreamProcessFunction<IN, OUT> {
+
+ /** User-defined {@link WindowProcessFunction}. */
+ private final OneInputWindowStreamProcessFunction<IN, OUT> windowProcessFunction;
+
+ private final WindowAssigner<IN, W> assigner;
+
+ private final Trigger<IN, W> trigger;
+
+ /**
+ * The allowed lateness for elements. This is used for:
+ *
+ * <ul>
+ * <li>Deciding if an element should be dropped from a window due to lateness.
+ * <li>Clearing the state of a window if the time out-of the {@code window.maxTimestamp +
+ * allowedLateness} landmark.
+ * </ul>
+ */
+ private final long allowedLateness;
+
+ private final WindowStrategy windowStrategy;
+
+ public InternalOneInputWindowStreamProcessFunction(
+ OneInputWindowStreamProcessFunction<IN, OUT> windowProcessFunction,
+ WindowAssigner<IN, W> assigner,
+ Trigger<IN, W> trigger,
+ long allowedLateness,
+ WindowStrategy windowStrategy) {
+ this.windowProcessFunction = windowProcessFunction;
+ this.assigner = assigner;
+ this.trigger = trigger;
+ this.allowedLateness = allowedLateness;
+ this.windowStrategy = windowStrategy;
+ }
+
+ @Override
+ public void processRecord(IN record, Collector<OUT> output, PartitionedContext<OUT> ctx)
+ throws Exception {
+ // Do nothing as this will translator to windowOperator instead of processOperator, and this
+ // method will never be invoked.
+ }
+
+ public WindowAssigner<IN, W> getAssigner() {
+ return assigner;
+ }
+
+ public Trigger<IN, W> getTrigger() {
+ return trigger;
+ }
+
+ public long getAllowedLateness() {
+ return allowedLateness;
+ }
+
+ public WindowStrategy getWindowStrategy() {
+ return windowStrategy;
+ }
+
+ public OneInputWindowStreamProcessFunction<IN, OUT> getWindowProcessFunction() {
+ return windowProcessFunction;
+ }
+
+ @Override
+ public Set<StateDeclaration> usesStates() {
+ return windowProcessFunction.usesStates();
+ }
+}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/function/InternalTwoInputWindowStreamProcessFunction.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/function/InternalTwoInputWindowStreamProcessFunction.java
new file mode 100644
index 00000000000..2bc3419e2b3
--- /dev/null
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/function/InternalTwoInputWindowStreamProcessFunction.java
@@ -0,0 +1,120 @@
+/*
+ * 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.datastream.impl.extension.window.function;
+
+import org.apache.flink.api.common.state.StateDeclaration;
+import org.apache.flink.datastream.api.common.Collector;
+import org.apache.flink.datastream.api.context.PartitionedContext;
+import org.apache.flink.datastream.api.extension.window.function.TwoInputNonBroadcastWindowStreamProcessFunction;
+import org.apache.flink.datastream.api.extension.window.function.WindowProcessFunction;
+import org.apache.flink.datastream.api.extension.window.strategy.WindowStrategy;
+import org.apache.flink.datastream.api.function.TwoInputNonBroadcastStreamProcessFunction;
+import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.windows.Window;
+import org.apache.flink.util.TaggedUnion;
+
+import java.util.Set;
+
+/**
+ * A class that wrap a {@link TwoInputNonBroadcastWindowStreamProcessFunction} to process function.
+ * This will be translated to a window operator instead of vanilla process operator.
+ *
+ * @param <IN1> Type of the first input elements.
+ * @param <IN2> Type of the second input elements.
+ * @param <OUT> Type of the output elements.
+ * @param <W> Type of the window.
+ */
+public class InternalTwoInputWindowStreamProcessFunction<IN1, IN2, OUT, W extends Window>
+ implements TwoInputNonBroadcastStreamProcessFunction<IN1, IN2, OUT> {
+
+ /** User-defined {@link WindowProcessFunction}. */
+ private final TwoInputNonBroadcastWindowStreamProcessFunction<IN1, IN2, OUT>
+ windowProcessFunction;
+
+ private final WindowAssigner<TaggedUnion<IN1, IN2>, W> assigner;
+
+ private final Trigger<TaggedUnion<IN1, IN2>, W> trigger;
+
+ /**
+ * The allowed lateness for elements. This is used for:
+ *
+ * <ul>
+ * <li>Deciding if an element should be dropped from a window due to lateness.
+ * <li>Clearing the state of a window if the time out-of the {@code window.maxTimestamp +
+ * allowedLateness} landmark.
+ * </ul>
+ */
+ private final long allowedLateness;
+
+ private final WindowStrategy windowStrategy;
+
+ public InternalTwoInputWindowStreamProcessFunction(
+ TwoInputNonBroadcastWindowStreamProcessFunction<IN1, IN2, OUT> windowProcessFunction,
+ WindowAssigner<TaggedUnion<IN1, IN2>, W> assigner,
+ Trigger<TaggedUnion<IN1, IN2>, W> trigger,
+ long allowedLateness,
+ WindowStrategy windowStrategy) {
+ this.windowProcessFunction = windowProcessFunction;
+ this.assigner = assigner;
+ this.trigger = trigger;
+ this.allowedLateness = allowedLateness;
+ this.windowStrategy = windowStrategy;
+ }
+
+ @Override
+ public void processRecordFromFirstInput(
+ IN1 record, Collector<OUT> output, PartitionedContext<OUT> ctx) throws Exception {
+ // Do nothing as this will translator to windowOperator instead of processOperator, and this
+ // method will never be invoked.
+ }
+
+ @Override
+ public void processRecordFromSecondInput(
+ IN2 record, Collector<OUT> output, PartitionedContext<OUT> ctx) throws Exception {
+ // Do nothing as this will translator to windowOperator instead of processOperator, and this
+ // method will never be invoked.
+ }
+
+ public TwoInputNonBroadcastWindowStreamProcessFunction<IN1, IN2, OUT>
+ getWindowProcessFunction() {
+ return windowProcessFunction;
+ }
+
+ @Override
+ public Set<StateDeclaration> usesStates() {
+ return windowProcessFunction.usesStates();
+ }
+
+ public WindowAssigner<TaggedUnion<IN1, IN2>, W> getAssigner() {
+ return assigner;
+ }
+
+ public Trigger<TaggedUnion<IN1, IN2>, W> getTrigger() {
+ return trigger;
+ }
+
+ public long getAllowedLateness() {
+ return allowedLateness;
+ }
+
+ public WindowStrategy getWindowStrategy() {
+ return windowStrategy;
+ }
+}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/function/InternalTwoOutputWindowStreamProcessFunction.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/function/InternalTwoOutputWindowStreamProcessFunction.java
new file mode 100644
index 00000000000..87335b01980
--- /dev/null
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/function/InternalTwoOutputWindowStreamProcessFunction.java
@@ -0,0 +1,114 @@
+/*
+ * 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.datastream.impl.extension.window.function;
+
+import org.apache.flink.api.common.state.StateDeclaration;
+import org.apache.flink.datastream.api.common.Collector;
+import org.apache.flink.datastream.api.context.TwoOutputPartitionedContext;
+import org.apache.flink.datastream.api.extension.window.function.TwoOutputWindowStreamProcessFunction;
+import org.apache.flink.datastream.api.extension.window.function.WindowProcessFunction;
+import org.apache.flink.datastream.api.extension.window.strategy.WindowStrategy;
+import org.apache.flink.datastream.api.function.TwoOutputStreamProcessFunction;
+import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.windows.Window;
+
+import java.util.Set;
+
+/**
+ * A class that wrap a {@link TwoOutputWindowStreamProcessFunction} to process function. This will
+ * be translated to a window operator instead of vanilla process operator.
+ *
+ * @param <IN> Type of the input elements.
+ * @param <OUT1> Type of the first output elements.
+ * @param <OUT2> Type of the second output elements.
+ * @param <W> Type of the window.
+ */
+public class InternalTwoOutputWindowStreamProcessFunction<IN, OUT1, OUT2, W extends Window>
+ implements TwoOutputStreamProcessFunction<IN, OUT1, OUT2> {
+
+ /** User-defined {@link WindowProcessFunction}. */
+ private final TwoOutputWindowStreamProcessFunction<IN, OUT1, OUT2> windowProcessFunction;
+
+ private final WindowAssigner<IN, W> assigner;
+
+ private final Trigger<IN, W> trigger;
+
+ /**
+ * The allowed lateness for elements. This is used for:
+ *
+ * <ul>
+ * <li>Deciding if an element should be dropped from a window due to lateness.
+ * <li>Clearing the state of a window if the time out-of the {@code window.maxTimestamp +
+ * allowedLateness} landmark.
+ * </ul>
+ */
+ private final long allowedLateness;
+
+ private final WindowStrategy windowStrategy;
+
+ public InternalTwoOutputWindowStreamProcessFunction(
+ TwoOutputWindowStreamProcessFunction<IN, OUT1, OUT2> windowProcessFunction,
+ WindowAssigner<IN, W> assigner,
+ Trigger<IN, W> trigger,
+ long allowedLateness,
+ WindowStrategy windowStrategy) {
+ this.windowProcessFunction = windowProcessFunction;
+ this.assigner = assigner;
+ this.trigger = trigger;
+ this.allowedLateness = allowedLateness;
+ this.windowStrategy = windowStrategy;
+ }
+
+ @Override
+ public void processRecord(
+ IN record,
+ Collector<OUT1> output1,
+ Collector<OUT2> output2,
+ TwoOutputPartitionedContext<OUT1, OUT2> ctx)
+ throws Exception {
+ // Do nothing as this will translator to windowOperator instead of processOperator, and this
+ // method will never be invoked.
+ }
+
+ public WindowAssigner<IN, W> getAssigner() {
+ return assigner;
+ }
+
+ public Trigger<IN, W> getTrigger() {
+ return trigger;
+ }
+
+ public long getAllowedLateness() {
+ return allowedLateness;
+ }
+
+ public WindowStrategy getWindowStrategy() {
+ return windowStrategy;
+ }
+
+ public TwoOutputWindowStreamProcessFunction<IN, OUT1, OUT2> getWindowProcessFunction() {
+ return windowProcessFunction;
+ }
+
+ @Override
+ public Set<StateDeclaration> usesStates() {
+ return windowProcessFunction.usesStates();
+ }
+}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/MergingWindowSet.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/MergingWindowSet.java
new file mode 100644
index 00000000000..416ab948dbb
--- /dev/null
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/MergingWindowSet.java
@@ -0,0 +1,246 @@
+/*
+ * 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.datastream.impl.extension.window.operators;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.runtime.state.v2.internal.InternalListState;
+import org.apache.flink.streaming.api.windowing.assigners.MergingWindowAssigner;
+import org.apache.flink.streaming.api.windowing.windows.Window;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Utility for keeping track of merging {@link Window Windows} when using a {@link
+ * MergingWindowAssigner} in a {@code WindowProcessOperator}.
+ *
+ * <p>Please refer to {@link
+ * org.apache.flink.streaming.runtime.operators.windowing.MergingWindowSet}. We do not use this
+ * directly because we are utilizing State V2, while it is based on State V1.
+ */
+public class MergingWindowSet<W extends Window> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MergingWindowSet.class);
+
+ /**
+ * Mapping from window to the window that keeps the window state. When we are incrementally
+ * merging windows starting from some window we keep that starting window as the state window to
+ * prevent costly state juggling.
+ */
+ private final Map<W, W> mapping;
+
+ /**
+ * Mapping when we created the {@code MergingWindowSet}. We use this to decide whether we need
+ * to persist any changes to state.
+ */
+ private final Map<W, W> initialMapping;
+
+ private final InternalListState<?, VoidNamespace, Tuple2<W, W>> state;
+
+ /** Our window assigner. */
+ private final MergingWindowAssigner<?, W> windowAssigner;
+
+ /** Restores a {@link MergingWindowSet} from the given state. */
+ public MergingWindowSet(
+ MergingWindowAssigner<?, W> windowAssigner,
+ InternalListState<?, VoidNamespace, Tuple2<W, W>> state)
+ throws Exception {
+ this.windowAssigner = windowAssigner;
+ mapping = new HashMap<>();
+
+ Iterable<Tuple2<W, W>> windowState = state.get();
+ if (windowState != null) {
+ for (Tuple2<W, W> window : windowState) {
+ mapping.put(window.f0, window.f1);
+ }
+ }
+
+ this.state = state;
+
+ initialMapping = new HashMap<>();
+ initialMapping.putAll(mapping);
+ }
+
+ /**
+ * Persist the updated mapping to the given state if the mapping changed since initialization.
+ */
+ public void persist() throws Exception {
+ if (!mapping.equals(initialMapping)) {
+ state.update(
+ mapping.entrySet().stream()
+ .map((w) -> new Tuple2<>(w.getKey(), w.getValue()))
+ .collect(Collectors.toList()));
+ }
+ }
+
+ /**
+ * Returns the state window for the given in-flight {@code Window}. The state window is the
+ * {@code Window} in which we keep the actual state of a given in-flight window. Windows might
+ * expand but we keep to original state window for keeping the elements of the window to avoid
+ * costly state juggling.
+ *
+ * @param window The window for which to get the state window.
+ */
+ public W getStateWindow(W window) {
+ return mapping.get(window);
+ }
+
+ /**
+ * Removes the given window from the set of in-flight windows.
+ *
+ * @param window The {@code Window} to remove.
+ */
+ public void retireWindow(W window) {
+ W removed = this.mapping.remove(window);
+ if (removed == null) {
+ throw new IllegalStateException(
+ "Window " + window + " is not in in-flight window set.");
+ }
+ }
+
+ /**
+ * Adds a new {@code Window} to the set of in-flight windows. It might happen that this triggers
+ * merging of previously in-flight windows. In that case, the provided {@link MergeFunction} is
+ * called.
+ *
+ * <p>This returns the window that is the representative of the added window after adding. This
+ * can either be the new window itself, if no merge occurred, or the newly merged window. Adding
+ * an element to a window or calling trigger functions should only happen on the returned
+ * representative. This way, we never have to deal with a new window that is immediately
+ * swallowed up by another window.
+ *
+ * <p>If the new window is merged, the {@code MergeFunction} callback arguments also don't
+ * contain the new window as part of the list of merged windows.
+ *
+ * @param newWindow The new {@code Window} to add.
+ * @param mergeFunction The callback to be invoked in case a merge occurs.
+ * @return The {@code Window} that new new {@code Window} ended up in. This can also be the new
+ * {@code Window} itself in case no merge occurred.
+ * @throws Exception
+ */
+ public W addWindow(W newWindow, MergeFunction<W> mergeFunction) throws Exception {
+
+ List<W> windows = new ArrayList<>();
+
+ windows.addAll(this.mapping.keySet());
+ windows.add(newWindow);
+
+ final Map<W, Collection<W>> mergeResults = new HashMap<>();
+ windowAssigner.mergeWindows(
+ windows,
+ new MergingWindowAssigner.MergeCallback<W>() {
+ @Override
+ public void merge(Collection<W> toBeMerged, W mergeResult) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Merging {} into {}", toBeMerged, mergeResult);
+ }
+ mergeResults.put(mergeResult, toBeMerged);
+ }
+ });
+
+ W resultWindow = newWindow;
+ boolean mergedNewWindow = false;
+
+ // perform the merge
+ for (Map.Entry<W, Collection<W>> c : mergeResults.entrySet()) {
+ W mergeResult = c.getKey();
+ Collection<W> mergedWindows = c.getValue();
+
+ // if our new window is in the merged windows make the merge result the
+ // result window
+ if (mergedWindows.remove(newWindow)) {
+ mergedNewWindow = true;
+ resultWindow = mergeResult;
+ }
+
+ // pick any of the merged windows and choose that window's state window
+ // as the state window for the merge result
+ W mergedStateWindow = this.mapping.get(mergedWindows.iterator().next());
+
+ // figure out the state windows that we are merging
+ List<W> mergedStateWindows = new ArrayList<>();
+ for (W mergedWindow : mergedWindows) {
+ W res = this.mapping.remove(mergedWindow);
+ if (res != null) {
+ mergedStateWindows.add(res);
+ }
+ }
+
+ this.mapping.put(mergeResult, mergedStateWindow);
+
+ // don't put the target state window into the merged windows
+ mergedStateWindows.remove(mergedStateWindow);
+
+ // don't merge the new window itself, it never had any state associated with it
+ // i.e. if we are only merging one pre-existing window into itself
+ // without extending the pre-existing window
+ if (!(mergedWindows.contains(mergeResult) && mergedWindows.size() == 1)) {
+ mergeFunction.merge(
+ mergeResult,
+ mergedWindows,
+ this.mapping.get(mergeResult),
+ mergedStateWindows);
+ }
+ }
+
+ // the new window created a new, self-contained window without merging
+ if (mergeResults.isEmpty() || (resultWindow.equals(newWindow) && !mergedNewWindow)) {
+ this.mapping.put(resultWindow, resultWindow);
+ }
+
+ return resultWindow;
+ }
+
+ /**
+ * Callback for {@link #addWindow(Window, MergeFunction)}.
+ *
+ * @param <W>
+ */
+ public interface MergeFunction<W> {
+
+ /**
+ * This gets called when a merge occurs.
+ *
+ * @param mergeResult The newly resulting merged {@code Window}.
+ * @param mergedWindows The merged {@code Window Windows}.
+ * @param stateWindowResult The state window of the merge result.
+ * @param mergedStateWindows The merged state windows.
+ * @throws Exception
+ */
+ void merge(
+ W mergeResult,
+ Collection<W> mergedWindows,
+ W stateWindowResult,
+ Collection<W> mergedStateWindows)
+ throws Exception;
+ }
+
+ @Override
+ public String toString() {
+ return "MergingWindowSet{" + "windows=" + mapping + '}';
+ }
+}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/OneInputWindowProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/OneInputWindowProcessOperator.java
new file mode 100644
index 00000000000..8982e81c153
--- /dev/null
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/OneInputWindowProcessOperator.java
@@ -0,0 +1,532 @@
+/*
+ * 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.datastream.impl.extension.window.operators;
+
+import org.apache.flink.api.common.state.v2.AppendingState;
+import org.apache.flink.api.common.state.v2.ListStateDescriptor;
+import org.apache.flink.api.common.state.v2.StateDescriptor;
+import org.apache.flink.api.common.state.v2.StateIterator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
+import org.apache.flink.datastream.api.context.ProcessingTimeManager;
+import org.apache.flink.datastream.api.extension.window.function.OneInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.api.extension.window.function.WindowProcessFunction;
+import org.apache.flink.datastream.api.stream.KeyedPartitionStream;
+import org.apache.flink.datastream.impl.context.UnsupportedProcessingTimeManager;
+import org.apache.flink.datastream.impl.extension.window.context.DefaultOneInputWindowContext;
+import org.apache.flink.datastream.impl.extension.window.context.WindowTriggerContext;
+import org.apache.flink.datastream.impl.extension.window.function.InternalOneInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.utils.WindowUtils;
+import org.apache.flink.datastream.impl.operators.BaseKeyedProcessOperator;
+import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.runtime.state.VoidNamespaceSerializer;
+import org.apache.flink.runtime.state.v2.internal.InternalAppendingState;
+import org.apache.flink.runtime.state.v2.internal.InternalListState;
+import org.apache.flink.runtime.state.v2.internal.InternalMergingState;
+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.windowing.assigners.MergingWindowAssigner;
+import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
+import org.apache.flink.streaming.api.windowing.windows.Window;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.util.Collection;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/** Operator for {@link OneInputWindowStreamProcessFunction} in {@link KeyedPartitionStream}. */
+public class OneInputWindowProcessOperator<K, IN, OUT, W extends Window>
+ extends BaseKeyedProcessOperator<K, IN, OUT> implements Triggerable<K, W> {
+
+ private static final long serialVersionUID = 1L;
+
+ // ------------------------------------------------------------------------
+ // Configuration values and user functions
+ // ------------------------------------------------------------------------
+
+ /** User-defined {@link WindowProcessFunction}. */
+ private final OneInputWindowStreamProcessFunction<IN, OUT> windowProcessFunction;
+
+ /**
+ * The allowed lateness for elements. This is used for:
+ *
+ * <ul>
+ * <li>Deciding if an element should be dropped from a window due to lateness.
+ * <li>Clearing the state of a window if the time out-of the {@code window.maxTimestamp +
+ * allowedLateness} landmark.
+ * </ul>
+ */
+ private final long allowedLateness;
+
+ // ------------------------------------------------------------------------
+ // Operator components
+ // ------------------------------------------------------------------------
+
+ private transient InternalTimerService<W> internalTimerService;
+
+ /** For serializing the window in checkpoints. */
+ private final TypeSerializer<W> windowSerializer;
+
+ // ------------------------------------------------------------------------
+ // Window assigner and trigger
+ // ------------------------------------------------------------------------
+
+ private final WindowAssigner<? super IN, W> windowAssigner;
+
+ private transient WindowAssigner.WindowAssignerContext windowAssignerContext;
+
+ private final Trigger<? super IN, ? super W> trigger;
+
+ private transient WindowTriggerContext<K, IN, W> triggerContext;
+
+ private transient DefaultOneInputWindowContext<K, IN, W> windowFunctionContext;
+
+ // ------------------------------------------------------------------------
+ // State that is not checkpointed
+ // ------------------------------------------------------------------------
+
+ private final StateDescriptor<IN> windowStateDescriptor;
+
+ /** The state in which the window contents is stored. Each window is a namespace */
+ private transient InternalAppendingState<K, W, IN, IN, StateIterator<IN>, Iterable<IN>>
+ windowState;
+
+ /**
+ * The {@link #windowState}, typed to merging state for merging windows. Null if the window
+ * state is not mergeable.
+ */
+ private transient InternalMergingState<K, W, IN, IN, StateIterator<IN>, Iterable<IN>>
+ windowMergingState;
+
+ /** The state that holds the merging window metadata (the sets that describe what is merged). */
+ private transient InternalListState<K, VoidNamespace, Tuple2<W, W>> mergingSetsState;
+
+ public OneInputWindowProcessOperator(
+ InternalOneInputWindowStreamProcessFunction<IN, OUT, W> windowProcessFunction,
+ WindowAssigner<? super IN, W> windowAssigner,
+ Trigger<? super IN, ? super W> trigger,
+ TypeSerializer<W> windowSerializer,
+ StateDescriptor<IN> windowStateDescriptor,
+ long allowedLateness) {
+ super(windowProcessFunction);
+ checkArgument(allowedLateness >= 0);
+ this.windowProcessFunction = windowProcessFunction.getWindowProcessFunction();
+ this.windowAssigner = windowAssigner;
+ this.trigger = trigger;
+ this.windowSerializer = windowSerializer;
+ this.windowStateDescriptor = windowStateDescriptor;
+ this.allowedLateness = allowedLateness;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ internalTimerService =
+ getInternalTimerService("process-window-timers", windowSerializer, this);
+
+ // create (or restore) the state that hold the actual window contents
+ // NOTE - the state may be null in the case of the overriding evicting window operator
+ if (windowStateDescriptor != null) {
+ windowState =
+ getOrCreateKeyedState(
+ windowSerializer.createInstance(),
+ windowSerializer,
+ windowStateDescriptor);
+ }
+
+ // create the typed and helper states for merging windows
+ if (windowAssigner instanceof MergingWindowAssigner) {
+
+ // store a typed reference for the state of merging windows - sanity check
+ if (windowState instanceof InternalMergingState) {
+ windowMergingState =
+ (InternalMergingState<K, W, IN, IN, StateIterator<IN>, Iterable<IN>>)
+ windowState;
+ } else if (windowState != null) {
+ throw new IllegalStateException(
+ "The window uses a merging assigner, but the window state is not mergeable.");
+ }
+
+ @SuppressWarnings("unchecked")
+ final Class<Tuple2<W, W>> typedTuple = (Class<Tuple2<W, W>>) (Class<?>) Tuple2.class;
+
+ final TupleSerializer<Tuple2<W, W>> tupleSerializer =
+ new TupleSerializer<>(
+ typedTuple, new TypeSerializer[] {windowSerializer, windowSerializer});
+
+ final ListStateDescriptor<Tuple2<W, W>> mergingSetsStateDescriptor =
+ new ListStateDescriptor<>("merging-window-set", tupleSerializer);
+
+ // get the state that stores the merging sets
+ mergingSetsState =
+ getOrCreateKeyedState(
+ VoidNamespaceSerializer.INSTANCE.createInstance(),
+ VoidNamespaceSerializer.INSTANCE,
+ mergingSetsStateDescriptor);
+ mergingSetsState.setCurrentNamespace(VoidNamespace.INSTANCE);
+ }
+
+ triggerContext =
+ new WindowTriggerContext<>(
+ null, null, this, internalTimerService, trigger, windowSerializer);
+ windowAssignerContext =
+ new WindowAssigner.WindowAssignerContext() {
+ @Override
+ public long getCurrentProcessingTime() {
+ return internalTimerService.currentProcessingTime();
+ }
+ };
+ windowFunctionContext =
+ new DefaultOneInputWindowContext<>(
+ null,
+ windowState,
+ windowProcessFunction,
+ this,
+ windowSerializer,
+ windowMergingState != null);
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ triggerContext = null;
+ windowFunctionContext = null;
+ windowAssignerContext = null;
+ }
+
+ @Override
+ public void processElement(StreamRecord<IN> element) throws Exception {
+ final Collection<W> elementWindows =
+ windowAssigner.assignWindows(
+ element.getValue(), element.getTimestamp(), windowAssignerContext);
+
+ // if element is handled by none of assigned elementWindows
+ boolean isSkippedElement = true;
+
+ final K key = (K) this.getCurrentKey();
+
+ if (windowAssigner instanceof MergingWindowAssigner) {
+ MergingWindowSet<W> mergingWindows = getMergingWindowSet();
+
+ for (W window : elementWindows) {
+
+ // adding the new window might result in a merge, in that case the actualWindow
+ // is the merged window and we work with that. If we don't merge then
+ // actualWindow == window
+ W actualWindow =
+ mergingWindows.addWindow(
+ window,
+ new MergingWindowSet.MergeFunction<>() {
+ @Override
+ public void merge(
+ W mergeResult,
+ Collection<W> mergedWindows,
+ W stateWindowResult,
+ Collection<W> mergedStateWindows)
+ throws Exception {
+
+ if ((windowAssigner.isEventTime()
+ && mergeResult.maxTimestamp() + allowedLateness
+ <= internalTimerService
+ .currentWatermark())) {
+ throw new UnsupportedOperationException(
+ "The end timestamp of an "
+ + "event-time window cannot become earlier than the current watermark "
+ + "by merging. Current event time: "
+ + internalTimerService
+ .currentWatermark()
+ + " window: "
+ + mergeResult);
+ } else if (!windowAssigner.isEventTime()) {
+ long currentProcessingTime =
+ internalTimerService.currentProcessingTime();
+ if (mergeResult.maxTimestamp()
+ <= currentProcessingTime) {
+ throw new UnsupportedOperationException(
+ "The end timestamp of a "
+ + "processing-time window cannot become earlier than the current processing time "
+ + "by merging. Current processing time: "
+ + currentProcessingTime
+ + " window: "
+ + mergeResult);
+ }
+ }
+
+ triggerContext.setKey(key);
+ triggerContext.setWindow(mergeResult);
+
+ triggerContext.onMerge(mergedWindows);
+
+ for (W m : mergedWindows) {
+ triggerContext.setWindow(m);
+ triggerContext.clear();
+ WindowUtils.deleteCleanupTimer(
+ m,
+ windowAssigner,
+ triggerContext,
+ allowedLateness);
+ }
+
+ // merge the merged state windows into the newly resulting
+ // state window
+ windowMergingState.mergeNamespaces(
+ stateWindowResult, mergedStateWindows);
+ }
+ });
+
+ // drop if the window is already late
+ if (WindowUtils.isWindowLate(
+ actualWindow, windowAssigner, internalTimerService, allowedLateness)) {
+ mergingWindows.retireWindow(actualWindow);
+ continue;
+ }
+ isSkippedElement = false;
+
+ W stateWindow = mergingWindows.getStateWindow(actualWindow);
+ if (stateWindow == null) {
+ throw new IllegalStateException(
+ "Window " + window + " is not in in-flight window set.");
+ }
+
+ windowState.setCurrentNamespace(stateWindow);
+ outputCollector.setTimestamp(window.maxTimestamp());
+ windowFunctionContext.setWindow(window);
+ windowProcessFunction.onRecord(
+ element.getValue(),
+ outputCollector,
+ partitionedContext,
+ windowFunctionContext);
+
+ triggerContext.setKey(key);
+ triggerContext.setWindow(actualWindow);
+
+ TriggerResult triggerResult = triggerContext.onElement(element);
+
+ if (triggerResult.isFire()) {
+ emitWindowContents(actualWindow);
+ }
+
+ if (triggerResult.isPurge()) {
+ windowState.clear();
+ }
+
+ WindowUtils.registerCleanupTimer(
+ actualWindow, windowAssigner, triggerContext, allowedLateness);
+ }
+
+ // need to make sure to update the merging state in state
+ mergingWindows.persist();
+ } else {
+ for (W window : elementWindows) {
+
+ // drop if the window is already late
+ if (WindowUtils.isWindowLate(
+ window, windowAssigner, internalTimerService, allowedLateness)) {
+ continue;
+ }
+ isSkippedElement = false;
+
+ windowState.setCurrentNamespace(window);
+ outputCollector.setTimestamp(window.maxTimestamp());
+ windowFunctionContext.setWindow(window);
+ windowProcessFunction.onRecord(
+ element.getValue(),
+ outputCollector,
+ partitionedContext,
+ windowFunctionContext);
+
+ triggerContext.setKey(key);
+ triggerContext.setWindow(window);
+
+ TriggerResult triggerResult = triggerContext.onElement(element);
+
+ if (triggerResult.isFire()) {
+ emitWindowContents(window);
+ }
+
+ if (triggerResult.isPurge()) {
+ windowState.clear();
+ }
+
+ WindowUtils.registerCleanupTimer(
+ window, windowAssigner, triggerContext, allowedLateness);
+ }
+ }
+
+ // side output input event if element not handled by any window late arriving tag has been
+ // set windowAssigner is event time and current timestamp + allowed lateness no less than
+ // element timestamp.
+ if (isSkippedElement
+ && WindowUtils.isElementLate(
+ element, windowAssigner, allowedLateness, internalTimerService)) {
+ windowProcessFunction.onLateRecord(
+ element.getValue(), outputCollector, partitionedContext);
+ }
+ }
+
+ @Override
+ public void onEventTime(InternalTimer<K, W> timer) throws Exception {
+ triggerContext.setKey(timer.getKey());
+ triggerContext.setWindow(timer.getNamespace());
+
+ MergingWindowSet<W> mergingWindows;
+
+ if (windowAssigner instanceof MergingWindowAssigner) {
+ mergingWindows = getMergingWindowSet();
+ W stateWindow = mergingWindows.getStateWindow(triggerContext.getWindow());
+ if (stateWindow == null) {
+ // Timer firing for non-existent window, this can only happen if a
+ // trigger did not clean up timers. We have already cleared the merging
+ // window and therefore the Trigger state, however, so nothing to do.
+ return;
+ } else {
+ windowState.setCurrentNamespace(stateWindow);
+ }
+ } else {
+ windowState.setCurrentNamespace(triggerContext.getWindow());
+ mergingWindows = null;
+ }
+
+ TriggerResult triggerResult = triggerContext.onEventTime(timer.getTimestamp());
+
+ if (triggerResult.isFire()) {
+ emitWindowContents(triggerContext.getWindow());
+ }
+
+ if (triggerResult.isPurge()) {
+ windowState.clear();
+ }
+
+ if (windowAssigner.isEventTime()
+ && WindowUtils.isCleanupTime(
+ triggerContext.getWindow(),
+ timer.getTimestamp(),
+ windowAssigner,
+ allowedLateness)) {
+ clearAllState(triggerContext.getWindow(), windowState, mergingWindows);
+ }
+
+ if (mergingWindows != null) {
+ // need to make sure to update the merging state in state
+ mergingWindows.persist();
+ }
+ }
+
+ @Override
+ public void onProcessingTime(InternalTimer<K, W> timer) throws Exception {
+ triggerContext.setKey(timer.getKey());
+ triggerContext.setWindow(timer.getNamespace());
+
+ MergingWindowSet<W> mergingWindows;
+
+ if (windowAssigner instanceof MergingWindowAssigner) {
+ mergingWindows = getMergingWindowSet();
+ W stateWindow = mergingWindows.getStateWindow(triggerContext.getWindow());
+ if (stateWindow == null) {
+ // Timer firing for non-existent window, this can only happen if a
+ // trigger did not clean up timers. We have already cleared the merging
+ // window and therefore the Trigger state, however, so nothing to do.
+ return;
+ } else {
+ windowState.setCurrentNamespace(stateWindow);
+ }
+ } else {
+ windowState.setCurrentNamespace(triggerContext.getWindow());
+ mergingWindows = null;
+ }
+
+ TriggerResult triggerResult = triggerContext.onProcessingTime(timer.getTimestamp());
+
+ if (triggerResult.isFire()) {
+ emitWindowContents(triggerContext.getWindow());
+ }
+
+ if (triggerResult.isPurge()) {
+ windowState.clear();
+ }
+
+ if (!windowAssigner.isEventTime()
+ && WindowUtils.isCleanupTime(
+ triggerContext.getWindow(),
+ timer.getTimestamp(),
+ windowAssigner,
+ allowedLateness)) {
+ clearAllState(triggerContext.getWindow(), windowState, mergingWindows);
+ }
+
+ if (mergingWindows != null) {
+ // need to make sure to update the merging state in state
+ mergingWindows.persist();
+ }
+ }
+
+ @Override
+ protected ProcessingTimeManager getProcessingTimeManager() {
+ // we don't support user utilize processing time in window operators
+ return UnsupportedProcessingTimeManager.INSTANCE;
+ }
+
+ /**
+ * Drops all state for the given window and calls {@link Trigger#clear(Window,
+ * Trigger.TriggerContext)}.
+ *
+ * <p>The caller must ensure that the correct key is set in the state backend and the
+ * triggerContext object.
+ */
+ private void clearAllState(
+ W window,
+ AppendingState<IN, StateIterator<IN>, Iterable<IN>> windowState,
+ MergingWindowSet<W> mergingWindows)
+ throws Exception {
+ windowState.clear();
+ triggerContext.clear();
+ windowFunctionContext.setWindow(window);
+ windowProcessFunction.onClear(outputCollector, partitionedContext, windowFunctionContext);
+ if (mergingWindows != null) {
+ mergingWindows.retireWindow(window);
+ mergingWindows.persist();
+ }
+ }
+
+ /** Emits the contents of the given window using the {@link WindowProcessFunction}. */
+ private void emitWindowContents(W window) throws Exception {
+ // only time window touch the time concept.
+ outputCollector.setTimestamp(window.maxTimestamp());
+ windowFunctionContext.setWindow(window);
+ windowProcessFunction.onTrigger(outputCollector, partitionedContext, windowFunctionContext);
+ }
+
+ /**
+ * Retrieves the {@link MergingWindowSet} for the currently active key. The caller must ensure
+ * that the correct key is set in the state backend.
+ *
+ * <p>The caller must also ensure to properly persist changes to state using {@link
+ * MergingWindowSet#persist()}.
+ */
+ protected MergingWindowSet<W> getMergingWindowSet() throws Exception {
+
+ MergingWindowAssigner<? super IN, W> mergingAssigner =
+ (MergingWindowAssigner<? super IN, W>) windowAssigner;
+ return new MergingWindowSet<>(mergingAssigner, mergingSetsState);
+ }
+}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/TwoInputNonBroadcastWindowProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/TwoInputNonBroadcastWindowProcessOperator.java
new file mode 100644
index 00000000000..694efea79a0
--- /dev/null
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/TwoInputNonBroadcastWindowProcessOperator.java
@@ -0,0 +1,763 @@
+/*
+ * 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.datastream.impl.extension.window.operators;
+
+import org.apache.flink.api.common.state.v2.AppendingState;
+import org.apache.flink.api.common.state.v2.ListStateDescriptor;
+import org.apache.flink.api.common.state.v2.StateDescriptor;
+import org.apache.flink.api.common.state.v2.StateIterator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
+import org.apache.flink.datastream.api.context.ProcessingTimeManager;
+import org.apache.flink.datastream.api.extension.window.function.TwoInputNonBroadcastWindowStreamProcessFunction;
+import org.apache.flink.datastream.api.stream.KeyedPartitionStream;
+import org.apache.flink.datastream.impl.context.UnsupportedProcessingTimeManager;
+import org.apache.flink.datastream.impl.extension.window.context.DefaultTwoInputWindowContext;
+import org.apache.flink.datastream.impl.extension.window.context.WindowTriggerContext;
+import org.apache.flink.datastream.impl.extension.window.function.InternalTwoInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.utils.WindowUtils;
+import org.apache.flink.datastream.impl.operators.BaseKeyedTwoInputNonBroadcastProcessOperator;
+import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.runtime.state.VoidNamespaceSerializer;
+import org.apache.flink.runtime.state.v2.internal.InternalAppendingState;
+import org.apache.flink.runtime.state.v2.internal.InternalListState;
+import org.apache.flink.runtime.state.v2.internal.InternalMergingState;
+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.windowing.assigners.MergingWindowAssigner;
+import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
+import org.apache.flink.streaming.api.windowing.windows.Window;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.TaggedUnion;
+
+import java.util.Collection;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Operator for {@link TwoInputNonBroadcastWindowStreamProcessFunction} in {@link
+ * KeyedPartitionStream}.
+ */
+public class TwoInputNonBroadcastWindowProcessOperator<K, IN1, IN2, OUT, W extends Window>
+ extends BaseKeyedTwoInputNonBroadcastProcessOperator<K, IN1, IN2, OUT>
+ implements Triggerable<K, W> {
+
+ private static final long serialVersionUID = 1L;
+
+ // ------------------------------------------------------------------------
+ // Configuration values and user functions
+ // ------------------------------------------------------------------------
+
+ private final TwoInputNonBroadcastWindowStreamProcessFunction<IN1, IN2, OUT>
+ windowProcessFunction;
+
+ /**
+ * The allowed lateness for elements. This is used for:
+ *
+ * <ul>
+ * <li>Deciding if an element should be dropped from a window due to lateness.
+ * <li>Clearing the state of a window if the time out-of the {@code window.maxTimestamp +
+ * allowedLateness} landmark.
+ * </ul>
+ */
+ protected final long allowedLateness;
+
+ // ------------------------------------------------------------------------
+ // Operator components
+ // ------------------------------------------------------------------------
+
+ protected transient InternalTimerService<W> internalTimerService;
+
+ /** For serializing the window in checkpoints. */
+ private final TypeSerializer<W> windowSerializer;
+
+ // ------------------------------------------------------------------------
+ // Window assigner and trigger
+ // ------------------------------------------------------------------------
+
+ private final WindowAssigner<? super TaggedUnion<IN1, IN2>, W> windowAssigner;
+
+ private transient WindowAssigner.WindowAssignerContext windowAssignerContext;
+
+ private final Trigger<? super TaggedUnion<IN1, IN2>, ? super W> trigger;
+
+ private transient WindowTriggerContext<K, ? super TaggedUnion<IN1, IN2>, W> triggerContext;
+
+ private transient DefaultTwoInputWindowContext<K, IN1, IN2, W> windowFunctionContext;
+
+ // ------------------------------------------------------------------------
+ // State that is not checkpointed
+ // ------------------------------------------------------------------------
+
+ private final StateDescriptor<IN1> leftWindowStateDescriptor;
+
+ private final StateDescriptor<IN2> rightWindowStateDescriptor;
+
+ /**
+ * The state in which the window contents from the left input are stored. Each window is a
+ * namespace.
+ */
+ private transient InternalAppendingState<K, W, IN1, IN1, StateIterator<IN1>, Iterable<IN1>>
+ leftWindowState;
+
+ /**
+ * The state in which the window contents from the right input are stored. Each window is a
+ * namespace
+ */
+ private transient InternalAppendingState<K, W, IN2, IN2, StateIterator<IN2>, Iterable<IN2>>
+ rightWindowState;
+
+ /**
+ * The {@link #leftWindowState}, typed to merging state for merging windows. Null if the window
+ * state is not mergeable.
+ */
+ private transient InternalMergingState<K, W, IN1, IN1, StateIterator<IN1>, Iterable<IN1>>
+ leftWindowMergingState;
+
+ /**
+ * The {@link #rightWindowState}, typed to merging state for merging windows. Null if the window
+ * state is not mergeable.
+ */
+ private transient InternalMergingState<K, W, IN2, IN2, StateIterator<IN2>, Iterable<IN2>>
+ rightWindowMergingState;
+
+ /** The state that holds the merging window metadata (the sets that describe what is merged). */
+ private transient InternalListState<K, VoidNamespace, Tuple2<W, W>> mergingSetsState;
+
+ public TwoInputNonBroadcastWindowProcessOperator(
+ InternalTwoInputWindowStreamProcessFunction<IN1, IN2, OUT, W> windowFunction,
+ WindowAssigner<? super TaggedUnion<IN1, IN2>, W> windowAssigner,
+ Trigger<? super TaggedUnion<IN1, IN2>, ? super W> trigger,
+ TypeSerializer<W> windowSerializer,
+ StateDescriptor<IN1> leftWindowStateDescriptor,
+ StateDescriptor<IN2> rightWindowStateDescriptor,
+ long allowedLateness) {
+ super(windowFunction);
+ checkArgument(allowedLateness >= 0);
+ this.windowProcessFunction = windowFunction.getWindowProcessFunction();
+ this.windowAssigner = windowAssigner;
+ this.trigger = trigger;
+ this.windowSerializer = windowSerializer;
+ this.leftWindowStateDescriptor = leftWindowStateDescriptor;
+ this.rightWindowStateDescriptor = rightWindowStateDescriptor;
+ this.allowedLateness = allowedLateness;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ internalTimerService =
+ getInternalTimerService("process-window-timers", windowSerializer, this);
+
+ // create (or restore) the state that hold the actual window contents
+ // NOTE - the state may be null in the case of the overriding evicting window operator
+ if (leftWindowStateDescriptor != null) {
+ leftWindowState =
+ getOrCreateKeyedState(
+ windowSerializer.createInstance(),
+ windowSerializer,
+ leftWindowStateDescriptor);
+ }
+
+ if (rightWindowStateDescriptor != null) {
+ rightWindowState =
+ getOrCreateKeyedState(
+ windowSerializer.createInstance(),
+ windowSerializer,
+ rightWindowStateDescriptor);
+ }
+
+ // create the typed and helper states for merging windows
+ if (windowAssigner instanceof MergingWindowAssigner) {
+
+ // store a typed reference for the state of merging windows - sanity check
+ if (leftWindowState instanceof InternalMergingState) {
+ leftWindowMergingState =
+ (InternalMergingState<K, W, IN1, IN1, StateIterator<IN1>, Iterable<IN1>>)
+ leftWindowState;
+ } else if (leftWindowState != null) {
+ throw new IllegalStateException(
+ "The window uses a merging assigner, but the window state is not mergeable.");
+ }
+
+ if (rightWindowState instanceof InternalMergingState) {
+ rightWindowMergingState =
+ (InternalMergingState<K, W, IN2, IN2, StateIterator<IN2>, Iterable<IN2>>)
+ rightWindowState;
+ } else if (rightWindowState != null) {
+ throw new IllegalStateException(
+ "The window uses a merging assigner, but the window state is not mergeable.");
+ }
+
+ @SuppressWarnings("unchecked")
+ final Class<Tuple2<W, W>> typedTuple = (Class<Tuple2<W, W>>) (Class<?>) Tuple2.class;
+
+ final TupleSerializer<Tuple2<W, W>> tupleSerializer =
+ new TupleSerializer<>(
+ typedTuple, new TypeSerializer[] {windowSerializer, windowSerializer});
+
+ final ListStateDescriptor<Tuple2<W, W>> mergingSetsStateDescriptor =
+ new ListStateDescriptor<>("merging-window-set", tupleSerializer);
+
+ // get the state that stores the merging sets
+ mergingSetsState =
+ getOrCreateKeyedState(
+ VoidNamespaceSerializer.INSTANCE.createInstance(),
+ VoidNamespaceSerializer.INSTANCE,
+ mergingSetsStateDescriptor);
+ mergingSetsState.setCurrentNamespace(VoidNamespace.INSTANCE);
+ }
+
+ triggerContext =
+ new WindowTriggerContext<>(
+ null, null, this, internalTimerService, trigger, windowSerializer);
+ windowAssignerContext =
+ new WindowAssigner.WindowAssignerContext() {
+ @Override
+ public long getCurrentProcessingTime() {
+ return internalTimerService.currentProcessingTime();
+ }
+ };
+ windowFunctionContext =
+ new DefaultTwoInputWindowContext<>(
+ null,
+ leftWindowState,
+ rightWindowState,
+ windowProcessFunction,
+ this,
+ windowSerializer,
+ leftWindowMergingState != null);
+ }
+
+ @Override
+ public void processElement1(StreamRecord<IN1> element) throws Exception {
+ final Collection<W> elementWindows =
+ windowAssigner.assignWindows(
+ TaggedUnion.one(element.getValue()),
+ element.getTimestamp(),
+ windowAssignerContext);
+
+ // if element is handled by none of assigned elementWindows
+ boolean isSkippedElement = true;
+
+ final K key = (K) this.getCurrentKey();
+
+ if (windowAssigner instanceof MergingWindowAssigner) {
+ MergingWindowSet<W> mergingWindows = getMergingWindowSet();
+
+ for (W window : elementWindows) {
+
+ // adding the new window might result in a merge, in that case the actualWindow
+ // is the merged window and we work with that. If we don't merge then
+ // actualWindow == window
+ W actualWindow =
+ mergingWindows.addWindow(
+ window,
+ new MergingWindowSet.MergeFunction<>() {
+ @Override
+ public void merge(
+ W mergeResult,
+ Collection<W> mergedWindows,
+ W stateWindowResult,
+ Collection<W> mergedStateWindows)
+ throws Exception {
+
+ if ((windowAssigner.isEventTime()
+ && mergeResult.maxTimestamp() + allowedLateness
+ <= internalTimerService
+ .currentWatermark())) {
+ throw new UnsupportedOperationException(
+ "The end timestamp of an "
+ + "event-time window cannot become earlier than the current watermark "
+ + "by merging. Current event time: "
+ + internalTimerService
+ .currentWatermark()
+ + " window: "
+ + mergeResult);
+ } else if (!windowAssigner.isEventTime()) {
+ long currentProcessingTime =
+ internalTimerService.currentProcessingTime();
+ if (mergeResult.maxTimestamp()
+ <= currentProcessingTime) {
+ throw new UnsupportedOperationException(
+ "The end timestamp of a "
+ + "processing-time window cannot become earlier than the current processing time "
+ + "by merging. Current processing time: "
+ + currentProcessingTime
+ + " window: "
+ + mergeResult);
+ }
+ }
+
+ triggerContext.setKey(key);
+ triggerContext.setWindow(mergeResult);
+
+ triggerContext.onMerge(mergedWindows);
+
+ for (W m : mergedWindows) {
+ triggerContext.setWindow(m);
+ triggerContext.clear();
+ WindowUtils.deleteCleanupTimer(
+ m,
+ windowAssigner,
+ triggerContext,
+ allowedLateness);
+ }
+
+ // merge the merged state windows into the newly resulting
+ // state window
+ leftWindowMergingState.mergeNamespaces(
+ stateWindowResult, mergedStateWindows);
+ rightWindowMergingState.mergeNamespaces(
+ stateWindowResult, mergedStateWindows);
+ }
+ });
+
+ // drop if the window is already late
+ if (WindowUtils.isWindowLate(
+ actualWindow, windowAssigner, internalTimerService, allowedLateness)) {
+ mergingWindows.retireWindow(actualWindow);
+ continue;
+ }
+ isSkippedElement = false;
+
+ W stateWindow = mergingWindows.getStateWindow(actualWindow);
+ if (stateWindow == null) {
+ throw new IllegalStateException(
+ "Window " + window + " is not in in-flight window set.");
+ }
+
+ leftWindowState.setCurrentNamespace(stateWindow);
+ collector.setTimestamp(window.maxTimestamp());
+ windowFunctionContext.setWindow(window);
+ windowProcessFunction.onRecord1(
+ element.getValue(), collector, partitionedContext, windowFunctionContext);
+
+ triggerContext.setKey(key);
+ triggerContext.setWindow(actualWindow);
+
+ TriggerResult triggerResult =
+ triggerContext.onElement(
+ new StreamRecord<>(
+ TaggedUnion.one(element.getValue()),
+ element.getTimestamp()));
+
+ if (triggerResult.isFire()) {
+ emitWindowContents(actualWindow);
+ }
+
+ if (triggerResult.isPurge()) {
+ leftWindowState.clear();
+ rightWindowState.clear();
+ }
+
+ WindowUtils.registerCleanupTimer(
+ actualWindow, windowAssigner, triggerContext, allowedLateness);
+ }
+
+ // need to make sure to update the merging state in state
+ mergingWindows.persist();
+ } else {
+
+ for (W window : elementWindows) {
+ // drop if the window is already late
+ if (WindowUtils.isWindowLate(
+ window, windowAssigner, internalTimerService, allowedLateness)) {
+ continue;
+ }
+ isSkippedElement = false;
+
+ leftWindowState.setCurrentNamespace(window);
+ collector.setTimestamp(window.maxTimestamp());
+ windowFunctionContext.setWindow(window);
+ windowProcessFunction.onRecord1(
+ element.getValue(), collector, partitionedContext, windowFunctionContext);
+
+ triggerContext.setKey(key);
+ triggerContext.setWindow(window);
+
+ TriggerResult triggerResult =
+ triggerContext.onElement(
+ new StreamRecord<>(
+ TaggedUnion.one(element.getValue()),
+ element.getTimestamp()));
+
+ if (triggerResult.isFire()) {
+ emitWindowContents(window);
+ }
+
+ if (triggerResult.isPurge()) {
+ leftWindowState.clear();
+ rightWindowState.clear();
+ }
+
+ WindowUtils.registerCleanupTimer(
+ window, windowAssigner, triggerContext, allowedLateness);
+ }
+ }
+
+ // side output input event if element not handled by any window late arriving tag has been
+ // set windowAssigner is event time and current timestamp + allowed lateness no less than
+ // element timestamp.
+ if (isSkippedElement
+ && WindowUtils.isElementLate(
+ element, windowAssigner, allowedLateness, internalTimerService)) {
+ windowProcessFunction.onLateRecord1(element.getValue(), collector, partitionedContext);
+ }
+ }
+
+ @Override
+ public void processElement2(StreamRecord<IN2> element) throws Exception {
+ final Collection<W> elementWindows =
+ windowAssigner.assignWindows(
+ TaggedUnion.two(element.getValue()),
+ element.getTimestamp(),
+ windowAssignerContext);
+
+ // if element is handled by none of assigned elementWindows
+ boolean isSkippedElement = true;
+
+ final K key = (K) this.getCurrentKey();
+
+ if (windowAssigner instanceof MergingWindowAssigner) {
+ MergingWindowSet<W> mergingWindows = getMergingWindowSet();
+
+ for (W window : elementWindows) {
+
+ // adding the new window might result in a merge, in that case the actualWindow
+ // is the merged window and we work with that. If we don't merge then
+ // actualWindow == window
+ W actualWindow =
+ mergingWindows.addWindow(
+ window,
+ new MergingWindowSet.MergeFunction<>() {
+ @Override
+ public void merge(
+ W mergeResult,
+ Collection<W> mergedWindows,
+ W stateWindowResult,
+ Collection<W> mergedStateWindows)
+ throws Exception {
+
+ if ((windowAssigner.isEventTime()
+ && mergeResult.maxTimestamp() + allowedLateness
+ <= internalTimerService
+ .currentWatermark())) {
+ throw new UnsupportedOperationException(
+ "The end timestamp of an "
+ + "event-time window cannot become earlier than the current watermark "
+ + "by merging. Current event time: "
+ + internalTimerService
+ .currentWatermark()
+ + " window: "
+ + mergeResult);
+ } else if (!windowAssigner.isEventTime()) {
+ long currentProcessingTime =
+ internalTimerService.currentProcessingTime();
+ if (mergeResult.maxTimestamp()
+ <= currentProcessingTime) {
+ throw new UnsupportedOperationException(
+ "The end timestamp of a "
+ + "processing-time window cannot become earlier than the current processing time "
+ + "by merging. Current processing time: "
+ + currentProcessingTime
+ + " window: "
+ + mergeResult);
+ }
+ }
+
+ triggerContext.setKey(key);
+ triggerContext.setWindow(mergeResult);
+
+ triggerContext.onMerge(mergedWindows);
+
+ for (W m : mergedWindows) {
+ triggerContext.setWindow(m);
+ triggerContext.clear();
+ WindowUtils.deleteCleanupTimer(
+ m,
+ windowAssigner,
+ triggerContext,
+ allowedLateness);
+ }
+
+ // merge the merged state windows into the newly resulting
+ // state window
+ leftWindowMergingState.mergeNamespaces(
+ stateWindowResult, mergedStateWindows);
+ rightWindowMergingState.mergeNamespaces(
+ stateWindowResult, mergedStateWindows);
+ }
+ });
+
+ // drop if the window is already late
+ if (WindowUtils.isWindowLate(
+ actualWindow, windowAssigner, internalTimerService, allowedLateness)) {
+ mergingWindows.retireWindow(actualWindow);
+ continue;
+ }
+ isSkippedElement = false;
+
+ W stateWindow = mergingWindows.getStateWindow(actualWindow);
+ if (stateWindow == null) {
+ throw new IllegalStateException(
+ "Window " + window + " is not in in-flight window set.");
+ }
+
+ rightWindowState.setCurrentNamespace(stateWindow);
+ collector.setTimestamp(window.maxTimestamp());
+ windowFunctionContext.setWindow(window);
+ windowProcessFunction.onRecord2(
+ element.getValue(), collector, partitionedContext, windowFunctionContext);
+
+ triggerContext.setKey(key);
+ triggerContext.setWindow(actualWindow);
+
+ TriggerResult triggerResult =
+ triggerContext.onElement(
+ new StreamRecord<>(
+ TaggedUnion.two(element.getValue()),
+ element.getTimestamp()));
+
+ if (triggerResult.isFire()) {
+ emitWindowContents(actualWindow);
+ }
+
+ if (triggerResult.isPurge()) {
+ leftWindowState.clear();
+ rightWindowState.clear();
+ }
+
+ WindowUtils.registerCleanupTimer(
+ actualWindow, windowAssigner, triggerContext, allowedLateness);
+ }
+
+ // need to make sure to update the merging state in state
+ mergingWindows.persist();
+ } else {
+
+ for (W window : elementWindows) {
+ // drop if the window is already late
+ if (WindowUtils.isWindowLate(
+ window, windowAssigner, internalTimerService, allowedLateness)) {
+ continue;
+ }
+ isSkippedElement = false;
+
+ rightWindowState.setCurrentNamespace(window);
+ collector.setTimestamp(window.maxTimestamp());
+ windowFunctionContext.setWindow(window);
+ windowProcessFunction.onRecord2(
+ element.getValue(), collector, partitionedContext, windowFunctionContext);
+
+ triggerContext.setKey(key);
+ triggerContext.setWindow(window);
+
+ TriggerResult triggerResult =
+ triggerContext.onElement(
+ new StreamRecord<>(
+ TaggedUnion.two(element.getValue()),
+ element.getTimestamp()));
+
+ if (triggerResult.isFire()) {
+ emitWindowContents(window);
+ }
+
+ if (triggerResult.isPurge()) {
+ leftWindowState.clear();
+ rightWindowState.clear();
+ }
+
+ WindowUtils.registerCleanupTimer(
+ window, windowAssigner, triggerContext, allowedLateness);
+ }
+ }
+
+ // side output input event if element not handled by any window late arriving tag has been
+ // set windowAssigner is event time and current timestamp + allowed lateness no less than
+ // element timestamp.
+ if (isSkippedElement
+ && WindowUtils.isElementLate(
+ element, windowAssigner, allowedLateness, internalTimerService)) {
+ windowProcessFunction.onLateRecord2(element.getValue(), collector, partitionedContext);
+ }
+ }
+
+ @Override
+ public void onEventTime(InternalTimer<K, W> timer) throws Exception {
+ triggerContext.setKey(timer.getKey());
+ triggerContext.setWindow(timer.getNamespace());
+
+ MergingWindowSet<W> mergingWindows;
+
+ if (windowAssigner instanceof MergingWindowAssigner) {
+ mergingWindows = getMergingWindowSet();
+ W stateWindow = mergingWindows.getStateWindow(triggerContext.getWindow());
+ if (stateWindow == null) {
+ // Timer firing for non-existent window, this can only happen if a
+ // trigger did not clean up timers. We have already cleared the merging
+ // window and therefore the Trigger state, however, so nothing to do.
+ return;
+ } else {
+ leftWindowState.setCurrentNamespace(stateWindow);
+ rightWindowState.setCurrentNamespace(stateWindow);
+ }
+ } else {
+ leftWindowState.setCurrentNamespace(triggerContext.getWindow());
+ rightWindowState.setCurrentNamespace(triggerContext.getWindow());
+ mergingWindows = null;
+ }
+
+ TriggerResult triggerResult = triggerContext.onEventTime(timer.getTimestamp());
+
+ if (triggerResult.isFire()) {
+ emitWindowContents(triggerContext.getWindow());
+ }
+
+ if (triggerResult.isPurge()) {
+ leftWindowState.clear();
+ rightWindowState.clear();
+ }
+
+ if (windowAssigner.isEventTime()
+ && WindowUtils.isCleanupTime(
+ triggerContext.getWindow(),
+ timer.getTimestamp(),
+ windowAssigner,
+ allowedLateness)) {
+ clearAllState(
+ triggerContext.getWindow(), leftWindowState, rightWindowState, mergingWindows);
+ }
+
+ if (mergingWindows != null) {
+ // need to make sure to update the merging state in state
+ mergingWindows.persist();
+ }
+ }
+
+ @Override
+ public void onProcessingTime(InternalTimer<K, W> timer) throws Exception {
+ triggerContext.setKey(timer.getKey());
+ triggerContext.setWindow(timer.getNamespace());
+
+ MergingWindowSet<W> mergingWindows;
+
+ if (windowAssigner instanceof MergingWindowAssigner) {
+ mergingWindows = getMergingWindowSet();
+ W stateWindow = mergingWindows.getStateWindow(triggerContext.getWindow());
+ if (stateWindow == null) {
+ // Timer firing for non-existent window, this can only happen if a
+ // trigger did not clean up timers. We have already cleared the merging
+ // window and therefore the Trigger state, however, so nothing to do.
+ return;
+ } else {
+ leftWindowState.setCurrentNamespace(stateWindow);
+ rightWindowState.setCurrentNamespace(stateWindow);
+ }
+ } else {
+ leftWindowState.setCurrentNamespace(triggerContext.getWindow());
+ rightWindowState.setCurrentNamespace(triggerContext.getWindow());
+ mergingWindows = null;
+ }
+
+ TriggerResult triggerResult = triggerContext.onProcessingTime(timer.getTimestamp());
+
+ if (triggerResult.isFire()) {
+ emitWindowContents(triggerContext.getWindow());
+ }
+
+ if (triggerResult.isPurge()) {
+ leftWindowState.clear();
+ rightWindowState.clear();
+ }
+
+ if (!windowAssigner.isEventTime()
+ && WindowUtils.isCleanupTime(
+ triggerContext.getWindow(),
+ timer.getTimestamp(),
+ windowAssigner,
+ allowedLateness)) {
+ clearAllState(
+ triggerContext.getWindow(), leftWindowState, rightWindowState, mergingWindows);
+ }
+
+ if (mergingWindows != null) {
+ // need to make sure to update the merging state in state
+ mergingWindows.persist();
+ }
+ }
+
+ @Override
+ protected ProcessingTimeManager getProcessingTimeManager() {
+ // we don't support user utilize processing time in window operators
+ return UnsupportedProcessingTimeManager.INSTANCE;
+ }
+
+ /**
+ * Drops all state for the given window and calls {@link Trigger#clear(Window,
+ * Trigger.TriggerContext)}.
+ *
+ * <p>The caller must ensure that the correct key is set in the state backend and the
+ * triggerContext object.
+ */
+ private void clearAllState(
+ W window,
+ AppendingState<IN1, StateIterator<IN1>, Iterable<IN1>> leftWindowState,
+ AppendingState<IN2, StateIterator<IN2>, Iterable<IN2>> rightWindowState,
+ MergingWindowSet<W> mergingWindows)
+ throws Exception {
+ leftWindowState.clear();
+ rightWindowState.clear();
+ triggerContext.clear();
+ windowFunctionContext.setWindow(window);
+ windowProcessFunction.onClear(collector, partitionedContext, windowFunctionContext);
+ if (mergingWindows != null) {
+ mergingWindows.retireWindow(window);
+ mergingWindows.persist();
+ }
+ }
+
+ /**
+ * Emits the contents of the given window using the user-defined {@link
+ * TwoInputNonBroadcastWindowStreamProcessFunction}.
+ */
+ private void emitWindowContents(W window) throws Exception {
+ // only time window touch the time concept.
+ collector.setTimestamp(window.maxTimestamp());
+ windowFunctionContext.setWindow(window);
+ windowProcessFunction.onTrigger(collector, partitionedContext, windowFunctionContext);
+ }
+
+ /**
+ * Retrieves the {@link MergingWindowSet} for the currently active key. The caller must ensure
+ * that the correct key is set in the state backend.
+ *
+ * <p>The caller must also ensure to properly persist changes to state using {@link
+ * MergingWindowSet#persist()}.
+ */
+ protected MergingWindowSet<W> getMergingWindowSet() throws Exception {
+ MergingWindowAssigner<? super TaggedUnion<IN1, IN2>, W> mergingAssigner =
+ (MergingWindowAssigner<? super TaggedUnion<IN1, IN2>, W>) windowAssigner;
+ return new MergingWindowSet<>(mergingAssigner, mergingSetsState);
+ }
+}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/TwoOutputWindowProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/TwoOutputWindowProcessOperator.java
new file mode 100644
index 00000000000..bb564f202a1
--- /dev/null
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/TwoOutputWindowProcessOperator.java
@@ -0,0 +1,549 @@
+/*
+ * 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.datastream.impl.extension.window.operators;
+
+import org.apache.flink.api.common.state.v2.AppendingState;
+import org.apache.flink.api.common.state.v2.ListStateDescriptor;
+import org.apache.flink.api.common.state.v2.StateDescriptor;
+import org.apache.flink.api.common.state.v2.StateIterator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
+import org.apache.flink.datastream.api.context.ProcessingTimeManager;
+import org.apache.flink.datastream.api.extension.window.function.TwoOutputWindowStreamProcessFunction;
+import org.apache.flink.datastream.api.extension.window.function.WindowProcessFunction;
+import org.apache.flink.datastream.api.stream.KeyedPartitionStream;
+import org.apache.flink.datastream.impl.context.UnsupportedProcessingTimeManager;
+import org.apache.flink.datastream.impl.extension.window.context.DefaultOneInputWindowContext;
+import org.apache.flink.datastream.impl.extension.window.context.WindowTriggerContext;
+import org.apache.flink.datastream.impl.extension.window.function.InternalTwoOutputWindowStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.utils.WindowUtils;
+import org.apache.flink.datastream.impl.operators.BaseKeyedTwoOutputProcessOperator;
+import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.runtime.state.VoidNamespaceSerializer;
+import org.apache.flink.runtime.state.v2.internal.InternalAppendingState;
+import org.apache.flink.runtime.state.v2.internal.InternalListState;
+import org.apache.flink.runtime.state.v2.internal.InternalMergingState;
+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.windowing.assigners.MergingWindowAssigner;
+import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
+import org.apache.flink.streaming.api.windowing.windows.Window;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import javax.annotation.Nullable;
+
+import java.util.Collection;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/** Operator for {@link TwoOutputWindowStreamProcessFunction} in {@link KeyedPartitionStream}. */
+public class TwoOutputWindowProcessOperator<K, IN, OUT_MAIN, OUT_SIDE, W extends Window>
+ extends BaseKeyedTwoOutputProcessOperator<K, IN, OUT_MAIN, OUT_SIDE>
+ implements Triggerable<K, W> {
+
+ private static final long serialVersionUID = 1L;
+
+ // ------------------------------------------------------------------------
+ // Configuration values and user functions
+ // ------------------------------------------------------------------------
+
+ private final TwoOutputWindowStreamProcessFunction<IN, OUT_MAIN, OUT_SIDE>
+ windowProcessFunction;
+
+ /**
+ * The allowed lateness for elements. This is used for:
+ *
+ * <ul>
+ * <li>Deciding if an element should be dropped from a window due to lateness.
+ * <li>Clearing the state of a window if the time out-of the {@code window.maxTimestamp +
+ * allowedLateness} landmark.
+ * </ul>
+ */
+ private final long allowedLateness;
+
+ // ------------------------------------------------------------------------
+ // Operator components
+ // ------------------------------------------------------------------------
+
+ private transient InternalTimerService<W> internalTimerService;
+
+ /** For serializing the window in checkpoints. */
+ private final TypeSerializer<W> windowSerializer;
+
+ // ------------------------------------------------------------------------
+ // Window assigner and trigger
+ // ------------------------------------------------------------------------
+
+ private final WindowAssigner<? super IN, W> windowAssigner;
+
+ private transient WindowAssigner.WindowAssignerContext windowAssignerContext;
+
+ private final Trigger<? super IN, ? super W> trigger;
+
+ private transient WindowTriggerContext<K, IN, W> triggerContext;
+
+ private transient DefaultOneInputWindowContext<K, IN, W> windowFunctionContext;
+
+ // ------------------------------------------------------------------------
+ // State that is not checkpointed
+ // ------------------------------------------------------------------------
+
+ private final StateDescriptor<IN> windowStateDescriptor;
+
+ /** The state in which the window contents is stored. Each window is a namespace */
+ private transient InternalAppendingState<K, W, IN, IN, StateIterator<IN>, Iterable<IN>>
+ windowState;
+
+ /**
+ * The {@link #windowState}, typed to merging state for merging windows. Null if the window
+ * state is not mergeable.
+ */
+ private transient InternalMergingState<K, W, IN, IN, StateIterator<IN>, Iterable<IN>>
+ windowMergingState;
+
+ /** The state that holds the merging window metadata (the sets that describe what is merged). */
+ private transient InternalListState<K, VoidNamespace, Tuple2<W, W>> mergingSetsState;
+
+ public TwoOutputWindowProcessOperator(
+ InternalTwoOutputWindowStreamProcessFunction<IN, OUT_MAIN, OUT_SIDE, W>
+ windowProcessFunction,
+ OutputTag<OUT_SIDE> outputTag,
+ @Nullable KeySelector<OUT_MAIN, K> mainOutKeySelector,
+ @Nullable KeySelector<OUT_SIDE, K> sideOutKeySelector,
+ WindowAssigner<? super IN, W> windowAssigner,
+ Trigger<? super IN, ? super W> trigger,
+ TypeSerializer<W> windowSerializer,
+ StateDescriptor<IN> windowStateDescriptor,
+ long allowedLateness) {
+ super(windowProcessFunction, outputTag, mainOutKeySelector, sideOutKeySelector);
+ checkArgument(allowedLateness >= 0);
+ this.windowProcessFunction = windowProcessFunction.getWindowProcessFunction();
+ this.windowAssigner = windowAssigner;
+ this.trigger = trigger;
+ this.windowSerializer = windowSerializer;
+ this.windowStateDescriptor = windowStateDescriptor;
+ this.allowedLateness = allowedLateness;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ internalTimerService =
+ getInternalTimerService("process-window-timers", windowSerializer, this);
+
+ // create (or restore) the state that hold the actual window contents
+ // NOTE - the state may be null in the case of the overriding evicting window operator
+ if (windowStateDescriptor != null) {
+ windowState =
+ getOrCreateKeyedState(
+ windowSerializer.createInstance(),
+ windowSerializer,
+ windowStateDescriptor);
+ }
+
+ // create the typed and helper states for merging windows
+ if (windowAssigner instanceof MergingWindowAssigner) {
+
+ // store a typed reference for the state of merging windows - sanity check
+ if (windowState instanceof InternalMergingState) {
+ windowMergingState =
+ (InternalMergingState<K, W, IN, IN, StateIterator<IN>, Iterable<IN>>)
+ windowState;
+ } else if (windowState != null) {
+ throw new IllegalStateException(
+ "The window uses a merging assigner, but the window state is not mergeable.");
+ }
+
+ @SuppressWarnings("unchecked")
+ final Class<Tuple2<W, W>> typedTuple = (Class<Tuple2<W, W>>) (Class<?>) Tuple2.class;
+
+ final TupleSerializer<Tuple2<W, W>> tupleSerializer =
+ new TupleSerializer<>(
+ typedTuple, new TypeSerializer[] {windowSerializer, windowSerializer});
+
+ final ListStateDescriptor<Tuple2<W, W>> mergingSetsStateDescriptor =
+ new ListStateDescriptor<>("merging-window-set", tupleSerializer);
+
+ // get the state that stores the merging sets
+ mergingSetsState =
+ getOrCreateKeyedState(
+ VoidNamespaceSerializer.INSTANCE.createInstance(),
+ VoidNamespaceSerializer.INSTANCE,
+ mergingSetsStateDescriptor);
+ mergingSetsState.setCurrentNamespace(VoidNamespace.INSTANCE);
+ }
+
+ triggerContext =
+ new WindowTriggerContext<>(
+ null, null, this, internalTimerService, trigger, windowSerializer);
+ windowAssignerContext =
+ new WindowAssigner.WindowAssignerContext() {
+ @Override
+ public long getCurrentProcessingTime() {
+ return internalTimerService.currentProcessingTime();
+ }
+ };
+ windowFunctionContext =
+ new DefaultOneInputWindowContext<>(
+ null,
+ windowState,
+ windowProcessFunction,
+ this,
+ windowSerializer,
+ windowMergingState != null);
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ triggerContext = null;
+ windowFunctionContext = null;
+ windowAssignerContext = null;
+ }
+
+ @Override
+ public void processElement(StreamRecord<IN> element) throws Exception {
+ final Collection<W> elementWindows =
+ windowAssigner.assignWindows(
+ element.getValue(), element.getTimestamp(), windowAssignerContext);
+
+ // if element is handled by none of assigned elementWindows
+ boolean isSkippedElement = true;
+
+ final K key = (K) this.getCurrentKey();
+
+ if (windowAssigner instanceof MergingWindowAssigner) {
+ MergingWindowSet<W> mergingWindows = getMergingWindowSet();
+
+ for (W window : elementWindows) {
+
+ // adding the new window might result in a merge, in that case the actualWindow
+ // is the merged window and we work with that. If we don't merge then
+ // actualWindow == window
+ W actualWindow =
+ mergingWindows.addWindow(
+ window,
+ new MergingWindowSet.MergeFunction<>() {
+ @Override
+ public void merge(
+ W mergeResult,
+ Collection<W> mergedWindows,
+ W stateWindowResult,
+ Collection<W> mergedStateWindows)
+ throws Exception {
+
+ if ((windowAssigner.isEventTime()
+ && mergeResult.maxTimestamp() + allowedLateness
+ <= internalTimerService
+ .currentWatermark())) {
+ throw new UnsupportedOperationException(
+ "The end timestamp of an "
+ + "event-time window cannot become earlier than the current watermark "
+ + "by merging. Current event time: "
+ + internalTimerService
+ .currentWatermark()
+ + " window: "
+ + mergeResult);
+ } else if (!windowAssigner.isEventTime()) {
+ long currentProcessingTime =
+ internalTimerService.currentProcessingTime();
+ if (mergeResult.maxTimestamp()
+ <= currentProcessingTime) {
+ throw new UnsupportedOperationException(
+ "The end timestamp of a "
+ + "processing-time window cannot become earlier than the current processing time "
+ + "by merging. Current processing time: "
+ + currentProcessingTime
+ + " window: "
+ + mergeResult);
+ }
+ }
+
+ triggerContext.setKey(key);
+ triggerContext.setWindow(mergeResult);
+
+ triggerContext.onMerge(mergedWindows);
+
+ for (W m : mergedWindows) {
+ triggerContext.setWindow(m);
+ triggerContext.clear();
+ WindowUtils.deleteCleanupTimer(
+ m,
+ windowAssigner,
+ triggerContext,
+ allowedLateness);
+ }
+
+ // merge the merged state windows into the newly resulting
+ // state window
+ windowMergingState.mergeNamespaces(
+ stateWindowResult, mergedStateWindows);
+ }
+ });
+
+ // drop if the window is already late
+ if (WindowUtils.isWindowLate(
+ actualWindow, windowAssigner, internalTimerService, allowedLateness)) {
+ mergingWindows.retireWindow(actualWindow);
+ continue;
+ }
+ isSkippedElement = false;
+
+ W stateWindow = mergingWindows.getStateWindow(actualWindow);
+ if (stateWindow == null) {
+ throw new IllegalStateException(
+ "Window " + window + " is not in in-flight window set.");
+ }
+
+ windowState.setCurrentNamespace(stateWindow);
+ mainCollector.setTimestamp(window.maxTimestamp());
+ sideCollector.setTimestamp(window.maxTimestamp());
+ windowFunctionContext.setWindow(window);
+ windowProcessFunction.onRecord(
+ element.getValue(),
+ mainCollector,
+ sideCollector,
+ partitionedContext,
+ windowFunctionContext);
+
+ triggerContext.setKey(key);
+ triggerContext.setWindow(actualWindow);
+
+ TriggerResult triggerResult = triggerContext.onElement(element);
+
+ if (triggerResult.isFire()) {
+ emitWindowContents(actualWindow);
+ }
+
+ if (triggerResult.isPurge()) {
+ windowState.clear();
+ }
+
+ WindowUtils.registerCleanupTimer(
+ actualWindow, windowAssigner, triggerContext, allowedLateness);
+ }
+
+ // need to make sure to update the merging state in state
+ mergingWindows.persist();
+ } else {
+ for (W window : elementWindows) {
+
+ // drop if the window is already late
+ if (WindowUtils.isWindowLate(
+ window, windowAssigner, internalTimerService, allowedLateness)) {
+ continue;
+ }
+ isSkippedElement = false;
+
+ windowState.setCurrentNamespace(window);
+ mainCollector.setTimestamp(window.maxTimestamp());
+ sideCollector.setTimestamp(window.maxTimestamp());
+ windowFunctionContext.setWindow(window);
+ windowProcessFunction.onRecord(
+ element.getValue(),
+ mainCollector,
+ sideCollector,
+ partitionedContext,
+ windowFunctionContext);
+
+ triggerContext.setKey(key);
+ triggerContext.setWindow(window);
+
+ TriggerResult triggerResult = triggerContext.onElement(element);
+
+ if (triggerResult.isFire()) {
+ emitWindowContents(window);
+ }
+
+ if (triggerResult.isPurge()) {
+ windowState.clear();
+ }
+
+ WindowUtils.registerCleanupTimer(
+ window, windowAssigner, triggerContext, allowedLateness);
+ }
+ }
+
+ // side output input event if element not handled by any window late arriving tag has been
+ // set windowAssigner is event time and current timestamp + allowed lateness no less than
+ // element timestamp.
+ if (isSkippedElement
+ && WindowUtils.isElementLate(
+ element, windowAssigner, allowedLateness, internalTimerService)) {
+ windowProcessFunction.onLateRecord(
+ element.getValue(), mainCollector, sideCollector, partitionedContext);
+ }
+ }
+
+ @Override
+ public void onEventTime(InternalTimer<K, W> timer) throws Exception {
+ triggerContext.setKey(timer.getKey());
+ triggerContext.setWindow(timer.getNamespace());
+
+ MergingWindowSet<W> mergingWindows;
+
+ if (windowAssigner instanceof MergingWindowAssigner) {
+ mergingWindows = getMergingWindowSet();
+ W stateWindow = mergingWindows.getStateWindow(triggerContext.getWindow());
+ if (stateWindow == null) {
+ // Timer firing for non-existent window, this can only happen if a
+ // trigger did not clean up timers. We have already cleared the merging
+ // window and therefore the Trigger state, however, so nothing to do.
+ return;
+ } else {
+ windowState.setCurrentNamespace(stateWindow);
+ }
+ } else {
+ windowState.setCurrentNamespace(triggerContext.getWindow());
+ mergingWindows = null;
+ }
+
+ TriggerResult triggerResult = triggerContext.onEventTime(timer.getTimestamp());
+
+ if (triggerResult.isFire()) {
+ emitWindowContents(triggerContext.getWindow());
+ }
+
+ if (triggerResult.isPurge()) {
+ windowState.clear();
+ }
+
+ if (windowAssigner.isEventTime()
+ && WindowUtils.isCleanupTime(
+ triggerContext.getWindow(),
+ timer.getTimestamp(),
+ windowAssigner,
+ allowedLateness)) {
+ clearAllState(triggerContext.getWindow(), windowState, mergingWindows);
+ }
+
+ if (mergingWindows != null) {
+ // need to make sure to update the merging state in state
+ mergingWindows.persist();
+ }
+ }
+
+ @Override
+ public void onProcessingTime(InternalTimer<K, W> timer) throws Exception {
+ triggerContext.setKey(timer.getKey());
+ triggerContext.setWindow(timer.getNamespace());
+
+ MergingWindowSet<W> mergingWindows;
+
+ if (windowAssigner instanceof MergingWindowAssigner) {
+ mergingWindows = getMergingWindowSet();
+ W stateWindow = mergingWindows.getStateWindow(triggerContext.getWindow());
+ if (stateWindow == null) {
+ // Timer firing for non-existent window, this can only happen if a
+ // trigger did not clean up timers. We have already cleared the merging
+ // window and therefore the Trigger state, however, so nothing to do.
+ return;
+ } else {
+ windowState.setCurrentNamespace(stateWindow);
+ }
+ } else {
+ windowState.setCurrentNamespace(triggerContext.getWindow());
+ mergingWindows = null;
+ }
+
+ TriggerResult triggerResult = triggerContext.onProcessingTime(timer.getTimestamp());
+
+ if (triggerResult.isFire()) {
+ emitWindowContents(triggerContext.getWindow());
+ }
+
+ if (triggerResult.isPurge()) {
+ windowState.clear();
+ }
+
+ if (!windowAssigner.isEventTime()
+ && WindowUtils.isCleanupTime(
+ triggerContext.getWindow(),
+ timer.getTimestamp(),
+ windowAssigner,
+ allowedLateness)) {
+ clearAllState(triggerContext.getWindow(), windowState, mergingWindows);
+ }
+
+ if (mergingWindows != null) {
+ // need to make sure to update the merging state in state
+ mergingWindows.persist();
+ }
+ }
+
+ @Override
+ protected ProcessingTimeManager getProcessingTimeManager() {
+ // we don't support user utilize processing time in window operators
+ return UnsupportedProcessingTimeManager.INSTANCE;
+ }
+
+ /**
+ * Drops all state for the given window and calls {@link Trigger#clear(Window,
+ * Trigger.TriggerContext)}.
+ *
+ * <p>The caller must ensure that the correct key is set in the state backend and the
+ * triggerContext object.
+ */
+ private void clearAllState(
+ W window,
+ AppendingState<IN, StateIterator<IN>, Iterable<IN>> windowState,
+ MergingWindowSet<W> mergingWindows)
+ throws Exception {
+ windowState.clear();
+ triggerContext.clear();
+ windowFunctionContext.setWindow(window);
+ windowProcessFunction.onClear(
+ mainCollector, sideCollector, partitionedContext, windowFunctionContext);
+ if (mergingWindows != null) {
+ mergingWindows.retireWindow(window);
+ mergingWindows.persist();
+ }
+ }
+
+ /**
+ * Emits the contents of the given window using the user-defined {@link WindowProcessFunction}.
+ */
+ private void emitWindowContents(W window) throws Exception {
+ // only time window touch the time concept.
+ mainCollector.setTimestamp(window.maxTimestamp());
+ sideCollector.setTimestamp(window.maxTimestamp());
+ windowFunctionContext.setWindow(window);
+ windowProcessFunction.onTrigger(
+ mainCollector, sideCollector, partitionedContext, windowFunctionContext);
+ }
+
+ /**
+ * Retrieves the {@link MergingWindowSet} for the currently active key. The caller must ensure
+ * that the correct key is set in the state backend.
+ *
+ * <p>The caller must also ensure to properly persist changes to state using {@link
+ * MergingWindowSet#persist()}.
+ */
+ protected MergingWindowSet<W> getMergingWindowSet() throws Exception {
+ MergingWindowAssigner<? super IN, W> mergingAssigner =
+ (MergingWindowAssigner<? super IN, W>) windowAssigner;
+ return new MergingWindowSet<>(mergingAssigner, mergingSetsState);
+ }
+}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/utils/WindowUtils.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/utils/WindowUtils.java
new file mode 100644
index 00000000000..93f4f2bbadb
--- /dev/null
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/utils/WindowUtils.java
@@ -0,0 +1,230 @@
+/*
+ * 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.datastream.impl.extension.window.utils;
+
+import org.apache.flink.datastream.api.extension.window.strategy.GlobalWindowStrategy;
+import org.apache.flink.datastream.api.extension.window.strategy.SessionWindowStrategy;
+import org.apache.flink.datastream.api.extension.window.strategy.SlidingTimeWindowStrategy;
+import org.apache.flink.datastream.api.extension.window.strategy.TumblingTimeWindowStrategy;
+import org.apache.flink.datastream.api.extension.window.strategy.WindowStrategy;
+import org.apache.flink.datastream.impl.extension.window.context.WindowTriggerContext;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
+import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
+import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
+import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
+import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
+import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
+import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
+import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
+import org.apache.flink.streaming.api.windowing.windows.Window;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+/** Utilities for Window extension. */
+public final class WindowUtils {
+
+ /**
+ * Get the allowed lateness for the window. Note that, in the current design, only
+ * TumblingTimeWindow and SlidingTimeWindow can have allowed lateness configured; for other
+ * types of windows, the returned allowed lateness will be 0.
+ */
+ public static long getAllowedLateness(WindowStrategy windowStrategy) {
+ if (windowStrategy instanceof TumblingTimeWindowStrategy) {
+ return ((TumblingTimeWindowStrategy) windowStrategy).getAllowedLateness().toMillis();
+ } else if (windowStrategy instanceof SlidingTimeWindowStrategy) {
+ return ((SlidingTimeWindowStrategy) windowStrategy).getAllowedLateness().toMillis();
+ } else {
+ return 0;
+ }
+ }
+
+ /** Create a {@link WindowAssigner} according to {@link WindowStrategy}. */
+ public static WindowAssigner<?, ?> createWindowAssigner(WindowStrategy windowStrategy) {
+ if (windowStrategy instanceof GlobalWindowStrategy) {
+ return createGlobalWindowAssigner();
+ } else if (windowStrategy instanceof TumblingTimeWindowStrategy) {
+ return createTumblingTimeWindowAssigner((TumblingTimeWindowStrategy) windowStrategy);
+ } else if (windowStrategy instanceof SlidingTimeWindowStrategy) {
+ return createSlidingTimeWindowAssigner((SlidingTimeWindowStrategy) windowStrategy);
+ } else if (windowStrategy instanceof SessionWindowStrategy) {
+ return createSessionWindowAssigner((SessionWindowStrategy) windowStrategy);
+ } else {
+ throw new IllegalArgumentException(
+ "Unsupported type of window strategy : " + windowStrategy.getClass());
+ }
+ }
+
+ /** Create window assigner for global window. */
+ private static WindowAssigner<?, ?> createGlobalWindowAssigner() {
+ return GlobalWindows.createWithEndOfStreamTrigger();
+ }
+
+ /** Create window assigner for tumbling time window. */
+ private static WindowAssigner<?, ?> createTumblingTimeWindowAssigner(
+ TumblingTimeWindowStrategy windowStrategy) {
+ switch (windowStrategy.getTimeType()) {
+ case PROCESSING:
+ return TumblingProcessingTimeWindows.of(windowStrategy.getWindowSize());
+ case EVENT:
+ return TumblingEventTimeWindows.of(windowStrategy.getWindowSize());
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported time type : " + windowStrategy.getTimeType());
+ }
+ }
+
+ /** Create window assigner for sliding time window. */
+ private static WindowAssigner<?, ?> createSlidingTimeWindowAssigner(
+ SlidingTimeWindowStrategy windowStrategy) {
+ switch (windowStrategy.getTimeType()) {
+ case PROCESSING:
+ return SlidingProcessingTimeWindows.of(
+ windowStrategy.getWindowSize(), windowStrategy.getWindowSlideInterval());
+ case EVENT:
+ return SlidingEventTimeWindows.of(
+ windowStrategy.getWindowSize(), windowStrategy.getWindowSlideInterval());
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported time type : " + windowStrategy.getTimeType());
+ }
+ }
+
+ /** Create window assigner for session time window. */
+ private static WindowAssigner<?, ?> createSessionWindowAssigner(
+ SessionWindowStrategy windowStrategy) {
+ switch (windowStrategy.getTimeType()) {
+ case PROCESSING:
+ return ProcessingTimeSessionWindows.withGap(windowStrategy.getSessionGap());
+ case EVENT:
+ return EventTimeSessionWindows.withGap(windowStrategy.getSessionGap());
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported time type : " + windowStrategy.getTimeType());
+ }
+ }
+
+ /**
+ * Returns {@code true} if the event time is after the end timestamp plus the allowed lateness
+ * of the given window.
+ */
+ public static <W extends Window> boolean isWindowLate(
+ W window,
+ WindowAssigner<?, W> windowAssigner,
+ InternalTimerService<W> internalTimerService,
+ long allowedLateness) {
+ return (windowAssigner.isEventTime()
+ && (cleanupTime(window, windowAssigner, allowedLateness)
+ <= internalTimerService.currentWatermark()));
+ }
+
+ /**
+ * Decide if a record is currently late, based on current event time and allowed lateness.
+ *
+ * @param element The element to check
+ * @param windowAssigner The window assigner
+ * @param allowedLateness The allowed lateness of the window
+ * @param internalTimerService The timer service to get current event time
+ * @return The element for which should be considered late
+ */
+ public static boolean isElementLate(
+ StreamRecord<?> element,
+ WindowAssigner<?, ?> windowAssigner,
+ long allowedLateness,
+ InternalTimerService<?> internalTimerService) {
+ return (windowAssigner.isEventTime())
+ && (element.getTimestamp() + allowedLateness
+ <= internalTimerService.currentWatermark());
+ }
+
+ /**
+ * Deletes the cleanup timer set for the contents of the provided window.
+ *
+ * @param window the window whose state to discard
+ * @param windowAssigner The window assigner
+ * @param triggerContext The trigger context
+ * @param allowedLateness The allowed lateness of the window
+ */
+ public static <W extends Window> void deleteCleanupTimer(
+ W window,
+ WindowAssigner<?, ?> windowAssigner,
+ WindowTriggerContext<?, ?, W> triggerContext,
+ long allowedLateness) {
+ long cleanupTime = cleanupTime(window, windowAssigner, allowedLateness);
+ if (cleanupTime == Long.MAX_VALUE) {
+ // no need to clean up because we didn't set one
+ return;
+ }
+ if (windowAssigner.isEventTime()) {
+ triggerContext.deleteEventTimeTimer(cleanupTime);
+ } else {
+ triggerContext.deleteProcessingTimeTimer(cleanupTime);
+ }
+ }
+
+ /**
+ * Registers a timer to cleanup the content of the window.
+ *
+ * @param window the window whose state to discard
+ * @param windowAssigner The window assigner
+ * @param triggerContext The trigger context
+ * @param allowedLateness The allowed lateness of the window
+ */
+ public static <W extends Window> void registerCleanupTimer(
+ W window,
+ WindowAssigner<?, ?> windowAssigner,
+ WindowTriggerContext<?, ?, W> triggerContext,
+ long allowedLateness) {
+ long cleanupTime = cleanupTime(window, windowAssigner, allowedLateness);
+ if (cleanupTime == Long.MAX_VALUE) {
+ // don't set a GC timer for "end of time"
+ return;
+ }
+
+ if (windowAssigner.isEventTime()) {
+ triggerContext.registerEventTimeTimer(cleanupTime);
+ } else {
+ triggerContext.registerProcessingTimeTimer(cleanupTime);
+ }
+ }
+
+ /**
+ * Returns the cleanup time for a window, which is {@code window.maxTimestamp +
+ * allowedLateness}. In case this leads to a value greater than {@link Long#MAX_VALUE} then a
+ * cleanup time of {@link Long#MAX_VALUE} is returned.
+ *
+ * @param window the window whose cleanup time we are computing.
+ * @param windowAssigner The window assigner
+ * @param allowedLateness The allowed lateness of the window
+ */
+ private static <W extends Window> long cleanupTime(
+ W window, WindowAssigner<?, ?> windowAssigner, long allowedLateness) {
+ if (windowAssigner.isEventTime()) {
+ long cleanupTime = window.maxTimestamp() + allowedLateness;
+ return cleanupTime >= window.maxTimestamp() ? cleanupTime : Long.MAX_VALUE;
+ } else {
+ return window.maxTimestamp();
+ }
+ }
+
+ /** Returns {@code true} if the given time is the cleanup time for the given window. */
+ public static <W extends Window> boolean isCleanupTime(
+ W window, long time, WindowAssigner<?, ?> windowAssigner, long allowedLateness) {
+ return time == cleanupTime(window, windowAssigner, allowedLateness);
+ }
+}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/BaseKeyedProcessOperator.java
similarity index 57%
copy from flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedProcessOperator.java
copy to flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/BaseKeyedProcessOperator.java
index 23bae4b9921..d47c3a0d231 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedProcessOperator.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/BaseKeyedProcessOperator.java
@@ -20,43 +20,31 @@ package org.apache.flink.datastream.impl.operators;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.datastream.api.context.NonPartitionedContext;
-import org.apache.flink.datastream.api.context.ProcessingTimeManager;
import org.apache.flink.datastream.api.function.OneInputStreamProcessFunction;
import org.apache.flink.datastream.api.stream.KeyedPartitionStream;
import org.apache.flink.datastream.impl.common.KeyCheckedOutputCollector;
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.DefaultProcessingTimeManager;
-import org.apache.flink.datastream.impl.extension.eventtime.functions.EventTimeWrappedOneInputStreamProcessFunction;
-import org.apache.flink.runtime.state.VoidNamespace;
-import org.apache.flink.runtime.state.VoidNamespaceSerializer;
-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.runtime.streamrecord.StreamRecord;
import javax.annotation.Nullable;
import java.util.HashSet;
import java.util.Set;
-import java.util.function.Supplier;
-
-/** Operator for {@link OneInputStreamProcessFunction} in {@link KeyedPartitionStream}. */
-public class KeyedProcessOperator<KEY, IN, OUT> extends ProcessOperator<IN, OUT>
- implements Triggerable<KEY, VoidNamespace> {
- private transient InternalTimerService<VoidNamespace> timerService;
+/** Base operator for {@link OneInputStreamProcessFunction} in {@link KeyedPartitionStream}. */
+public class BaseKeyedProcessOperator<KEY, IN, OUT> extends ProcessOperator<IN, OUT> {
// TODO Restore this keySet when task initialized from checkpoint.
- private transient Set<Object> keySet;
+ protected transient Set<Object> keySet;
- @Nullable private final KeySelector<OUT, KEY> outKeySelector;
+ @Nullable protected final KeySelector<OUT, KEY> outKeySelector;
- public KeyedProcessOperator(OneInputStreamProcessFunction<IN, OUT> userFunction) {
+ public BaseKeyedProcessOperator(OneInputStreamProcessFunction<IN, OUT> userFunction) {
this(userFunction, null);
}
- public KeyedProcessOperator(
+ public BaseKeyedProcessOperator(
OneInputStreamProcessFunction<IN, OUT> userFunction,
@Nullable KeySelector<OUT, KEY> outKeySelector) {
super(userFunction);
@@ -65,8 +53,6 @@ public class KeyedProcessOperator<KEY, IN, OUT> extends ProcessOperator<IN, OUT>
@Override
public void open() throws Exception {
- this.timerService =
- getInternalTimerService("processing timer", VoidNamespaceSerializer.INSTANCE, this);
this.keySet = new HashSet<>();
super.open();
}
@@ -84,25 +70,6 @@ public class KeyedProcessOperator<KEY, IN, OUT> extends ProcessOperator<IN, OUT>
return getCurrentKey();
}
- @Override
- public void onEventTime(InternalTimer<KEY, VoidNamespace> timer) throws Exception {
- if (userFunction instanceof EventTimeWrappedOneInputStreamProcessFunction) {
- ((EventTimeWrappedOneInputStreamProcessFunction<IN, OUT>) userFunction)
- .onEventTime(timer.getTimestamp(), getOutputCollector(), partitionedContext);
- }
- }
-
- @Override
- public void onProcessingTime(InternalTimer<KEY, VoidNamespace> timer) throws Exception {
- userFunction.onProcessingTimer(
- timer.getTimestamp(), getOutputCollector(), partitionedContext);
- }
-
- @Override
- protected ProcessingTimeManager getProcessingTimeManager() {
- return new DefaultProcessingTimeManager(timerService);
- }
-
@Override
protected NonPartitionedContext<OUT> getNonPartitionedContext() {
return new DefaultNonPartitionedContext<>(
@@ -126,14 +93,4 @@ public class KeyedProcessOperator<KEY, IN, OUT> extends ProcessOperator<IN, OUT>
public boolean isAsyncStateProcessingEnabled() {
return true;
}
-
- @Override
- protected InternalTimerService<VoidNamespace> getTimerService() {
- return timerService;
- }
-
- @Override
- protected Supplier<Long> getEventTimeSupplier() {
- return () -> timerService.currentWatermark();
- }
}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedTwoInputNonBroadcastProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/BaseKeyedTwoInputNonBroadcastProcessOperator.java
similarity index 59%
copy from flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedTwoInputNonBroadcastProcessOperator.java
copy to flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/BaseKeyedTwoInputNonBroadcastProcessOperator.java
index f6e0f8b71ec..0861189a2ab 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedTwoInputNonBroadcastProcessOperator.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/BaseKeyedTwoInputNonBroadcastProcessOperator.java
@@ -20,47 +20,37 @@ package org.apache.flink.datastream.impl.operators;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.datastream.api.context.NonPartitionedContext;
-import org.apache.flink.datastream.api.context.ProcessingTimeManager;
import org.apache.flink.datastream.api.function.TwoInputNonBroadcastStreamProcessFunction;
import org.apache.flink.datastream.api.stream.KeyedPartitionStream;
import org.apache.flink.datastream.impl.common.KeyCheckedOutputCollector;
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.DefaultProcessingTimeManager;
-import org.apache.flink.datastream.impl.extension.eventtime.functions.EventTimeWrappedTwoInputNonBroadcastStreamProcessFunction;
-import org.apache.flink.runtime.state.VoidNamespace;
-import org.apache.flink.runtime.state.VoidNamespaceSerializer;
-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.runtime.streamrecord.StreamRecord;
import javax.annotation.Nullable;
import java.util.HashSet;
import java.util.Set;
-import java.util.function.Supplier;
/**
- * Operator for {@link TwoInputNonBroadcastStreamProcessFunction} in {@link KeyedPartitionStream}.
+ * Base operator for {@link TwoInputNonBroadcastStreamProcessFunction} in {@link
+ * KeyedPartitionStream}.
*/
-public class KeyedTwoInputNonBroadcastProcessOperator<KEY, IN1, IN2, OUT>
- extends TwoInputNonBroadcastProcessOperator<IN1, IN2, OUT>
- implements Triggerable<KEY, VoidNamespace> {
- private transient InternalTimerService<VoidNamespace> timerService;
+public class BaseKeyedTwoInputNonBroadcastProcessOperator<KEY, IN1, IN2, OUT>
+ extends TwoInputNonBroadcastProcessOperator<IN1, IN2, OUT> {
// TODO Restore this keySet when task initialized from checkpoint.
- private transient Set<Object> keySet;
+ protected transient Set<Object> keySet;
- @Nullable private final KeySelector<OUT, KEY> outKeySelector;
+ @Nullable protected final KeySelector<OUT, KEY> outKeySelector;
- public KeyedTwoInputNonBroadcastProcessOperator(
+ public BaseKeyedTwoInputNonBroadcastProcessOperator(
TwoInputNonBroadcastStreamProcessFunction<IN1, IN2, OUT> userFunction) {
this(userFunction, null);
}
- public KeyedTwoInputNonBroadcastProcessOperator(
+ public BaseKeyedTwoInputNonBroadcastProcessOperator(
TwoInputNonBroadcastStreamProcessFunction<IN1, IN2, OUT> userFunction,
@Nullable KeySelector<OUT, KEY> outKeySelector) {
super(userFunction);
@@ -69,8 +59,6 @@ public class KeyedTwoInputNonBroadcastProcessOperator<KEY, IN1, IN2, OUT>
@Override
public void open() throws Exception {
- this.timerService =
- getInternalTimerService("processing timer", VoidNamespaceSerializer.INSTANCE, this);
this.keySet = new HashSet<>();
super.open();
}
@@ -88,25 +76,6 @@ public class KeyedTwoInputNonBroadcastProcessOperator<KEY, IN1, IN2, OUT>
return getCurrentKey();
}
- protected ProcessingTimeManager getProcessingTimeManager() {
- return new DefaultProcessingTimeManager(timerService);
- }
-
- @Override
- public void onEventTime(InternalTimer<KEY, VoidNamespace> timer) throws Exception {
- if (userFunction instanceof EventTimeWrappedTwoInputNonBroadcastStreamProcessFunction) {
- ((EventTimeWrappedTwoInputNonBroadcastStreamProcessFunction<IN1, IN2, OUT>)
- userFunction)
- .onEventTime(timer.getTimestamp(), getOutputCollector(), partitionedContext);
- }
- }
-
- @Override
- public void onProcessingTime(InternalTimer<KEY, VoidNamespace> timer) throws Exception {
- userFunction.onProcessingTimer(
- timer.getTimestamp(), getOutputCollector(), partitionedContext);
- }
-
@Override
protected NonPartitionedContext<OUT> getNonPartitionedContext() {
return new DefaultNonPartitionedContext<>(
@@ -137,14 +106,4 @@ public class KeyedTwoInputNonBroadcastProcessOperator<KEY, IN1, IN2, OUT>
public boolean isAsyncStateProcessingEnabled() {
return true;
}
-
- @Override
- protected InternalTimerService<VoidNamespace> getTimerService() {
- return timerService;
- }
-
- @Override
- protected Supplier<Long> getEventTimeSupplier() {
- return () -> timerService.currentWatermark();
- }
}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedTwoOutputProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/BaseKeyedTwoOutputProcessOperator.java
similarity index 62%
copy from flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedTwoOutputProcessOperator.java
copy to flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/BaseKeyedTwoOutputProcessOperator.java
index db9dfc2cb42..cc3ba7b6363 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedTwoOutputProcessOperator.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/BaseKeyedTwoOutputProcessOperator.java
@@ -19,20 +19,13 @@
package org.apache.flink.datastream.impl.operators;
import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.datastream.api.context.ProcessingTimeManager;
import org.apache.flink.datastream.api.context.TwoOutputNonPartitionedContext;
import org.apache.flink.datastream.api.function.TwoOutputStreamProcessFunction;
+import org.apache.flink.datastream.api.stream.KeyedPartitionStream;
import org.apache.flink.datastream.impl.common.KeyCheckedOutputCollector;
import org.apache.flink.datastream.impl.common.OutputCollector;
import org.apache.flink.datastream.impl.common.TimestampCollector;
-import org.apache.flink.datastream.impl.context.DefaultProcessingTimeManager;
import org.apache.flink.datastream.impl.context.DefaultTwoOutputNonPartitionedContext;
-import org.apache.flink.datastream.impl.extension.eventtime.functions.EventTimeWrappedTwoOutputStreamProcessFunction;
-import org.apache.flink.runtime.state.VoidNamespace;
-import org.apache.flink.runtime.state.VoidNamespaceSerializer;
-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.runtime.streamrecord.StreamRecord;
import org.apache.flink.util.OutputTag;
import org.apache.flink.util.Preconditions;
@@ -41,28 +34,19 @@ import javax.annotation.Nullable;
import java.util.HashSet;
import java.util.Set;
-import java.util.function.Supplier;
-/** */
-public class KeyedTwoOutputProcessOperator<KEY, IN, OUT_MAIN, OUT_SIDE>
- extends TwoOutputProcessOperator<IN, OUT_MAIN, OUT_SIDE>
- implements Triggerable<KEY, VoidNamespace> {
- private transient InternalTimerService<VoidNamespace> timerService;
+/** Base operator for {@link TwoOutputStreamProcessFunction} in {@link KeyedPartitionStream}. */
+public class BaseKeyedTwoOutputProcessOperator<KEY, IN, OUT_MAIN, OUT_SIDE>
+ extends TwoOutputProcessOperator<IN, OUT_MAIN, OUT_SIDE> {
// TODO Restore this keySet when task initialized from checkpoint.
- private transient Set<Object> keySet;
+ protected transient Set<Object> keySet;
- @Nullable private final KeySelector<OUT_MAIN, KEY> mainOutKeySelector;
+ @Nullable protected final KeySelector<OUT_MAIN, KEY> mainOutKeySelector;
- @Nullable private final KeySelector<OUT_SIDE, KEY> sideOutKeySelector;
+ @Nullable protected final KeySelector<OUT_SIDE, KEY> sideOutKeySelector;
- public KeyedTwoOutputProcessOperator(
- TwoOutputStreamProcessFunction<IN, OUT_MAIN, OUT_SIDE> userFunction,
- OutputTag<OUT_SIDE> outputTag) {
- this(userFunction, outputTag, null, null);
- }
-
- public KeyedTwoOutputProcessOperator(
+ public BaseKeyedTwoOutputProcessOperator(
TwoOutputStreamProcessFunction<IN, OUT_MAIN, OUT_SIDE> userFunction,
OutputTag<OUT_SIDE> outputTag,
@Nullable KeySelector<OUT_MAIN, KEY> mainOutKeySelector,
@@ -78,8 +62,6 @@ public class KeyedTwoOutputProcessOperator<KEY, IN, OUT_MAIN, OUT_SIDE>
@Override
public void open() throws Exception {
- this.timerService =
- getInternalTimerService("processing timer", VoidNamespaceSerializer.INSTANCE, this);
this.keySet = new HashSet<>();
super.open();
}
@@ -109,28 +91,6 @@ public class KeyedTwoOutputProcessOperator<KEY, IN, OUT_MAIN, OUT_SIDE>
return getCurrentKey();
}
- protected ProcessingTimeManager getProcessingTimeManager() {
- return new DefaultProcessingTimeManager(timerService);
- }
-
- @Override
- public void onEventTime(InternalTimer<KEY, VoidNamespace> timer) throws Exception {
- if (userFunction instanceof EventTimeWrappedTwoOutputStreamProcessFunction) {
- ((EventTimeWrappedTwoOutputStreamProcessFunction<IN, OUT_MAIN, OUT_SIDE>) userFunction)
- .onEventTime(
- timer.getTimestamp(),
- getMainCollector(),
- getSideCollector(),
- partitionedContext);
- }
- }
-
- @Override
- public void onProcessingTime(InternalTimer<KEY, VoidNamespace> timer) throws Exception {
- userFunction.onProcessingTimer(
- timer.getTimestamp(), getMainCollector(), getSideCollector(), partitionedContext);
- }
-
@Override
protected TwoOutputNonPartitionedContext<OUT_MAIN, OUT_SIDE> getNonPartitionedContext() {
return new DefaultTwoOutputNonPartitionedContext<>(
@@ -155,14 +115,4 @@ public class KeyedTwoOutputProcessOperator<KEY, IN, OUT_MAIN, OUT_SIDE>
public boolean isAsyncStateProcessingEnabled() {
return true;
}
-
- @Override
- protected InternalTimerService<VoidNamespace> getTimerService() {
- return timerService;
- }
-
- @Override
- protected Supplier<Long> getEventTimeSupplier() {
- return () -> timerService.currentWatermark();
- }
}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedProcessOperator.java
index 23bae4b9921..e10a908ede8 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedProcessOperator.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedProcessOperator.java
@@ -19,14 +19,9 @@
package org.apache.flink.datastream.impl.operators;
import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.datastream.api.context.NonPartitionedContext;
import org.apache.flink.datastream.api.context.ProcessingTimeManager;
import org.apache.flink.datastream.api.function.OneInputStreamProcessFunction;
import org.apache.flink.datastream.api.stream.KeyedPartitionStream;
-import org.apache.flink.datastream.impl.common.KeyCheckedOutputCollector;
-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.DefaultProcessingTimeManager;
import org.apache.flink.datastream.impl.extension.eventtime.functions.EventTimeWrappedOneInputStreamProcessFunction;
import org.apache.flink.runtime.state.VoidNamespace;
@@ -34,24 +29,16 @@ import org.apache.flink.runtime.state.VoidNamespaceSerializer;
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.runtime.streamrecord.StreamRecord;
import javax.annotation.Nullable;
-import java.util.HashSet;
-import java.util.Set;
import java.util.function.Supplier;
/** Operator for {@link OneInputStreamProcessFunction} in {@link KeyedPartitionStream}. */
-public class KeyedProcessOperator<KEY, IN, OUT> extends ProcessOperator<IN, OUT>
+public class KeyedProcessOperator<KEY, IN, OUT> extends BaseKeyedProcessOperator<KEY, IN, OUT>
implements Triggerable<KEY, VoidNamespace> {
private transient InternalTimerService<VoidNamespace> timerService;
- // TODO Restore this keySet when task initialized from checkpoint.
- private transient Set<Object> keySet;
-
- @Nullable private final KeySelector<OUT, KEY> outKeySelector;
-
public KeyedProcessOperator(OneInputStreamProcessFunction<IN, OUT> userFunction) {
this(userFunction, null);
}
@@ -59,31 +46,16 @@ public class KeyedProcessOperator<KEY, IN, OUT> extends ProcessOperator<IN, OUT>
public KeyedProcessOperator(
OneInputStreamProcessFunction<IN, OUT> userFunction,
@Nullable KeySelector<OUT, KEY> outKeySelector) {
- super(userFunction);
- this.outKeySelector = outKeySelector;
+ super(userFunction, outKeySelector);
}
@Override
public void open() throws Exception {
this.timerService =
getInternalTimerService("processing timer", VoidNamespaceSerializer.INSTANCE, this);
- this.keySet = new HashSet<>();
super.open();
}
- @Override
- protected TimestampCollector<OUT> getOutputCollector() {
- return outKeySelector != null
- ? new KeyCheckedOutputCollector<>(
- new OutputCollector<>(output), outKeySelector, () -> (KEY) getCurrentKey())
- : new OutputCollector<>(output);
- }
-
- @Override
- protected Object currentKey() {
- return getCurrentKey();
- }
-
@Override
public void onEventTime(InternalTimer<KEY, VoidNamespace> timer) throws Exception {
if (userFunction instanceof EventTimeWrappedOneInputStreamProcessFunction) {
@@ -103,30 +75,6 @@ public class KeyedProcessOperator<KEY, IN, OUT> extends ProcessOperator<IN, OUT>
return new DefaultProcessingTimeManager(timerService);
}
- @Override
- protected NonPartitionedContext<OUT> getNonPartitionedContext() {
- return new DefaultNonPartitionedContext<>(
- context,
- partitionedContext,
- outputCollector,
- true,
- keySet,
- output,
- watermarkDeclarationMap);
- }
-
- @Override
- @SuppressWarnings({"rawtypes"})
- public void setKeyContextElement1(StreamRecord record) throws Exception {
- super.setKeyContextElement1(record);
- keySet.add(getCurrentKey());
- }
-
- @Override
- public boolean isAsyncStateProcessingEnabled() {
- return true;
- }
-
@Override
protected InternalTimerService<VoidNamespace> getTimerService() {
return timerService;
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedTwoInputNonBroadcastProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedTwoInputNonBroadcastProcessOperator.java
index f6e0f8b71ec..ba301d8639e 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedTwoInputNonBroadcastProcessOperator.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedTwoInputNonBroadcastProcessOperator.java
@@ -19,14 +19,9 @@
package org.apache.flink.datastream.impl.operators;
import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.datastream.api.context.NonPartitionedContext;
import org.apache.flink.datastream.api.context.ProcessingTimeManager;
import org.apache.flink.datastream.api.function.TwoInputNonBroadcastStreamProcessFunction;
import org.apache.flink.datastream.api.stream.KeyedPartitionStream;
-import org.apache.flink.datastream.impl.common.KeyCheckedOutputCollector;
-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.DefaultProcessingTimeManager;
import org.apache.flink.datastream.impl.extension.eventtime.functions.EventTimeWrappedTwoInputNonBroadcastStreamProcessFunction;
import org.apache.flink.runtime.state.VoidNamespace;
@@ -34,27 +29,19 @@ import org.apache.flink.runtime.state.VoidNamespaceSerializer;
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.runtime.streamrecord.StreamRecord;
import javax.annotation.Nullable;
-import java.util.HashSet;
-import java.util.Set;
import java.util.function.Supplier;
/**
* Operator for {@link TwoInputNonBroadcastStreamProcessFunction} in {@link KeyedPartitionStream}.
*/
public class KeyedTwoInputNonBroadcastProcessOperator<KEY, IN1, IN2, OUT>
- extends TwoInputNonBroadcastProcessOperator<IN1, IN2, OUT>
+ extends BaseKeyedTwoInputNonBroadcastProcessOperator<KEY, IN1, IN2, OUT>
implements Triggerable<KEY, VoidNamespace> {
private transient InternalTimerService<VoidNamespace> timerService;
- // TODO Restore this keySet when task initialized from checkpoint.
- private transient Set<Object> keySet;
-
- @Nullable private final KeySelector<OUT, KEY> outKeySelector;
-
public KeyedTwoInputNonBroadcastProcessOperator(
TwoInputNonBroadcastStreamProcessFunction<IN1, IN2, OUT> userFunction) {
this(userFunction, null);
@@ -63,31 +50,16 @@ public class KeyedTwoInputNonBroadcastProcessOperator<KEY, IN1, IN2, OUT>
public KeyedTwoInputNonBroadcastProcessOperator(
TwoInputNonBroadcastStreamProcessFunction<IN1, IN2, OUT> userFunction,
@Nullable KeySelector<OUT, KEY> outKeySelector) {
- super(userFunction);
- this.outKeySelector = outKeySelector;
+ super(userFunction, outKeySelector);
}
@Override
public void open() throws Exception {
this.timerService =
getInternalTimerService("processing timer", VoidNamespaceSerializer.INSTANCE, this);
- this.keySet = new HashSet<>();
super.open();
}
- @Override
- protected TimestampCollector<OUT> getOutputCollector() {
- return outKeySelector == null
- ? new OutputCollector<>(output)
- : new KeyCheckedOutputCollector<>(
- new OutputCollector<>(output), outKeySelector, () -> (KEY) getCurrentKey());
- }
-
- @Override
- protected Object currentKey() {
- return getCurrentKey();
- }
-
protected ProcessingTimeManager getProcessingTimeManager() {
return new DefaultProcessingTimeManager(timerService);
}
@@ -107,37 +79,6 @@ public class KeyedTwoInputNonBroadcastProcessOperator<KEY, IN1, IN2, OUT>
timer.getTimestamp(), getOutputCollector(), partitionedContext);
}
- @Override
- protected NonPartitionedContext<OUT> getNonPartitionedContext() {
- return new DefaultNonPartitionedContext<>(
- context,
- partitionedContext,
- collector,
- true,
- keySet,
- output,
- watermarkDeclarationMap);
- }
-
- @Override
- @SuppressWarnings({"rawtypes"})
- public void setKeyContextElement1(StreamRecord record) throws Exception {
- super.setKeyContextElement1(record);
- keySet.add(getCurrentKey());
- }
-
- @Override
- @SuppressWarnings({"rawtypes"})
- public void setKeyContextElement2(StreamRecord record) throws Exception {
- super.setKeyContextElement2(record);
- keySet.add(getCurrentKey());
- }
-
- @Override
- public boolean isAsyncStateProcessingEnabled() {
- return true;
- }
-
@Override
protected InternalTimerService<VoidNamespace> getTimerService() {
return timerService;
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedTwoOutputProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedTwoOutputProcessOperator.java
index db9dfc2cb42..473c175b3bb 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedTwoOutputProcessOperator.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/KeyedTwoOutputProcessOperator.java
@@ -20,42 +20,27 @@ package org.apache.flink.datastream.impl.operators;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.datastream.api.context.ProcessingTimeManager;
-import org.apache.flink.datastream.api.context.TwoOutputNonPartitionedContext;
import org.apache.flink.datastream.api.function.TwoOutputStreamProcessFunction;
-import org.apache.flink.datastream.impl.common.KeyCheckedOutputCollector;
-import org.apache.flink.datastream.impl.common.OutputCollector;
-import org.apache.flink.datastream.impl.common.TimestampCollector;
import org.apache.flink.datastream.impl.context.DefaultProcessingTimeManager;
-import org.apache.flink.datastream.impl.context.DefaultTwoOutputNonPartitionedContext;
import org.apache.flink.datastream.impl.extension.eventtime.functions.EventTimeWrappedTwoOutputStreamProcessFunction;
import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.runtime.state.VoidNamespaceSerializer;
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.runtime.streamrecord.StreamRecord;
import org.apache.flink.util.OutputTag;
import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
-import java.util.HashSet;
-import java.util.Set;
import java.util.function.Supplier;
/** */
public class KeyedTwoOutputProcessOperator<KEY, IN, OUT_MAIN, OUT_SIDE>
- extends TwoOutputProcessOperator<IN, OUT_MAIN, OUT_SIDE>
+ extends BaseKeyedTwoOutputProcessOperator<KEY, IN, OUT_MAIN, OUT_SIDE>
implements Triggerable<KEY, VoidNamespace> {
private transient InternalTimerService<VoidNamespace> timerService;
- // TODO Restore this keySet when task initialized from checkpoint.
- private transient Set<Object> keySet;
-
- @Nullable private final KeySelector<OUT_MAIN, KEY> mainOutKeySelector;
-
- @Nullable private final KeySelector<OUT_SIDE, KEY> sideOutKeySelector;
-
public KeyedTwoOutputProcessOperator(
TwoOutputStreamProcessFunction<IN, OUT_MAIN, OUT_SIDE> userFunction,
OutputTag<OUT_SIDE> outputTag) {
@@ -67,48 +52,20 @@ public class KeyedTwoOutputProcessOperator<KEY, IN, OUT_MAIN, OUT_SIDE>
OutputTag<OUT_SIDE> outputTag,
@Nullable KeySelector<OUT_MAIN, KEY> mainOutKeySelector,
@Nullable KeySelector<OUT_SIDE, KEY> sideOutKeySelector) {
- super(userFunction, outputTag);
+ super(userFunction, outputTag, mainOutKeySelector, sideOutKeySelector);
Preconditions.checkArgument(
(mainOutKeySelector == null && sideOutKeySelector == null)
|| (mainOutKeySelector != null && sideOutKeySelector != null),
"Both mainOutKeySelector and sideOutKeySelector must be null or not null.");
- this.mainOutKeySelector = mainOutKeySelector;
- this.sideOutKeySelector = sideOutKeySelector;
}
@Override
public void open() throws Exception {
this.timerService =
getInternalTimerService("processing timer", VoidNamespaceSerializer.INSTANCE, this);
- this.keySet = new HashSet<>();
super.open();
}
- @Override
- protected TimestampCollector<OUT_MAIN> getMainCollector() {
- return mainOutKeySelector != null && sideOutKeySelector != null
- ? new KeyCheckedOutputCollector<>(
- new OutputCollector<>(output),
- mainOutKeySelector,
- () -> (KEY) getCurrentKey())
- : new OutputCollector<>(output);
- }
-
- @Override
- public TimestampCollector<OUT_SIDE> getSideCollector() {
- return mainOutKeySelector != null && sideOutKeySelector != null
- ? new KeyCheckedOutputCollector<>(
- new SideOutputCollector(output),
- sideOutKeySelector,
- () -> (KEY) getCurrentKey())
- : new SideOutputCollector(output);
- }
-
- @Override
- protected Object currentKey() {
- return getCurrentKey();
- }
-
protected ProcessingTimeManager getProcessingTimeManager() {
return new DefaultProcessingTimeManager(timerService);
}
@@ -131,31 +88,6 @@ public class KeyedTwoOutputProcessOperator<KEY, IN, OUT_MAIN, OUT_SIDE>
timer.getTimestamp(), getMainCollector(), getSideCollector(), partitionedContext);
}
- @Override
- protected TwoOutputNonPartitionedContext<OUT_MAIN, OUT_SIDE> getNonPartitionedContext() {
- return new DefaultTwoOutputNonPartitionedContext<>(
- context,
- partitionedContext,
- mainCollector,
- sideCollector,
- true,
- keySet,
- output,
- watermarkDeclarationMap);
- }
-
- @Override
- @SuppressWarnings({"rawtypes"})
- public void setKeyContextElement1(StreamRecord record) throws Exception {
- super.setKeyContextElement1(record);
- keySet.add(getCurrentKey());
- }
-
- @Override
- public boolean isAsyncStateProcessingEnabled() {
- return true;
- }
-
@Override
protected InternalTimerService<VoidNamespace> getTimerService() {
return timerService;
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/stream/GlobalStreamImpl.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/stream/GlobalStreamImpl.java
index 137917bca4f..9e3888df7bb 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/stream/GlobalStreamImpl.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/stream/GlobalStreamImpl.java
@@ -21,9 +21,11 @@ package org.apache.flink.datastream.impl.stream;
import org.apache.flink.api.common.attribute.Attribute;
import org.apache.flink.api.common.state.StateDeclaration;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.dsv2.Sink;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.functions.NullByteKeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.datastream.api.function.OneInputStreamProcessFunction;
import org.apache.flink.datastream.api.function.TwoInputNonBroadcastStreamProcessFunction;
@@ -35,6 +37,9 @@ import org.apache.flink.datastream.api.stream.NonKeyedPartitionStream;
import org.apache.flink.datastream.api.stream.ProcessConfigurable;
import org.apache.flink.datastream.impl.ExecutionEnvironmentImpl;
import org.apache.flink.datastream.impl.attribute.AttributeParser;
+import org.apache.flink.datastream.impl.extension.window.function.InternalOneInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.function.InternalTwoInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.function.InternalTwoOutputWindowStreamProcessFunction;
import org.apache.flink.datastream.impl.operators.ProcessOperator;
import org.apache.flink.datastream.impl.operators.TwoInputNonBroadcastProcessOperator;
import org.apache.flink.datastream.impl.operators.TwoOutputProcessOperator;
@@ -70,13 +75,30 @@ public class GlobalStreamImpl<T> extends AbstractDataStream<T> implements Global
TypeInformation<OUT> outType =
StreamUtils.getOutputTypeForOneInputProcessFunction(processFunction, getType());
- ProcessOperator<T, OUT> operator = new ProcessOperator<>(processFunction);
- return StreamUtils.wrapWithConfigureHandle(
- transform(
- "Global Process",
- outType,
- operator,
- AttributeParser.parseAttribute(processFunction)));
+
+ if (processFunction instanceof InternalOneInputWindowStreamProcessFunction) {
+ Transformation<OUT> outTransformation =
+ StreamUtils.transformOneInputWindow(
+ environment.getExecutionConfig(),
+ this,
+ getType(),
+ outType,
+ (InternalOneInputWindowStreamProcessFunction<T, OUT, ?>)
+ processFunction,
+ new NullByteKeySelector<>(),
+ Types.BYTE);
+ outTransformation.setParallelism(1, true);
+ return StreamUtils.wrapWithConfigureHandle(
+ new GlobalStreamImpl<>(this.environment, outTransformation));
+ } else {
+ ProcessOperator<T, OUT> operator = new ProcessOperator<>(processFunction);
+ return StreamUtils.wrapWithConfigureHandle(
+ transform(
+ "Global Process",
+ outType,
+ operator,
+ AttributeParser.parseAttribute(processFunction)));
+ }
}
@Override
@@ -91,16 +113,34 @@ public class GlobalStreamImpl<T> extends AbstractDataStream<T> implements Global
StreamUtils.getOutputTypesForTwoOutputProcessFunction(processFunction, getType());
TypeInformation<OUT1> firstOutputType = twoOutputType.f0;
TypeInformation<OUT2> secondOutputType = twoOutputType.f1;
- OutputTag<OUT2> secondOutputTag = new OutputTag<OUT2>("Second-Output", secondOutputType);
-
- TwoOutputProcessOperator<T, OUT1, OUT2> operator =
- new TwoOutputProcessOperator<>(processFunction, secondOutputTag);
- GlobalStreamImpl<OUT1> firstStream =
- transform(
- "Two-Output-Operator",
- firstOutputType,
- operator,
- AttributeParser.parseAttribute(processFunction));
+ OutputTag<OUT2> secondOutputTag = new OutputTag<>("Second-Output", secondOutputType);
+
+ GlobalStreamImpl<OUT1> firstStream;
+ if (processFunction instanceof InternalTwoOutputWindowStreamProcessFunction) {
+ Transformation<OUT1> outTransformation =
+ StreamUtils.transformTwoOutputWindow(
+ environment.getExecutionConfig(),
+ this,
+ getType(),
+ firstOutputType,
+ secondOutputType,
+ secondOutputTag,
+ (InternalTwoOutputWindowStreamProcessFunction<T, OUT1, OUT2, ?>)
+ processFunction,
+ new NullByteKeySelector<>(),
+ Types.BYTE);
+ outTransformation.setParallelism(1, true);
+ firstStream = new GlobalStreamImpl<>(environment, outTransformation);
+ } else {
+ TwoOutputProcessOperator<T, OUT1, OUT2> operator =
+ new TwoOutputProcessOperator<>(processFunction, secondOutputTag);
+ firstStream =
+ transform(
+ "Two-Output-Operator",
+ firstOutputType,
+ operator,
+ AttributeParser.parseAttribute(processFunction));
+ }
GlobalStreamImpl<OUT2> secondStream =
new GlobalStreamImpl<>(
environment, firstStream.getSideOutputTransform(secondOutputTag));
@@ -121,15 +161,34 @@ public class GlobalStreamImpl<T> extends AbstractDataStream<T> implements Global
TypeInformation<OUT> outTypeInfo =
StreamUtils.getOutputTypeForTwoInputNonBroadcastProcessFunction(
processFunction, getType(), ((GlobalStreamImpl<T_OTHER>) other).getType());
- TwoInputNonBroadcastProcessOperator<T, T_OTHER, OUT> processOperator =
- new TwoInputNonBroadcastProcessOperator<>(processFunction);
- Transformation<OUT> outTransformation =
- StreamUtils.getTwoInputTransformation(
- "Global-Global-TwoInput-Process",
- this,
- (GlobalStreamImpl<T_OTHER>) other,
- outTypeInfo,
- processOperator);
+
+ Transformation<OUT> outTransformation;
+ if (processFunction instanceof InternalTwoInputWindowStreamProcessFunction) {
+ outTransformation =
+ StreamUtils.transformTwoInputNonBroadcastWindow(
+ environment.getExecutionConfig(),
+ getTransformation(),
+ getType(),
+ ((GlobalStreamImpl<T_OTHER>) other).getTransformation(),
+ ((GlobalStreamImpl<T_OTHER>) other).getType(),
+ outTypeInfo,
+ (InternalTwoInputWindowStreamProcessFunction<T, T_OTHER, OUT, ?>)
+ processFunction,
+ new NullByteKeySelector<>(),
+ Types.BYTE,
+ new NullByteKeySelector<>(),
+ Types.BYTE);
+ } else {
+ TwoInputNonBroadcastProcessOperator<T, T_OTHER, OUT> processOperator =
+ new TwoInputNonBroadcastProcessOperator<>(processFunction);
+ outTransformation =
+ StreamUtils.getTwoInputTransformation(
+ "Global-Global-TwoInput-Process",
+ this,
+ (GlobalStreamImpl<T_OTHER>) other,
+ outTypeInfo,
+ processOperator);
+ }
// Operator parallelism should always be 1 for global stream.
// parallelismConfigured should be true to avoid overwritten by AdaptiveBatchScheduler.
outTransformation.setParallelism(1, true);
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/stream/KeyedPartitionStreamImpl.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/stream/KeyedPartitionStreamImpl.java
index 4bbabb9653f..8036297e14f 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/stream/KeyedPartitionStreamImpl.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/stream/KeyedPartitionStreamImpl.java
@@ -19,6 +19,7 @@
package org.apache.flink.datastream.impl.stream;
import org.apache.flink.api.common.state.StateDeclaration;
+import org.apache.flink.api.common.state.v2.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.connector.dsv2.Sink;
import org.apache.flink.api.dag.Transformation;
@@ -39,12 +40,14 @@ import org.apache.flink.datastream.api.stream.ProcessConfigurable;
import org.apache.flink.datastream.impl.attribute.AttributeParser;
import org.apache.flink.datastream.impl.extension.join.operators.TwoInputNonBroadcastJoinProcessFunction;
import org.apache.flink.datastream.impl.extension.join.operators.TwoInputNonBroadcastJoinProcessOperator;
+import org.apache.flink.datastream.impl.extension.window.function.InternalOneInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.function.InternalTwoInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.function.InternalTwoOutputWindowStreamProcessFunction;
import org.apache.flink.datastream.impl.operators.KeyedProcessOperator;
import org.apache.flink.datastream.impl.operators.KeyedTwoInputBroadcastProcessOperator;
import org.apache.flink.datastream.impl.operators.KeyedTwoInputNonBroadcastProcessOperator;
import org.apache.flink.datastream.impl.operators.KeyedTwoOutputProcessOperator;
import org.apache.flink.datastream.impl.utils.StreamUtils;
-import org.apache.flink.runtime.state.v2.ListStateDescriptor;
import org.apache.flink.streaming.api.graph.StreamGraphGenerator;
import org.apache.flink.streaming.api.transformations.DataStreamV2SinkTransformation;
import org.apache.flink.streaming.api.transformations.PartitionTransformation;
@@ -118,10 +121,24 @@ public class KeyedPartitionStreamImpl<K, V> extends AbstractDataStream<V>
TypeInformation<OUT> outType;
outType = StreamUtils.getOutputTypeForOneInputProcessFunction(processFunction, getType());
- KeyedProcessOperator<K, V, OUT> operator = new KeyedProcessOperator<>(processFunction);
- Transformation<OUT> transform =
- StreamUtils.getOneInputKeyedTransformation(
- "KeyedProcess", this, outType, operator, keySelector, keyType);
+ Transformation<OUT> transform;
+ if (processFunction instanceof InternalOneInputWindowStreamProcessFunction) {
+ transform =
+ StreamUtils.transformOneInputWindow(
+ environment.getExecutionConfig(),
+ this,
+ getType(),
+ outType,
+ (InternalOneInputWindowStreamProcessFunction<V, OUT, ?>)
+ processFunction,
+ keySelector,
+ keyType);
+ } else {
+ KeyedProcessOperator<K, V, OUT> operator = new KeyedProcessOperator<>(processFunction);
+ transform =
+ StreamUtils.getOneInputKeyedTransformation(
+ "KeyedProcess", this, outType, operator, keySelector, keyType);
+ }
transform.setAttribute(AttributeParser.parseAttribute(processFunction));
environment.addOperator(transform);
return StreamUtils.wrapWithConfigureHandle(
@@ -139,11 +156,26 @@ public class KeyedPartitionStreamImpl<K, V> extends AbstractDataStream<V>
TypeInformation<OUT> outType =
StreamUtils.getOutputTypeForOneInputProcessFunction(processFunction, getType());
- KeyedProcessOperator<K, V, OUT> operator =
- new KeyedProcessOperator<>(processFunction, checkNotNull(newKeySelector));
- Transformation<OUT> transform =
- StreamUtils.getOneInputKeyedTransformation(
- "KeyedProcess", this, outType, operator, keySelector, keyType);
+
+ Transformation<OUT> transform;
+ if (processFunction instanceof InternalOneInputWindowStreamProcessFunction) {
+ transform =
+ StreamUtils.transformOneInputWindow(
+ environment.getExecutionConfig(),
+ this,
+ getType(),
+ outType,
+ (InternalOneInputWindowStreamProcessFunction<V, OUT, ?>)
+ processFunction,
+ keySelector,
+ keyType);
+ } else {
+ KeyedProcessOperator<K, V, OUT> operator =
+ new KeyedProcessOperator<>(processFunction, checkNotNull(newKeySelector));
+ transform =
+ StreamUtils.getOneInputKeyedTransformation(
+ "KeyedProcess", this, outType, operator, keySelector, keyType);
+ }
NonKeyedPartitionStreamImpl<OUT> outputStream =
new NonKeyedPartitionStreamImpl<>(environment, transform);
transform.setAttribute(AttributeParser.parseAttribute(processFunction));
@@ -173,17 +205,33 @@ public class KeyedPartitionStreamImpl<K, V> extends AbstractDataStream<V>
TypeInformation<OUT2> secondOutputType = twoOutputType.f1;
OutputTag<OUT2> secondOutputTag = new OutputTag<>("Second-Output", secondOutputType);
- KeyedTwoOutputProcessOperator<K, V, OUT1, OUT2> operator =
- new KeyedTwoOutputProcessOperator<>(
- processFunction, secondOutputTag, keySelector1, keySelector2);
- Transformation<OUT1> mainOutputTransform =
- StreamUtils.getOneInputKeyedTransformation(
- "Two-Output-Process",
- this,
- firstOutputType,
- operator,
- keySelector,
- keyType);
+ Transformation<OUT1> mainOutputTransform;
+ if (processFunction instanceof InternalTwoOutputWindowStreamProcessFunction) {
+ mainOutputTransform =
+ StreamUtils.transformTwoOutputWindow(
+ environment.getExecutionConfig(),
+ this,
+ getType(),
+ firstOutputType,
+ secondOutputType,
+ secondOutputTag,
+ (InternalTwoOutputWindowStreamProcessFunction<V, OUT1, OUT2, ?>)
+ processFunction,
+ keySelector,
+ keyType);
+ } else {
+ KeyedTwoOutputProcessOperator<K, V, OUT1, OUT2> operator =
+ new KeyedTwoOutputProcessOperator<>(
+ processFunction, secondOutputTag, keySelector1, keySelector2);
+ mainOutputTransform =
+ StreamUtils.getOneInputKeyedTransformation(
+ "Two-Output-Process",
+ this,
+ firstOutputType,
+ operator,
+ keySelector,
+ keyType);
+ }
mainOutputTransform.setAttribute(AttributeParser.parseAttribute(processFunction));
NonKeyedPartitionStreamImpl<OUT1> nonKeyedMainOutputStream =
new NonKeyedPartitionStreamImpl<>(environment, mainOutputTransform);
@@ -226,16 +274,32 @@ public class KeyedPartitionStreamImpl<K, V> extends AbstractDataStream<V>
TypeInformation<OUT2> secondOutputType = twoOutputType.f1;
OutputTag<OUT2> secondOutputTag = new OutputTag<>("Second-Output", secondOutputType);
- KeyedTwoOutputProcessOperator<K, V, OUT1, OUT2> operator =
- new KeyedTwoOutputProcessOperator<>(processFunction, secondOutputTag);
- Transformation<OUT1> firstTransformation =
- StreamUtils.getOneInputKeyedTransformation(
- "Two-Output-Process",
- this,
- firstOutputType,
- operator,
- keySelector,
- keyType);
+ Transformation<OUT1> firstTransformation;
+ if (processFunction instanceof InternalTwoOutputWindowStreamProcessFunction) {
+ firstTransformation =
+ StreamUtils.transformTwoOutputWindow(
+ environment.getExecutionConfig(),
+ this,
+ getType(),
+ firstOutputType,
+ secondOutputType,
+ secondOutputTag,
+ (InternalTwoOutputWindowStreamProcessFunction<V, OUT1, OUT2, ?>)
+ processFunction,
+ keySelector,
+ keyType);
+ } else {
+ KeyedTwoOutputProcessOperator<K, V, OUT1, OUT2> operator =
+ new KeyedTwoOutputProcessOperator<>(processFunction, secondOutputTag);
+ firstTransformation =
+ StreamUtils.getOneInputKeyedTransformation(
+ "Two-Output-Process",
+ this,
+ firstOutputType,
+ operator,
+ keySelector,
+ keyType);
+ }
firstTransformation.setAttribute(AttributeParser.parseAttribute(processFunction));
NonKeyedPartitionStreamImpl<OUT1> firstStream =
new NonKeyedPartitionStreamImpl<>(environment, firstTransformation);
@@ -270,6 +334,21 @@ public class KeyedPartitionStreamImpl<K, V> extends AbstractDataStream<V>
if (processFunction instanceof TwoInputNonBroadcastJoinProcessFunction) {
outTransformation = getJoinTransformation(other, processFunction, outTypeInfo);
+ } else if (processFunction instanceof InternalTwoInputWindowStreamProcessFunction) {
+ outTransformation =
+ StreamUtils.transformTwoInputNonBroadcastWindow(
+ environment.getExecutionConfig(),
+ this.transformation,
+ getType(),
+ ((KeyedPartitionStreamImpl<K, T_OTHER>) other).getTransformation(),
+ ((KeyedPartitionStreamImpl<K, T_OTHER>) other).getType(),
+ outTypeInfo,
+ (InternalTwoInputWindowStreamProcessFunction<V, T_OTHER, OUT, ?>)
+ processFunction,
+ keySelector,
+ keyType,
+ ((KeyedPartitionStreamImpl<K, T_OTHER>) other).getKeySelector(),
+ ((KeyedPartitionStreamImpl<K, T_OTHER>) other).getKeyType());
} else {
KeyedTwoInputNonBroadcastProcessOperator<K, V, T_OTHER, OUT> processOperator =
new KeyedTwoInputNonBroadcastProcessOperator<>(processFunction);
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/stream/NonKeyedPartitionStreamImpl.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/stream/NonKeyedPartitionStreamImpl.java
index ea747f8c898..be523dc0f58 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/stream/NonKeyedPartitionStreamImpl.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/stream/NonKeyedPartitionStreamImpl.java
@@ -20,10 +20,13 @@ package org.apache.flink.datastream.impl.stream;
import org.apache.flink.api.common.state.StateDeclaration;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.dsv2.Sink;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.functions.NullByteKeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.datastream.api.extension.window.strategy.GlobalWindowStrategy;
import org.apache.flink.datastream.api.function.OneInputStreamProcessFunction;
import org.apache.flink.datastream.api.function.TwoInputBroadcastStreamProcessFunction;
import org.apache.flink.datastream.api.function.TwoInputNonBroadcastStreamProcessFunction;
@@ -35,6 +38,9 @@ import org.apache.flink.datastream.api.stream.NonKeyedPartitionStream;
import org.apache.flink.datastream.api.stream.ProcessConfigurable;
import org.apache.flink.datastream.impl.ExecutionEnvironmentImpl;
import org.apache.flink.datastream.impl.attribute.AttributeParser;
+import org.apache.flink.datastream.impl.extension.window.function.InternalOneInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.function.InternalTwoInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.function.InternalTwoOutputWindowStreamProcessFunction;
import org.apache.flink.datastream.impl.operators.ProcessOperator;
import org.apache.flink.datastream.impl.operators.TwoInputBroadcastProcessOperator;
import org.apache.flink.datastream.impl.operators.TwoInputNonBroadcastProcessOperator;
@@ -52,6 +58,7 @@ import java.util.Collections;
import java.util.HashSet;
import static org.apache.flink.datastream.impl.utils.StreamUtils.validateStates;
+import static org.apache.flink.util.Preconditions.checkState;
/** The implementation of {@link NonKeyedPartitionStream}. */
public class NonKeyedPartitionStreamImpl<T> extends AbstractDataStream<T>
@@ -73,9 +80,31 @@ public class NonKeyedPartitionStreamImpl<T> extends AbstractDataStream<T>
TypeInformation<OUT> outType =
StreamUtils.getOutputTypeForOneInputProcessFunction(processFunction, getType());
- ProcessOperator<T, OUT> operator = new ProcessOperator<>(processFunction);
- OneInputTransformation<T, OUT> outputTransform =
- StreamUtils.getOneInputTransformation("Process", this, outType, operator);
+ OneInputTransformation<T, OUT> outputTransform;
+
+ if (processFunction instanceof InternalOneInputWindowStreamProcessFunction) {
+ InternalOneInputWindowStreamProcessFunction<T, OUT, ?> internalProcessFunction =
+ (InternalOneInputWindowStreamProcessFunction<T, OUT, ?>) processFunction;
+ checkState(
+ internalProcessFunction.getWindowStrategy() instanceof GlobalWindowStrategy,
+ "Only the Global Window is permitted for execution in NonKeyedPartitionStream.");
+ outputTransform =
+ (OneInputTransformation<T, OUT>)
+ StreamUtils.transformOneInputWindow(
+ environment.getExecutionConfig(),
+ this,
+ getType(),
+ outType,
+ (InternalOneInputWindowStreamProcessFunction<T, OUT, ?>)
+ processFunction,
+ new NullByteKeySelector<>(),
+ Types.BYTE);
+ } else {
+ ProcessOperator<T, OUT> operator = new ProcessOperator<>(processFunction);
+ outputTransform =
+ StreamUtils.getOneInputTransformation("Process", this, outType, operator);
+ }
+
outputTransform.setAttribute(AttributeParser.parseAttribute(processFunction));
environment.addOperator(outputTransform);
return StreamUtils.wrapWithConfigureHandle(
@@ -98,11 +127,35 @@ public class NonKeyedPartitionStreamImpl<T> extends AbstractDataStream<T>
TypeInformation<OUT2> secondOutputType = twoOutputType.f1;
OutputTag<OUT2> secondOutputTag = new OutputTag<>("Second-Output", secondOutputType);
- TwoOutputProcessOperator<T, OUT1, OUT2> operator =
- new TwoOutputProcessOperator<>(processFunction, secondOutputTag);
- OneInputTransformation<T, OUT1> outTransformation =
- StreamUtils.getOneInputTransformation(
- "Two-Output-Operator", this, firstOutputType, operator);
+ OneInputTransformation<T, OUT1> outTransformation;
+ if (processFunction instanceof InternalTwoOutputWindowStreamProcessFunction) {
+ InternalTwoOutputWindowStreamProcessFunction<T, OUT1, OUT2, ?> internalProcessFunction =
+ (InternalTwoOutputWindowStreamProcessFunction<T, OUT1, OUT2, ?>)
+ processFunction;
+ checkState(
+ internalProcessFunction.getWindowStrategy() instanceof GlobalWindowStrategy,
+ "Only the Global Window is permitted for execution in NonKeyedPartitionStream.");
+
+ outTransformation =
+ (OneInputTransformation<T, OUT1>)
+ StreamUtils.transformTwoOutputWindow(
+ environment.getExecutionConfig(),
+ this,
+ getType(),
+ firstOutputType,
+ secondOutputType,
+ secondOutputTag,
+ (InternalTwoOutputWindowStreamProcessFunction<T, OUT1, OUT2, ?>)
+ processFunction,
+ new NullByteKeySelector<>(),
+ Types.BYTE);
+ } else {
+ TwoOutputProcessOperator<T, OUT1, OUT2> operator =
+ new TwoOutputProcessOperator<>(processFunction, secondOutputTag);
+ outTransformation =
+ StreamUtils.getOneInputTransformation(
+ "Two-Output-Operator", this, firstOutputType, operator);
+ }
outTransformation.setAttribute(AttributeParser.parseAttribute(processFunction));
NonKeyedPartitionStreamImpl<OUT1> firstStream =
new NonKeyedPartitionStreamImpl<>(environment, outTransformation);
@@ -135,15 +188,39 @@ public class NonKeyedPartitionStreamImpl<T> extends AbstractDataStream<T>
getType(),
((NonKeyedPartitionStreamImpl<T_OTHER>) other).getType());
- TwoInputNonBroadcastProcessOperator<T, T_OTHER, OUT> processOperator =
- new TwoInputNonBroadcastProcessOperator<>(processFunction);
- Transformation<OUT> outTransformation =
- StreamUtils.getTwoInputTransformation(
- "TwoInput-Process",
- this,
- (NonKeyedPartitionStreamImpl<T_OTHER>) other,
- outTypeInfo,
- processOperator);
+ Transformation<OUT> outTransformation;
+ if (processFunction instanceof InternalTwoInputWindowStreamProcessFunction) {
+ InternalTwoInputWindowStreamProcessFunction<T, T_OTHER, OUT, ?> internalWindowFunction =
+ (InternalTwoInputWindowStreamProcessFunction<T, T_OTHER, OUT, ?>)
+ processFunction;
+ checkState(
+ internalWindowFunction.getWindowStrategy() instanceof GlobalWindowStrategy,
+ "Only the Global Window is permitted for execution in NonKeyedPartitionStream.");
+
+ outTransformation =
+ StreamUtils.transformTwoInputNonBroadcastWindow(
+ environment.getExecutionConfig(),
+ this.transformation,
+ getType(),
+ ((NonKeyedPartitionStreamImpl<T_OTHER>) other).getTransformation(),
+ ((NonKeyedPartitionStreamImpl<T_OTHER>) other).getType(),
+ outTypeInfo,
+ internalWindowFunction,
+ new NullByteKeySelector<>(),
+ Types.BYTE,
+ new NullByteKeySelector<>(),
+ Types.BYTE);
+ } else {
+ TwoInputNonBroadcastProcessOperator<T, T_OTHER, OUT> processOperator =
+ new TwoInputNonBroadcastProcessOperator<>(processFunction);
+ outTransformation =
+ StreamUtils.getTwoInputTransformation(
+ "TwoInput-Process",
+ this,
+ (NonKeyedPartitionStreamImpl<T_OTHER>) other,
+ outTypeInfo,
+ processOperator);
+ }
outTransformation.setAttribute(AttributeParser.parseAttribute(processFunction));
environment.addOperator(outTransformation);
return StreamUtils.wrapWithConfigureHandle(
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/utils/StreamUtils.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/utils/StreamUtils.java
index 3f73608dcf9..a072e3e361c 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/utils/StreamUtils.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/utils/StreamUtils.java
@@ -18,15 +18,21 @@
package org.apache.flink.datastream.impl.utils;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.state.IllegalRedistributionModeException;
import org.apache.flink.api.common.state.StateDeclaration;
+import org.apache.flink.api.common.state.v2.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.connector.dsv2.Sink;
import org.apache.flink.api.connector.dsv2.WrappedSink;
+import org.apache.flink.api.dag.Transformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.datastream.api.extension.join.JoinFunction;
+import org.apache.flink.datastream.api.extension.window.function.OneInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.api.extension.window.function.TwoInputNonBroadcastWindowStreamProcessFunction;
+import org.apache.flink.datastream.api.extension.window.function.TwoOutputWindowStreamProcessFunction;
import org.apache.flink.datastream.api.function.OneInputStreamProcessFunction;
import org.apache.flink.datastream.api.function.TwoInputBroadcastStreamProcessFunction;
import org.apache.flink.datastream.api.function.TwoInputNonBroadcastStreamProcessFunction;
@@ -39,6 +45,12 @@ import org.apache.flink.datastream.impl.extension.eventtime.functions.EventTimeW
import org.apache.flink.datastream.impl.extension.eventtime.functions.EventTimeWrappedTwoInputNonBroadcastStreamProcessFunction;
import org.apache.flink.datastream.impl.extension.eventtime.functions.EventTimeWrappedTwoOutputStreamProcessFunction;
import org.apache.flink.datastream.impl.extension.join.operators.TwoInputNonBroadcastJoinProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.function.InternalOneInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.function.InternalTwoInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.function.InternalTwoOutputWindowStreamProcessFunction;
+import org.apache.flink.datastream.impl.extension.window.operators.OneInputWindowProcessOperator;
+import org.apache.flink.datastream.impl.extension.window.operators.TwoInputNonBroadcastWindowProcessOperator;
+import org.apache.flink.datastream.impl.extension.window.operators.TwoOutputWindowProcessOperator;
import org.apache.flink.datastream.impl.stream.AbstractDataStream;
import org.apache.flink.datastream.impl.stream.GlobalStreamImpl;
import org.apache.flink.datastream.impl.stream.KeyedPartitionStreamImpl;
@@ -54,6 +66,10 @@ import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
import org.apache.flink.streaming.api.transformations.DataStreamV2SinkTransformation;
import org.apache.flink.streaming.api.transformations.OneInputTransformation;
import org.apache.flink.streaming.api.transformations.TwoInputTransformation;
+import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
+import org.apache.flink.streaming.api.windowing.windows.Window;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.TaggedUnion;
import org.apache.flink.util.Utils;
import java.util.Set;
@@ -70,6 +86,20 @@ public final class StreamUtils {
public static <IN, OUT> TypeInformation<OUT> getOutputTypeForOneInputProcessFunction(
OneInputStreamProcessFunction<IN, OUT> processFunction,
TypeInformation<IN> inTypeInformation) {
+ TypeInformation<OUT> outType;
+ if (processFunction instanceof InternalOneInputWindowStreamProcessFunction) {
+ return TypeExtractor.getUnaryOperatorReturnType(
+ ((InternalOneInputWindowStreamProcessFunction<IN, OUT, ?>) processFunction)
+ .getWindowProcessFunction(),
+ OneInputWindowStreamProcessFunction.class,
+ 0,
+ 1,
+ new int[] {1, 0},
+ null,
+ Utils.getCallLocationName(),
+ false);
+ }
+
if (processFunction instanceof EventTimeWrappedOneInputStreamProcessFunction) {
processFunction =
((EventTimeWrappedOneInputStreamProcessFunction) processFunction)
@@ -111,6 +141,22 @@ public final class StreamUtils {
true);
}
+ if (processFunction instanceof InternalTwoInputWindowStreamProcessFunction) {
+ return TypeExtractor.getBinaryOperatorReturnType(
+ ((InternalTwoInputWindowStreamProcessFunction<IN1, IN2, OUT, ?>)
+ processFunction)
+ .getWindowProcessFunction(),
+ TwoInputNonBroadcastWindowStreamProcessFunction.class,
+ 0,
+ 1,
+ 2,
+ new int[] {2, 0},
+ null,
+ null,
+ Utils.getCallLocationName(),
+ false);
+ }
+
if (processFunction instanceof EventTimeWrappedTwoInputNonBroadcastStreamProcessFunction) {
processFunction =
((EventTimeWrappedTwoInputNonBroadcastStreamProcessFunction) processFunction)
@@ -168,6 +214,34 @@ public final class StreamUtils {
TwoOutputStreamProcessFunction<IN, OUT1, OUT2>
twoOutputStreamProcessFunction,
TypeInformation<IN> inTypeInformation) {
+ if (twoOutputStreamProcessFunction
+ instanceof InternalTwoOutputWindowStreamProcessFunction) {
+ TypeInformation<OUT1> firstOutputType =
+ TypeExtractor.getUnaryOperatorReturnType(
+ ((InternalTwoOutputWindowStreamProcessFunction<IN, OUT1, OUT2, ?>)
+ twoOutputStreamProcessFunction)
+ .getWindowProcessFunction(),
+ TwoOutputWindowStreamProcessFunction.class,
+ 0,
+ 1,
+ new int[] {1, 0},
+ inTypeInformation,
+ Utils.getCallLocationName(),
+ true);
+ TypeInformation<OUT2> secondOutputType =
+ TypeExtractor.getUnaryOperatorReturnType(
+ ((InternalTwoOutputWindowStreamProcessFunction<IN, OUT1, OUT2, ?>)
+ twoOutputStreamProcessFunction)
+ .getWindowProcessFunction(),
+ TwoOutputWindowStreamProcessFunction.class,
+ 0,
+ 2,
+ new int[] {2, 0},
+ inTypeInformation,
+ Utils.getCallLocationName(),
+ true);
+ return Tuple2.of(firstOutputType, secondOutputType);
+ }
if (twoOutputStreamProcessFunction
instanceof EventTimeWrappedTwoOutputStreamProcessFunction) {
@@ -293,6 +367,44 @@ public final class StreamUtils {
return transform;
}
+ /** Construct and return a {@link TwoInputTransformation} from two input streams. */
+ public static <K, IN1, IN2, OUT>
+ TwoInputTransformation<IN1, IN2, OUT> getTwoInputTransformation(
+ String operatorName,
+ Transformation<IN1> inputStream1,
+ Transformation<IN2> inputStream2,
+ TypeInformation<OUT> outTypeInformation,
+ TwoInputStreamOperator<IN1, IN2, OUT> operator,
+ KeySelector<IN1, K> keySelector1,
+ TypeInformation<K> keyType1,
+ KeySelector<IN2, K> keySelector2,
+ TypeInformation<K> keyType2) {
+
+ if (!(keyType1.canEqual(keyType2) && keyType1.equals(keyType2))) {
+ throw new UnsupportedOperationException(
+ "Key types if input KeyedStreams "
+ + "don't match: "
+ + keyType1
+ + " and "
+ + keyType2
+ + ".");
+ }
+
+ TwoInputTransformation<IN1, IN2, OUT> transform =
+ new TwoInputTransformation<>(
+ inputStream1,
+ inputStream2,
+ operatorName,
+ SimpleOperatorFactory.of(operator),
+ outTypeInformation,
+ // inputStream1 & 2 share the same env.
+ inputStream1.getParallelism(),
+ false);
+ transform.setStateKeySelectors(keySelector1, keySelector2);
+ transform.setStateKeyType(keyType1);
+ return transform;
+ }
+
/** Construct and return a new DataStream with one input operator. */
public static <T, R> AbstractDataStream<R> transformOneInputOperator(
String operatorName,
@@ -319,6 +431,118 @@ public final class StreamUtils {
return returnStream;
}
+ /** Construct and return a {@link OneInputTransformation} of one input window operator. */
+ public static <K, IN, OUT, W extends Window> Transformation<OUT> transformOneInputWindow(
+ ExecutionConfig executionConfig,
+ AbstractDataStream<IN> inputStream,
+ TypeInformation<IN> inputType,
+ TypeInformation<OUT> outType,
+ InternalOneInputWindowStreamProcessFunction<IN, OUT, W> internalWindowFunction,
+ KeySelector<IN, K> keySelector,
+ TypeInformation<K> keyType) {
+ WindowAssigner<IN, W> assigner = internalWindowFunction.getAssigner();
+ ListStateDescriptor<IN> stateDesc =
+ new ListStateDescriptor<>(
+ "window-state", TypeExtractor.createTypeInfo(inputType.getTypeClass()));
+
+ OneInputWindowProcessOperator<K, IN, OUT, W> windowProcessOperator =
+ new OneInputWindowProcessOperator<>(
+ internalWindowFunction,
+ assigner,
+ internalWindowFunction.getTrigger(),
+ assigner.getWindowSerializer(executionConfig),
+ stateDesc,
+ internalWindowFunction.getAllowedLateness());
+ return StreamUtils.getOneInputKeyedTransformation(
+ "Window", inputStream, outType, windowProcessOperator, keySelector, keyType);
+ }
+
+ /** Construct and return a {@link OneInputTransformation} of one input window operator. */
+ public static <K, IN1, IN2, OUT, W extends Window>
+ Transformation<OUT> transformTwoInputNonBroadcastWindow(
+ ExecutionConfig executionConfig,
+ Transformation<IN1> inputStream1,
+ TypeInformation<IN1> inputType1,
+ Transformation<IN2> inputStream2,
+ TypeInformation<IN2> inputType2,
+ TypeInformation<OUT> outType,
+ InternalTwoInputWindowStreamProcessFunction<IN1, IN2, OUT, W>
+ internalWindowFunction,
+ KeySelector<IN1, K> keySelector1,
+ TypeInformation<K> keyType1,
+ KeySelector<IN2, K> keySelector2,
+ TypeInformation<K> keyType2) {
+ WindowAssigner<TaggedUnion<IN1, IN2>, W> assigner = internalWindowFunction.getAssigner();
+ ListStateDescriptor<IN1> leftStateDesc =
+ new ListStateDescriptor<>(
+ "two-input-window-left-state",
+ TypeExtractor.createTypeInfo(inputType1.getTypeClass()));
+
+ ListStateDescriptor<IN2> rightStateDesc =
+ new ListStateDescriptor<>(
+ "two-input-window-right-state",
+ TypeExtractor.createTypeInfo(inputType2.getTypeClass()));
+
+ TwoInputNonBroadcastWindowProcessOperator<K, IN1, IN2, OUT, W> windowProcessOperator =
+ new TwoInputNonBroadcastWindowProcessOperator<>(
+ internalWindowFunction,
+ assigner,
+ internalWindowFunction.getTrigger(),
+ assigner.getWindowSerializer(executionConfig),
+ leftStateDesc,
+ rightStateDesc,
+ internalWindowFunction.getAllowedLateness());
+ return StreamUtils.getTwoInputTransformation(
+ "TwoInput-NonBroadcast-Window",
+ inputStream1,
+ inputStream2,
+ outType,
+ windowProcessOperator,
+ keySelector1,
+ keyType1,
+ keySelector2,
+ keyType2);
+ }
+
+ /** Construct and return a {@link OneInputTransformation} of two output window operator. */
+ public static <K, IN, OUT1, OUT2, W extends Window>
+ Transformation<OUT1> transformTwoOutputWindow(
+ ExecutionConfig executionConfig,
+ AbstractDataStream<IN> inputStream,
+ TypeInformation<IN> inputType,
+ TypeInformation<OUT1> outType1,
+ TypeInformation<OUT2> outType2,
+ OutputTag<OUT2> secondOutputTag,
+ InternalTwoOutputWindowStreamProcessFunction<IN, OUT1, OUT2, W>
+ internalWindowFunction,
+ KeySelector<IN, K> keySelector,
+ TypeInformation<K> keyType) {
+ WindowAssigner<IN, W> assigner = internalWindowFunction.getAssigner();
+ ListStateDescriptor<IN> stateDesc =
+ new ListStateDescriptor<>(
+ "two-output-window-state",
+ TypeExtractor.createTypeInfo(inputType.getTypeClass()));
+
+ TwoOutputWindowProcessOperator<K, IN, OUT1, OUT2, W> windowProcessOperator =
+ new TwoOutputWindowProcessOperator<>(
+ internalWindowFunction,
+ secondOutputTag,
+ null,
+ null,
+ assigner,
+ internalWindowFunction.getTrigger(),
+ assigner.getWindowSerializer(executionConfig),
+ stateDesc,
+ internalWindowFunction.getAllowedLateness());
+ return StreamUtils.getOneInputKeyedTransformation(
+ "TwoOutput-Window",
+ inputStream,
+ outType1,
+ windowProcessOperator,
+ keySelector,
+ keyType);
+ }
+
/** Add sink operator to the input stream. */
public static <T> DataStreamV2SinkTransformation<T, T> addSinkOperator(
AbstractDataStream<T> inputStream, Sink<T> sink, TypeInformation<T> typeInformation) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java
index 80493086d35..3ce835c8f28 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java
@@ -236,7 +236,7 @@ public abstract class AbstractAsyncStateStreamOperator<OUT> extends AbstractStre
}
/** Create new state (v2) based on new state descriptor. */
- protected <N, S extends State, T> S getOrCreateKeyedState(
+ public <N, S extends State, T> S getOrCreateKeyedState(
@Nonnull N defaultNamespace,
@Nonnull TypeSerializer<N> namespaceSerializer,
@Nonnull StateDescriptor<T> stateDescriptor)
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/CoGroupedStreams.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/CoGroupedStreams.java
index 16b9fae9284..cf102097cfc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/CoGroupedStreams.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/CoGroupedStreams.java
@@ -40,6 +40,7 @@ import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.TaggedUnion;
import javax.annotation.Nullable;
@@ -47,7 +48,6 @@ import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
-import java.util.Objects;
import java.util.Optional;
import static java.util.Objects.requireNonNull;
@@ -392,59 +392,9 @@ public class CoGroupedStreams<T1, T2> {
}
// ------------------------------------------------------------------------
- // Data type and type information for Tagged Union
+ // Type information for Tagged Union
// ------------------------------------------------------------------------
- /** Internal class for implementing tagged union co-group. */
- @Internal
- public static class TaggedUnion<T1, T2> {
- private final T1 one;
- private final T2 two;
-
- private TaggedUnion(T1 one, T2 two) {
- this.one = one;
- this.two = two;
- }
-
- public boolean isOne() {
- return one != null;
- }
-
- public boolean isTwo() {
- return two != null;
- }
-
- public T1 getOne() {
- return one;
- }
-
- public T2 getTwo() {
- return two;
- }
-
- public static <T1, T2> TaggedUnion<T1, T2> one(T1 one) {
- return new TaggedUnion<>(one, null);
- }
-
- public static <T1, T2> TaggedUnion<T1, T2> two(T2 two) {
- return new TaggedUnion<>(null, two);
- }
-
- @Override
- public boolean equals(Object obj) {
- if (obj == this) {
- return true;
- }
-
- if (!(obj instanceof TaggedUnion)) {
- return false;
- }
-
- TaggedUnion other = (TaggedUnion) obj;
- return Objects.equals(one, other.one) && Objects.equals(two, other.two);
- }
- }
-
private static class UnionTypeInfo<T1, T2> extends TypeInformation<TaggedUnion<T1, T2>> {
private static final long serialVersionUID = 1L;
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/JoinedStreams.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/JoinedStreams.java
index 11777fd4157..0608497501d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/JoinedStreams.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/JoinedStreams.java
@@ -27,12 +27,12 @@ import org.apache.flink.api.common.functions.WrappingFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.typeutils.TypeExtractor;
-import org.apache.flink.streaming.api.datastream.CoGroupedStreams.TaggedUnion;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.evictors.Evictor;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;
+import org.apache.flink.util.TaggedUnion;
import javax.annotation.Nullable;
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
index 7c9cbe908f1..69d29bb818c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
@@ -542,7 +542,7 @@ public abstract class AbstractStreamOperator<OUT>
* @throws IllegalStateException Thrown, if the key/value state was already initialized.
* @throws Exception Thrown, if the state backend cannot create the key/value state.
*/
- protected <S extends State, N> S getPartitionedState(
+ public <S extends State, N> S getPartitionedState(
N namespace,
TypeSerializer<N> namespaceSerializer,
StateDescriptor<S, ?> stateDescriptor)
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/datastream/UnionSerializerTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/datastream/UnionSerializerTest.java
index 4a8f0326fd5..8396f144aac 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/datastream/UnionSerializerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/datastream/UnionSerializerTest.java
@@ -22,9 +22,9 @@ import org.apache.flink.api.common.serialization.SerializerConfigImpl;
import org.apache.flink.api.common.typeutils.SerializerTestBase;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
-import org.apache.flink.streaming.api.datastream.CoGroupedStreams.TaggedUnion;
import org.apache.flink.streaming.api.datastream.CoGroupedStreams.UnionSerializer;
import org.apache.flink.testutils.DeeplyEqualsChecker;
+import org.apache.flink.util.TaggedUnion;
/** Serializer tests for {@link UnionSerializer}. */
class UnionSerializerTest extends SerializerTestBase<TaggedUnion<Object, Object>> {
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/datastream/UnionSerializerUpgradeTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/datastream/UnionSerializerUpgradeTest.java
index 59df994f731..a67a9696f61 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/datastream/UnionSerializerUpgradeTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/datastream/UnionSerializerUpgradeTest.java
@@ -25,8 +25,8 @@ import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
import org.apache.flink.api.common.typeutils.TypeSerializerUpgradeTestBase;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
-import org.apache.flink.streaming.api.datastream.CoGroupedStreams.TaggedUnion;
import org.apache.flink.streaming.api.datastream.CoGroupedStreams.UnionSerializer;
+import org.apache.flink.util.TaggedUnion;
import org.assertj.core.api.Condition;
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/extension/window/WindowITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/extension/window/WindowITCase.java
new file mode 100644
index 00000000000..3e3986730a1
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/extension/window/WindowITCase.java
@@ -0,0 +1,620 @@
+/*
+ * 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.test.streaming.api.datastream.extension.window;
+
+import org.apache.flink.api.common.state.StateDeclaration;
+import org.apache.flink.api.common.state.StateDeclarations;
+import org.apache.flink.api.common.state.ValueStateDeclaration;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.api.common.typeinfo.TypeDescriptors;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.dsv2.WrappedSink;
+import org.apache.flink.api.connector.dsv2.WrappedSource;
+import org.apache.flink.connector.datagen.source.DataGeneratorSource;
+import org.apache.flink.connector.datagen.source.GeneratorFunction;
+import org.apache.flink.datastream.api.ExecutionEnvironment;
+import org.apache.flink.datastream.api.builtin.BuiltinFuncs;
+import org.apache.flink.datastream.api.common.Collector;
+import org.apache.flink.datastream.api.context.PartitionedContext;
+import org.apache.flink.datastream.api.context.TwoOutputPartitionedContext;
+import org.apache.flink.datastream.api.extension.eventtime.EventTimeExtension;
+import org.apache.flink.datastream.api.extension.window.context.OneInputWindowContext;
+import org.apache.flink.datastream.api.extension.window.context.TwoInputWindowContext;
+import org.apache.flink.datastream.api.extension.window.function.OneInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.api.extension.window.function.TwoInputNonBroadcastWindowStreamProcessFunction;
+import org.apache.flink.datastream.api.extension.window.function.TwoOutputWindowStreamProcessFunction;
+import org.apache.flink.datastream.api.extension.window.strategy.WindowStrategy;
+import org.apache.flink.datastream.api.function.OneInputStreamProcessFunction;
+import org.apache.flink.datastream.api.function.TwoInputNonBroadcastStreamProcessFunction;
+import org.apache.flink.datastream.api.function.TwoOutputStreamProcessFunction;
+import org.apache.flink.datastream.api.stream.NonKeyedPartitionStream;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test the Window extension on DataStream V2. */
+class WindowITCase implements Serializable {
+
+ private static final long NUM_ELEMENT_PER_SOURCE = 10_000;
+ private static final long EXPECTED_ELEMENT_SUM_PER_SOURCE =
+ (0 + (NUM_ELEMENT_PER_SOURCE - 1)) * NUM_ELEMENT_PER_SOURCE / 2;
+
+ private transient ExecutionEnvironment env;
+
+ @BeforeEach
+ void before() throws Exception {
+ env = ExecutionEnvironment.getInstance();
+ }
+
+ @AfterEach
+ void after() throws Exception {
+ TestCollectResultProcessFunction.elementSum = 0;
+ TestCollectResultProcessFunction.elementCount = 0;
+ }
+
+ // =================== one input ==========================
+
+ @Test
+ void testOneInputWindowFunctionWithEventTimeGlobalWindow() throws Exception {
+ NonKeyedPartitionStream<ValueWithTimestamp> source = getSource("source", 0, 0, true);
+
+ OneInputStreamProcessFunction<ValueWithTimestamp, ValueWithTimestamp>
+ windowProcessFunction =
+ BuiltinFuncs.window(
+ WindowStrategy.global(),
+ new OneInputWindowStreamProcessFunction<
+ ValueWithTimestamp, ValueWithTimestamp>() {
+ @Override
+ public void onTrigger(
+ Collector<ValueWithTimestamp> output,
+ PartitionedContext<ValueWithTimestamp> ctx,
+ OneInputWindowContext<ValueWithTimestamp> windowContext)
+ throws Exception {
+ Iterable<ValueWithTimestamp> allRecords =
+ windowContext.getAllRecords();
+ int recordCount = 0;
+ for (ValueWithTimestamp valueWithTimestamp : allRecords) {
+ output.collect(valueWithTimestamp);
+ recordCount++;
+ }
+ assertThat(recordCount)
+ .isEqualTo(NUM_ELEMENT_PER_SOURCE / 2);
+ }
+ });
+
+ source.keyBy(x -> x.value % 2)
+ .process(windowProcessFunction)
+ .process(new TestCollectResultProcessFunction())
+ .withParallelism(1)
+ .toSink(new WrappedSink<>(new DiscardingSink<>()));
+ env.execute("testOneInputWindowFunctionWithEventTimeGlobalWindow");
+
+ assertThat(TestCollectResultProcessFunction.elementSum)
+ .isEqualTo(EXPECTED_ELEMENT_SUM_PER_SOURCE);
+ assertThat(TestCollectResultProcessFunction.elementCount).isEqualTo(NUM_ELEMENT_PER_SOURCE);
+ }
+
+ @Test
+ void testOneInputWindowFunctionWithEventTimeTumblingWindow() throws Exception {
+ NonKeyedPartitionStream<ValueWithTimestamp> source = getSource("source", 0, 0, true);
+
+ OneInputStreamProcessFunction<ValueWithTimestamp, ValueWithTimestamp>
+ windowProcessFunction =
+ BuiltinFuncs.window(
+ WindowStrategy.tumbling(Duration.ofSeconds(10)),
+ new OneInputWindowStreamProcessFunction<
+ ValueWithTimestamp, ValueWithTimestamp>() {
+ @Override
+ public void onTrigger(
+ Collector<ValueWithTimestamp> output,
+ PartitionedContext<ValueWithTimestamp> ctx,
+ OneInputWindowContext<ValueWithTimestamp> windowContext)
+ throws Exception {
+ Iterable<ValueWithTimestamp> allRecords =
+ windowContext.getAllRecords();
+ int recordCount = 0;
+ for (ValueWithTimestamp valueWithTimestamp : allRecords) {
+ output.collect(valueWithTimestamp);
+ recordCount++;
+ }
+ assertThat(recordCount).isEqualTo(5);
+ }
+ });
+
+ source.keyBy(x -> x.value % 2)
+ .process(windowProcessFunction)
+ .process(new TestCollectResultProcessFunction())
+ .withParallelism(1)
+ .toSink(new WrappedSink<>(new DiscardingSink<>()));
+ env.execute("testOneInputWindowFunctionWithEventTimeTumblingWindow");
+
+ assertThat(TestCollectResultProcessFunction.elementSum)
+ .isEqualTo(EXPECTED_ELEMENT_SUM_PER_SOURCE);
+ assertThat(TestCollectResultProcessFunction.elementCount).isEqualTo(NUM_ELEMENT_PER_SOURCE);
+ }
+
+ @Test
+ void testOneInputWindowFunctionWithEventTimeSlidingWindow() throws Exception {
+ NonKeyedPartitionStream<ValueWithTimestamp> source = getSource("source", 0, 0, true);
+
+ OneInputStreamProcessFunction<ValueWithTimestamp, ValueWithTimestamp>
+ windowProcessFunction =
+ BuiltinFuncs.window(
+ WindowStrategy.sliding(
+ Duration.ofSeconds(10), Duration.ofSeconds(1)),
+ new OneInputWindowStreamProcessFunction<
+ ValueWithTimestamp, ValueWithTimestamp>() {
+ long preWindowStartTime = -1;
+
+ @Override
+ public void onTrigger(
+ Collector<ValueWithTimestamp> output,
+ PartitionedContext<ValueWithTimestamp> ctx,
+ OneInputWindowContext<ValueWithTimestamp> windowContext)
+ throws Exception {
+ Iterable<ValueWithTimestamp> allRecords =
+ windowContext.getAllRecords();
+ for (ValueWithTimestamp valueWithTimestamp : allRecords) {
+ output.collect(valueWithTimestamp);
+ }
+ if (preWindowStartTime != -1) {
+ assertThat(
+ windowContext.getStartTime()
+ - preWindowStartTime)
+ .isEqualTo(1000);
+ preWindowStartTime = windowContext.getStartTime();
+ }
+ }
+ });
+
+ source.keyBy(x -> x.value % 2)
+ .process(windowProcessFunction)
+ .toSink(new WrappedSink<>(new DiscardingSink<>()));
+ env.execute("testOneInputWindowFunctionWithEventTimeSlidingWindow");
+ }
+
+ @Test
+ void testOneInputWindowFunctionWithEventTimeSessionWindow() throws Exception {
+ NonKeyedPartitionStream<ValueWithTimestamp> source = getSource("source", 5, 5, true);
+
+ OneInputStreamProcessFunction<ValueWithTimestamp, ValueWithTimestamp>
+ windowProcessFunction =
+ BuiltinFuncs.window(
+ WindowStrategy.session(Duration.ofSeconds(3)),
+ new OneInputWindowStreamProcessFunction<
+ ValueWithTimestamp, ValueWithTimestamp>() {
+ @Override
+ public void onTrigger(
+ Collector<ValueWithTimestamp> output,
+ PartitionedContext<ValueWithTimestamp> ctx,
+ OneInputWindowContext<ValueWithTimestamp> windowContext)
+ throws Exception {
+ Iterable<ValueWithTimestamp> allRecords =
+ windowContext.getAllRecords();
+ int recordCount = 0;
+ for (ValueWithTimestamp valueWithTimestamp : allRecords) {
+ output.collect(valueWithTimestamp);
+ recordCount++;
+ }
+ assertThat(recordCount).isEqualTo(5);
+ }
+ });
+
+ source.keyBy(x -> 0)
+ .process(windowProcessFunction)
+ .process(new TestCollectResultProcessFunction())
+ .withParallelism(1)
+ .toSink(new WrappedSink<>(new DiscardingSink<>()));
+ env.execute("testOneInputWindowFunctionWithEventTimeSessionWindow");
+
+ assertThat(TestCollectResultProcessFunction.elementSum)
+ .isEqualTo(EXPECTED_ELEMENT_SUM_PER_SOURCE);
+ assertThat(TestCollectResultProcessFunction.elementCount).isEqualTo(NUM_ELEMENT_PER_SOURCE);
+ }
+
+ @Test
+ void testOneInputWindowFunctionUseWindowStateToAgg() throws Exception {
+ NonKeyedPartitionStream<ValueWithTimestamp> source = getSource("source", 0, 0, true);
+
+ OneInputStreamProcessFunction<ValueWithTimestamp, ValueWithTimestamp>
+ windowProcessFunction =
+ BuiltinFuncs.window(
+ WindowStrategy.tumbling(Duration.ofSeconds(5)),
+ new OneInputWindowStreamProcessFunction<
+ ValueWithTimestamp, ValueWithTimestamp>() {
+
+ private ValueStateDeclaration<Long> sumStateDeclaration =
+ StateDeclarations.valueState(
+ "sum", TypeDescriptors.LONG);
+ private ValueStateDeclaration<Long> countStateDeclaration =
+ StateDeclarations.valueState(
+ "count", TypeDescriptors.LONG);
+
+ @Override
+ public Set<StateDeclaration> useWindowStates() {
+ return Set.of(sumStateDeclaration, countStateDeclaration);
+ }
+
+ @Override
+ public void onRecord(
+ ValueWithTimestamp record,
+ Collector<ValueWithTimestamp> output,
+ PartitionedContext<ValueWithTimestamp> ctx,
+ OneInputWindowContext<ValueWithTimestamp> windowContext)
+ throws Exception {
+ Optional<ValueState<Long>> countState =
+ windowContext.getWindowState(countStateDeclaration);
+ if (countState.isPresent()
+ && countState.get().value() != null) {
+ countState.get().update(countState.get().value() + 1);
+ } else {
+ countState.get().update(1L);
+ }
+
+ Optional<ValueState<Long>> sumState =
+ windowContext.getWindowState(sumStateDeclaration);
+ long sum = 0;
+ if (sumState.isPresent()
+ && sumState.get().value() != null) {
+ sum = sumState.get().value();
+ }
+ sumState.get().update(sum + record.getValue());
+ }
+
+ @Override
+ public void onTrigger(
+ Collector<ValueWithTimestamp> output,
+ PartitionedContext<ValueWithTimestamp> ctx,
+ OneInputWindowContext<ValueWithTimestamp> windowContext)
+ throws Exception {
+ Optional<ValueState<Long>> countState =
+ windowContext.getWindowState(countStateDeclaration);
+ assertThat(countState.get().value()).isEqualTo(5);
+
+ Optional<ValueState<Long>> sumState =
+ windowContext.getWindowState(sumStateDeclaration);
+ output.collect(
+ new ValueWithTimestamp(
+ sumState.get().value(), -1L));
+ }
+ });
+
+ source.keyBy(x -> 0)
+ .process(windowProcessFunction)
+ .process(new TestCollectResultProcessFunction())
+ .withParallelism(1)
+ .toSink(new WrappedSink<>(new DiscardingSink<>()));
+ env.execute("testOneInputWindowFunctionUseWindowStateToAgg");
+
+ assertThat(TestCollectResultProcessFunction.elementSum)
+ .isEqualTo(EXPECTED_ELEMENT_SUM_PER_SOURCE);
+ assertThat(TestCollectResultProcessFunction.elementCount)
+ .isEqualTo(NUM_ELEMENT_PER_SOURCE / 5);
+ }
+
+ @Test
+ void testOneInputWindowFunctionWithEventTimeTumblingWindowUseGlobalStream() throws Exception {
+ NonKeyedPartitionStream<ValueWithTimestamp> source = getSource("source", 0, 0, true);
+
+ OneInputStreamProcessFunction<ValueWithTimestamp, ValueWithTimestamp>
+ windowProcessFunction =
+ BuiltinFuncs.window(
+ WindowStrategy.tumbling(Duration.ofSeconds(5)),
+ new OneInputWindowStreamProcessFunction<
+ ValueWithTimestamp, ValueWithTimestamp>() {
+ @Override
+ public void onTrigger(
+ Collector<ValueWithTimestamp> output,
+ PartitionedContext<ValueWithTimestamp> ctx,
+ OneInputWindowContext<ValueWithTimestamp> windowContext)
+ throws Exception {
+ Iterable<ValueWithTimestamp> allRecords =
+ windowContext.getAllRecords();
+ int recordCount = 0;
+ for (ValueWithTimestamp valueWithTimestamp : allRecords) {
+ output.collect(valueWithTimestamp);
+ recordCount++;
+ }
+ assertThat(recordCount).isEqualTo(5);
+ }
+ });
+
+ source.global()
+ .process(windowProcessFunction)
+ .process(new TestCollectResultProcessFunction())
+ .withParallelism(1)
+ .toSink(new WrappedSink<>(new DiscardingSink<>()));
+ env.execute("testOneInputWindowFunctionWithEventTimeTumblingWindowUseGlobalStream");
+
+ assertThat(TestCollectResultProcessFunction.elementSum)
+ .isEqualTo(EXPECTED_ELEMENT_SUM_PER_SOURCE);
+ assertThat(TestCollectResultProcessFunction.elementCount).isEqualTo(NUM_ELEMENT_PER_SOURCE);
+ }
+
+ // ======================= two input ==========================
+
+ @Test
+ void testTwoInputWindowFunctionWithEventTimeTumblingWindow() throws Exception {
+ NonKeyedPartitionStream<ValueWithTimestamp> source1 = getSource("source1", 0, 0, true);
+ NonKeyedPartitionStream<ValueWithTimestamp> source2 = getSource("source2", 0, 0, true);
+
+ TwoInputNonBroadcastStreamProcessFunction<
+ ValueWithTimestamp, ValueWithTimestamp, ValueWithTimestamp>
+ windowProcessFunction =
+ BuiltinFuncs.window(
+ WindowStrategy.tumbling(Duration.ofSeconds(10)),
+ new TwoInputNonBroadcastWindowStreamProcessFunction<
+ ValueWithTimestamp,
+ ValueWithTimestamp,
+ ValueWithTimestamp>() {
+
+ @Override
+ public void onTrigger(
+ Collector<ValueWithTimestamp> output,
+ PartitionedContext<ValueWithTimestamp> ctx,
+ TwoInputWindowContext<
+ ValueWithTimestamp, ValueWithTimestamp>
+ windowContext)
+ throws Exception {
+ Iterable<ValueWithTimestamp> allRecords1 =
+ windowContext.getAllRecords1();
+ int recordCount1 = 0;
+ for (ValueWithTimestamp valueWithTimestamp : allRecords1) {
+ output.collect(valueWithTimestamp);
+ recordCount1++;
+ }
+ assertThat(recordCount1).isEqualTo(5);
+
+ Iterable<ValueWithTimestamp> allRecords2 =
+ windowContext.getAllRecords2();
+ int recordCount2 = 0;
+ for (ValueWithTimestamp valueWithTimestamp : allRecords2) {
+ output.collect(valueWithTimestamp);
+ recordCount2++;
+ }
+ assertThat(recordCount2).isEqualTo(5);
+ }
+ });
+
+ source1.keyBy(x -> x.value % 2)
+ .connectAndProcess(source2.keyBy(x -> x.value % 2), windowProcessFunction)
+ .process(new TestCollectResultProcessFunction())
+ .withParallelism(1)
+ .toSink(new WrappedSink<>(new DiscardingSink<>()));
+ env.execute("testTwoInputWindowFunctionWithEventTimeTumblingWindow");
+
+ assertThat(TestCollectResultProcessFunction.elementSum)
+ .isEqualTo(EXPECTED_ELEMENT_SUM_PER_SOURCE * 2);
+ assertThat(TestCollectResultProcessFunction.elementCount)
+ .isEqualTo(NUM_ELEMENT_PER_SOURCE * 2);
+ }
+
+ // ======================= two output ==========================
+
+ @Test
+ void testTwoOutputWindowFunctionWithEventTimeTumblingWindow() throws Exception {
+ NonKeyedPartitionStream<ValueWithTimestamp> source = getSource("source", 0, 0, true);
+
+ TwoOutputStreamProcessFunction<ValueWithTimestamp, ValueWithTimestamp, ValueWithTimestamp>
+ windowProcessFunction =
+ BuiltinFuncs.window(
+ WindowStrategy.tumbling(Duration.ofSeconds(10)),
+ new TwoOutputWindowStreamProcessFunction<
+ ValueWithTimestamp,
+ ValueWithTimestamp,
+ ValueWithTimestamp>() {
+
+ @Override
+ public void onTrigger(
+ Collector<ValueWithTimestamp> output1,
+ Collector<ValueWithTimestamp> output2,
+ TwoOutputPartitionedContext<
+ ValueWithTimestamp, ValueWithTimestamp>
+ ctx,
+ OneInputWindowContext<ValueWithTimestamp> windowContext)
+ throws Exception {
+ Iterable<ValueWithTimestamp> allRecords =
+ windowContext.getAllRecords();
+ int recordCount = 0;
+ for (ValueWithTimestamp valueWithTimestamp : allRecords) {
+ output1.collect(valueWithTimestamp);
+ output2.collect(valueWithTimestamp);
+ recordCount++;
+ }
+ assertThat(recordCount).isEqualTo(5);
+ }
+ });
+
+ NonKeyedPartitionStream.ProcessConfigurableAndTwoNonKeyedPartitionStream<
+ ValueWithTimestamp, ValueWithTimestamp>
+ twoOutputStream = source.keyBy(x -> x.value % 2).process(windowProcessFunction);
+
+ twoOutputStream
+ .getFirst()
+ .process(new TestCollectResultProcessFunction())
+ .withParallelism(1)
+ .toSink(new WrappedSink<>(new DiscardingSink<>()));
+
+ twoOutputStream
+ .getSecond()
+ .process(new TestCollectResultProcessFunction())
+ .withParallelism(1)
+ .toSink(new WrappedSink<>(new DiscardingSink<>()));
+
+ env.execute("testTwoOutputWindowFunctionWithEventTimeTumblingWindow");
+
+ assertThat(TestCollectResultProcessFunction.elementSum)
+ .isEqualTo(EXPECTED_ELEMENT_SUM_PER_SOURCE * 2);
+ assertThat(TestCollectResultProcessFunction.elementCount)
+ .isEqualTo(NUM_ELEMENT_PER_SOURCE * 2);
+ }
+
+ public static class ValueWithTimestamp {
+
+ private final long value;
+
+ /** The event time of the element in milliseconds. */
+ private final long timestamp;
+
+ public ValueWithTimestamp(long value, long timestamp) {
+ this.value = value;
+ this.timestamp = timestamp;
+ }
+
+ public long getValue() {
+ return value;
+ }
+
+ public long getTimestamp() {
+ return timestamp;
+ }
+
+ @Override
+ public String toString() {
+ return "ValueWithTimestamp{" + "value=" + value + ", timestamp=" + timestamp + '}';
+ }
+ }
+
+ /**
+ * {@link TestGeneratorFunction} is used to generate test data. The first data he generates is
+ * ValueWithTimestamp{value=0, timestamp=0}, each subsequent data than the previous data, value
+ * increased by one, timestamp increased by one second.
+ */
+ private static class TestGeneratorFunction
+ implements GeneratorFunction<Long, ValueWithTimestamp> {
+
+ private long curTime = 0L;
+
+ private int curValue = 0;
+
+ // Every {@code countStep} elements, time is incremented by {@code timeStep} seconds. Used
+ // to simulate the presence of gaps in the time of elements in the SessionWindow.
+ private long countStep;
+
+ private long timeStep;
+
+ private int generatedElementCount = 0;
+
+ public TestGeneratorFunction() {
+ this(0, 0);
+ }
+
+ public TestGeneratorFunction(long countStep, long timeStep) {
+ this.countStep = countStep;
+ this.timeStep = timeStep;
+ }
+
+ @Override
+ public ValueWithTimestamp map(Long value) throws Exception {
+ ValueWithTimestamp element = new ValueWithTimestamp(curValue, curTime);
+ curValue++;
+ curTime = curTime + Duration.ofSeconds(1).toMillis();
+ generatedElementCount++;
+ if (countStep != 0 && generatedElementCount % countStep == 0) {
+ curTime += Duration.ofSeconds(timeStep).toMillis();
+ }
+ return element;
+ }
+ }
+
+ private NonKeyedPartitionStream<ValueWithTimestamp> getSource(
+ String sourceName, long countStep, long timeStep, boolean isEventTime) {
+ NonKeyedPartitionStream<ValueWithTimestamp> stream =
+ env.fromSource(
+ new WrappedSource<ValueWithTimestamp>(
+ new DataGeneratorSource<ValueWithTimestamp>(
+ new TestGeneratorFunction(countStep, timeStep),
+ NUM_ELEMENT_PER_SOURCE,
+ TypeInformation.of(ValueWithTimestamp.class))),
+ sourceName);
+
+ if (isEventTime) {
+ stream =
+ stream.process(
+ EventTimeExtension.<ValueWithTimestamp>newWatermarkGeneratorBuilder(
+ element -> element.getTimestamp())
+ .perEventWatermark()
+ .buildAsProcessFunction());
+ }
+ return stream;
+ }
+
+ private static class TestOneInputWindowStreamProcessFunction
+ implements OneInputWindowStreamProcessFunction<ValueWithTimestamp, ValueWithTimestamp> {
+
+ private final boolean checkElementCountWhenTrigger;
+ private final int expectElementCountWhenTrigger;
+ private final boolean skipCheckWhenFirstTimeTrigger;
+ private boolean isFirstTimeTrigger = true;
+
+ private TestOneInputWindowStreamProcessFunction(
+ boolean checkElementCountWhenTrigger,
+ int expectElementCountWhenTrigger,
+ boolean skipCheckWhenFirstTimeTrigger) {
+ this.checkElementCountWhenTrigger = checkElementCountWhenTrigger;
+ this.expectElementCountWhenTrigger = expectElementCountWhenTrigger;
+ this.skipCheckWhenFirstTimeTrigger = skipCheckWhenFirstTimeTrigger;
+ }
+
+ @Override
+ public void onTrigger(
+ Collector<ValueWithTimestamp> output,
+ PartitionedContext<ValueWithTimestamp> ctx,
+ OneInputWindowContext<ValueWithTimestamp> windowContext)
+ throws Exception {
+ int elementCount = 0;
+ Iterable<ValueWithTimestamp> allRecords = windowContext.getAllRecords();
+ for (ValueWithTimestamp record : allRecords) {
+ output.collect(record);
+ elementCount++;
+ }
+ if (checkElementCountWhenTrigger
+ && (!skipCheckWhenFirstTimeTrigger || !isFirstTimeTrigger)) {
+ assertThat(elementCount).isEqualTo(expectElementCountWhenTrigger);
+ }
+
+ isFirstTimeTrigger = false;
+ }
+ }
+
+ private static class TestCollectResultProcessFunction
+ implements OneInputStreamProcessFunction<ValueWithTimestamp, ValueWithTimestamp> {
+ public static long elementSum;
+ public static long elementCount;
+
+ @Override
+ public void processRecord(
+ ValueWithTimestamp record,
+ Collector<ValueWithTimestamp> output,
+ PartitionedContext<ValueWithTimestamp> ctx)
+ throws Exception {
+ elementCount++;
+ elementSum += record.getValue();
+ }
+ }
+}