You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2024/10/01 14:21:42 UTC
(flink) branch master updated: [FLINK-36355][runtime] Remove deprecated SetupableStreamOperator
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
The following commit(s) were added to refs/heads/master by this push:
new 3b2a96854c4 [FLINK-36355][runtime] Remove deprecated SetupableStreamOperator
3b2a96854c4 is described below
commit 3b2a96854c4701f5de0c690c061d3bedbf234406
Author: yunfengzhou-hub <yu...@outlook.com>
AuthorDate: Tue Oct 1 09:21:27 2024 +0800
[FLINK-36355][runtime] Remove deprecated SetupableStreamOperator
---
.../5b9eed8a-5fb6-4373-98ac-3be2a71941b8 | 9 -
.../e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5 | 4 +-
.../compactor/operator/CompactCoordinator.java | 3 +
.../operator/CompactCoordinatorFactory.java | 7 +-
.../operator/CompactCoordinatorStateHandler.java | 5 +
.../CompactCoordinatorStateHandlerFactory.java | 7 +-
.../sink/compactor/operator/CompactorOperator.java | 3 +
.../operator/CompactorOperatorFactory.java | 5 +-
.../operator/CompactorOperatorStateHandler.java | 3 +
.../CompactorOperatorStateHandlerFactory.java | 8 +-
.../file/table/batch/compact/BatchFileWriter.java | 2 -
.../file/table/stream/AbstractStreamingWriter.java | 2 -
.../sink/compactor/CompactCoordinatorTest.java | 22 +-
.../file/sink/compactor/CompactorOperatorTest.java | 6 +-
.../datastream/impl/operators/ProcessOperator.java | 3 -
.../TwoInputBroadcastProcessOperator.java | 2 -
.../TwoInputNonBroadcastProcessOperator.java | 2 -
.../impl/operators/TwoOutputProcessOperator.java | 2 -
.../org/apache/flink/cep/operator/CepOperator.java | 2 +-
.../flink/state/api/KeyedStateTransformation.java | 5 +-
.../state/api/OneInputStateTransformation.java | 9 +-
.../state/api/SavepointWriterOperatorFactory.java | 3 +-
.../state/api/StateBootstrapTransformation.java | 12 +-
.../state/api/WindowedStateTransformation.java | 23 +-
.../api/output/operators/GroupReduceOperator.java | 2 -
.../operators/StateBootstrapWrapperOperator.java | 26 +--
.../StateBootstrapWrapperOperatorFactory.java | 105 +++++++++
.../python/AbstractPythonFunctionOperator.java | 2 -
.../AbstractAsyncStateUdfStreamOperator.java | 2 +-
.../connector/sink2/GlobalCommitterOperator.java | 2 +-
.../api/datastream/AllWindowedStream.java | 42 ++--
.../streaming/api/datastream/DataStreamSource.java | 27 ++-
.../NonKeyedPartitionWindowedStream.java | 4 +
.../source/ContinuousFileReaderOperator.java | 4 +-
.../ContinuousFileReaderOperatorFactory.java | 6 +-
.../api/operators/AbstractStreamOperator.java | 39 ++--
.../api/operators/AbstractUdfStreamOperator.java | 7 +-
.../api/operators/KeyedProcessOperator.java | 2 -
.../api/operators/LegacyKeyedProcessOperator.java | 2 -
.../api/operators/MapPartitionOperator.java | 3 -
.../streaming/api/operators/ProcessOperator.java | 2 -
.../api/operators/SimpleOperatorFactory.java | 18 +-
.../streaming/api/operators/SourceOperator.java | 5 +-
.../api/operators/SourceOperatorFactory.java | 7 +-
.../streaming/api/operators/StreamFilter.java | 1 -
.../streaming/api/operators/StreamFlatMap.java | 1 -
.../flink/streaming/api/operators/StreamMap.java | 1 -
...treamOperator.java => StreamOperatorUtils.java} | 37 ++-
.../streaming/api/operators/StreamProject.java | 2 -
.../flink/streaming/api/operators/StreamSink.java | 1 -
.../streaming/api/operators/StreamSource.java | 1 -
.../collect/CollectSinkOperatorFactory.java | 5 +-
.../sortpartition/KeyedSortPartitionOperator.java | 2 +-
.../sortpartition/SortPartitionOperator.java | 2 +-
.../operators/TimestampsAndWatermarksOperator.java | 8 +-
.../TimestampsAndWatermarksOperatorFactory.java | 60 +++++
.../runtime/operators/sink/CommitterOperator.java | 5 +-
.../operators/sink/CommitterOperatorFactory.java | 5 +-
.../runtime/operators/sink/SinkWriterOperator.java | 5 +-
.../operators/sink/SinkWriterOperatorFactory.java | 7 +-
.../windowing/EvictingWindowOperatorFactory.java | 120 ++++++++++
.../operators/windowing/WindowOperator.java | 18 +-
.../operators/windowing/WindowOperatorFactory.java | 145 ++++++++++++
.../runtime/tasks/StreamIterationTail.java | 19 +-
...tampsAndWatermarksTransformationTranslator.java | 8 +-
.../SourceOperatorSplitWatermarkAlignmentTest.java | 26 ++-
.../api/operators/SourceOperatorTestContext.java | 13 +-
.../api/operators/StateDescriptorPassingTest.java | 23 +-
.../api/operators/StreamingRuntimeContextTest.java | 21 +-
.../operators/source/TestingSourceOperator.java | 38 ++--
.../operators/StreamOperatorChainingTest.java | 7 +-
.../StreamSourceOperatorLatencyMetricsTest.java | 7 +-
.../TimestampsAndWatermarksOperatorTest.java | 12 +-
.../EvictingWindowOperatorContractTest.java | 4 +-
.../RegularWindowOperatorContractTest.java | 8 +-
.../streaming/runtime/tasks/OperatorChainTest.java | 7 +-
.../runtime/tasks/StreamOperatorWrapperTest.java | 2 +-
.../util/AbstractStreamOperatorTestHarness.java | 39 ++--
.../api/operators/async/AsyncWaitOperator.java | 11 +-
.../operators/async/AsyncWaitOperatorFactory.java | 5 +-
.../api/graph/StreamGraphGeneratorTest.java | 2 +-
.../AbstractUdfStreamOperatorLifecycleTest.java | 2 +-
.../runtime/io/RecordProcessorUtilsTest.java | 48 +++-
.../StreamSourceOperatorWatermarksTest.java | 3 +-
.../operators/StreamTaskOperatorTimerTest.java | 16 +-
.../windowing/AllWindowTranslationTest.java | 252 ++++++++++-----------
.../ContinuousProcessingTimeTriggerTest.java | 75 +++---
.../windowing/EvictingWindowOperatorTest.java | 38 ++--
.../windowing/WindowOperatorMigrationTest.java | 56 ++---
.../operators/windowing/WindowOperatorTest.java | 176 +++++++-------
...ToStreamOperatorRecipientExactlyOnceITCase.java | 24 +-
.../tasks/SubtaskCheckpointCoordinatorTest.java | 3 +-
.../exec/common/CommonExecTableSourceScan.java | 18 +-
.../runtime/operators/TableStreamOperator.java | 8 +-
.../window/LocalSlicingWindowAggOperator.java | 2 -
.../bundle/AbstractMapBundleOperator.java | 2 -
.../DynamicFilteringDataCollectorOperator.java | 3 +
...namicFilteringDataCollectorOperatorFactory.java | 6 +-
.../over/BufferDataOverWindowOperator.java | 14 ++
.../over/NonBufferOverWindowOperator.java | 10 +
.../table/runtime/operators/sink/SinkOperator.java | 2 -
.../operators/sort/BaseTemporalSortOperator.java | 5 +-
.../groupwindow/operator/WindowOperator.java | 5 -
.../window/tvf/common/WindowAggOperator.java | 2 -
.../operator/WindowTableFunctionOperatorBase.java | 3 -
.../ProcTimeMiniBatchAssignerOperator.java | 2 -
.../RowTimeMiniBatchAssginerOperator.java | 2 -
.../wmassigners/WatermarkAssignerOperator.java | 6 +-
.../WatermarkAssignerOperatorFactory.java | 14 +-
.../DynamicFilteringDataCollectorOperatorTest.java | 5 +-
.../multipleinput/MultipleInputTestBase.java | 22 +-
.../TestingOneInputStreamOperator.java | 13 +-
.../TestingTwoInputStreamOperator.java | 11 +
.../over/BufferDataOverWindowOperatorTest.java | 34 +--
.../over/NonBufferOverWindowOperatorTest.java | 34 +--
.../graph/OneInputTestStreamOperator.java | 7 +-
.../graph/OneInputTestStreamOperatorFactory.java | 8 +-
.../flink/test/checkpointing/SavepointITCase.java | 10 +-
.../flink/test/scheduling/JMFailoverITCase.java | 11 +-
.../runtime/StreamTaskSelectiveReadingITCase.java | 2 -
.../streaming/runtime/StreamTaskTimerITCase.java | 21 +-
.../test/streaming/runtime/TimestampITCase.java | 6 +-
122 files changed, 1338 insertions(+), 777 deletions(-)
diff --git a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/5b9eed8a-5fb6-4373-98ac-3be2a71941b8 b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/5b9eed8a-5fb6-4373-98ac-3be2a71941b8
index 56f0b028f9d..05081f810d9 100644
--- a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/5b9eed8a-5fb6-4373-98ac-3be2a71941b8
+++ b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/5b9eed8a-5fb6-4373-98ac-3be2a71941b8
@@ -123,24 +123,15 @@ org.apache.flink.streaming.api.operators.AbstractStreamOperator.processWatermark
org.apache.flink.streaming.api.operators.AbstractStreamOperator.processWatermarkStatus2(org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus): Argument leaf type org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.api.operators.AbstractStreamOperator.setKeyContextElement1(org.apache.flink.streaming.runtime.streamrecord.StreamRecord): Argument leaf type org.apache.flink.streaming.runtime.streamrecord.StreamRecord does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.api.operators.AbstractStreamOperator.setKeyContextElement2(org.apache.flink.streaming.runtime.streamrecord.StreamRecord): Argument leaf type org.apache.flink.streaming.runtime.streamrecord.StreamRecord does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
-org.apache.flink.streaming.api.operators.AbstractStreamOperator.setup(org.apache.flink.streaming.runtime.tasks.StreamTask, org.apache.flink.streaming.api.graph.StreamConfig, org.apache.flink.streaming.api.operators.Output): Argument leaf type org.apache.flink.streaming.api.graph.StreamConfig does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
-org.apache.flink.streaming.api.operators.AbstractStreamOperator.setup(org.apache.flink.streaming.runtime.tasks.StreamTask, org.apache.flink.streaming.api.graph.StreamConfig, org.apache.flink.streaming.api.operators.Output): Argument leaf type org.apache.flink.streaming.runtime.streamrecord.StreamRecord does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
-org.apache.flink.streaming.api.operators.AbstractStreamOperator.setup(org.apache.flink.streaming.runtime.tasks.StreamTask, org.apache.flink.streaming.api.graph.StreamConfig, org.apache.flink.streaming.api.operators.Output): Argument leaf type org.apache.flink.streaming.runtime.tasks.StreamTask does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(long, long, org.apache.flink.runtime.checkpoint.CheckpointOptions, org.apache.flink.runtime.state.CheckpointStreamFactory): Argument leaf type org.apache.flink.runtime.checkpoint.CheckpointOptions does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(long, long, org.apache.flink.runtime.checkpoint.CheckpointOptions, org.apache.flink.runtime.state.CheckpointStreamFactory): Argument leaf type org.apache.flink.runtime.state.CheckpointStreamFactory does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(long, long, org.apache.flink.runtime.checkpoint.CheckpointOptions, org.apache.flink.runtime.state.CheckpointStreamFactory): Returned leaf type org.apache.flink.streaming.api.operators.OperatorSnapshotFutures does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
-org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.setup(org.apache.flink.streaming.runtime.tasks.StreamTask, org.apache.flink.streaming.api.graph.StreamConfig, org.apache.flink.streaming.api.operators.Output): Argument leaf type org.apache.flink.streaming.api.graph.StreamConfig does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
-org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.setup(org.apache.flink.streaming.runtime.tasks.StreamTask, org.apache.flink.streaming.api.graph.StreamConfig, org.apache.flink.streaming.api.operators.Output): Argument leaf type org.apache.flink.streaming.runtime.streamrecord.StreamRecord does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @D [...]
-org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.setup(org.apache.flink.streaming.runtime.tasks.StreamTask, org.apache.flink.streaming.api.graph.StreamConfig, org.apache.flink.streaming.api.operators.Output): Argument leaf type org.apache.flink.streaming.runtime.tasks.StreamTask does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.api.operators.Input.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord): Argument leaf type org.apache.flink.streaming.runtime.streamrecord.StreamRecord does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.api.operators.Input.processWatermarkStatus(org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus): Argument leaf type org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.api.operators.Input.setKeyContextElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord): Argument leaf type org.apache.flink.streaming.runtime.streamrecord.StreamRecord does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.api.operators.OneInputStreamOperator.setKeyContextElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord): Argument leaf type org.apache.flink.streaming.runtime.streamrecord.StreamRecord does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.api.operators.Output.collect(org.apache.flink.util.OutputTag, org.apache.flink.streaming.runtime.streamrecord.StreamRecord): Argument leaf type org.apache.flink.streaming.runtime.streamrecord.StreamRecord does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.api.operators.Output.emitWatermarkStatus(org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus): Argument leaf type org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
-org.apache.flink.streaming.api.operators.SetupableStreamOperator.setup(org.apache.flink.streaming.runtime.tasks.StreamTask, org.apache.flink.streaming.api.graph.StreamConfig, org.apache.flink.streaming.api.operators.Output): Argument leaf type org.apache.flink.streaming.api.graph.StreamConfig does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
-org.apache.flink.streaming.api.operators.SetupableStreamOperator.setup(org.apache.flink.streaming.runtime.tasks.StreamTask, org.apache.flink.streaming.api.graph.StreamConfig, org.apache.flink.streaming.api.operators.Output): Argument leaf type org.apache.flink.streaming.runtime.streamrecord.StreamRecord does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
-org.apache.flink.streaming.api.operators.SetupableStreamOperator.setup(org.apache.flink.streaming.runtime.tasks.StreamTask, org.apache.flink.streaming.api.graph.StreamConfig, org.apache.flink.streaming.api.operators.Output): Argument leaf type org.apache.flink.streaming.runtime.tasks.StreamTask does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.api.operators.StreamOperator.getOperatorID(): Returned leaf type org.apache.flink.runtime.jobgraph.OperatorID does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.api.operators.StreamOperator.initializeState(org.apache.flink.streaming.api.operators.StreamTaskStateInitializer): Argument leaf type org.apache.flink.streaming.api.operators.StreamTaskStateInitializer does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.api.operators.StreamOperator.setKeyContextElement1(org.apache.flink.streaming.runtime.streamrecord.StreamRecord): Argument leaf type org.apache.flink.streaming.runtime.streamrecord.StreamRecord does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
diff --git a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5 b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5
index f12a0a13237..0d0ba49c3ef 100644
--- a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5
+++ b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5
@@ -19,7 +19,7 @@ Method <org.apache.flink.runtime.taskexecutor.TaskManagerConfiguration.fromConfi
Method <org.apache.flink.runtime.taskexecutor.TaskManagerConfiguration.fromConfiguration(org.apache.flink.configuration.Configuration, org.apache.flink.runtime.taskexecutor.TaskExecutorResourceSpec, java.lang.String, java.io.File)> calls method <org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils.generateTotalAvailableResourceProfile(org.apache.flink.runtime.taskexecutor.TaskExecutorResourceSpec)> in (TaskManagerConfiguration.java:240)
Method <org.apache.flink.runtime.taskexecutor.TaskManagerServices.createTaskSlotTable(int, org.apache.flink.runtime.taskexecutor.TaskExecutorResourceSpec, long, int, java.util.concurrent.Executor)> calls method <org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils.generateDefaultSlotResourceProfile(org.apache.flink.runtime.taskexecutor.TaskExecutorResourceSpec, int)> in (TaskManagerServices.java:481)
Method <org.apache.flink.runtime.taskexecutor.TaskManagerServices.createTaskSlotTable(int, org.apache.flink.runtime.taskexecutor.TaskExecutorResourceSpec, long, int, java.util.concurrent.Executor)> calls method <org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils.generateTotalAvailableResourceProfile(org.apache.flink.runtime.taskexecutor.TaskExecutorResourceSpec)> in (TaskManagerServices.java:479)
-Method <org.apache.flink.streaming.api.operators.SourceOperator$1$1.asClassLoader()> calls method <org.apache.flink.streaming.api.operators.SourceOperator.getRuntimeContext()> in (SourceOperator.java:310)
-Method <org.apache.flink.streaming.api.operators.SourceOperator$1$1.registerReleaseHookIfAbsent(java.lang.String, java.lang.Runnable)> calls method <org.apache.flink.streaming.api.operators.SourceOperator.getRuntimeContext()> in (SourceOperator.java:316)
+Method <org.apache.flink.streaming.api.operators.SourceOperator$1$1.asClassLoader()> calls method <org.apache.flink.streaming.api.operators.SourceOperator.getRuntimeContext()> in (SourceOperator.java:311)
+Method <org.apache.flink.streaming.api.operators.SourceOperator$1$1.registerReleaseHookIfAbsent(java.lang.String, java.lang.Runnable)> calls method <org.apache.flink.streaming.api.operators.SourceOperator.getRuntimeContext()> in (SourceOperator.java:317)
Method <org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask.init()> calls method <org.apache.flink.streaming.api.operators.SourceOperator.getSourceReader()> in (SourceOperatorStreamTask.java:96)
Method <org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl.isIdle()> calls method <org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.isDefaultActionAvailable()> in (MailboxExecutorImpl.java:63)
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinator.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinator.java
index c7ab72fcd7c..c1546548321 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinator.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinator.java
@@ -34,6 +34,7 @@ import org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWr
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -77,8 +78,10 @@ public class CompactCoordinator extends AbstractStreamOperator<CompactorRequest>
private ListState<FileSinkCommittable> remainingCommittableState;
public CompactCoordinator(
+ StreamOperatorParameters<CompactorRequest> parameters,
FileCompactStrategy strategy,
SimpleVersionedSerializer<FileSinkCommittable> committableSerializer) {
+ super(parameters);
this.strategy = strategy;
this.committableSerializer = checkNotNull(committableSerializer);
}
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinatorFactory.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinatorFactory.java
index f06b09eae84..a4440991c73 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinatorFactory.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinatorFactory.java
@@ -56,11 +56,8 @@ public class CompactCoordinatorFactory extends AbstractStreamOperatorFactory<Com
StreamOperatorParameters<CompactorRequest> parameters) {
try {
final CompactCoordinator compactOperator =
- new CompactCoordinator(strategy, committableSerializerSupplier.get());
- compactOperator.setup(
- parameters.getContainingTask(),
- parameters.getStreamConfig(),
- parameters.getOutput());
+ new CompactCoordinator(
+ parameters, strategy, committableSerializerSupplier.get());
return (T) compactOperator;
} catch (Exception e) {
throw new IllegalStateException(
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinatorStateHandler.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinatorStateHandler.java
index 1e4d0f969c0..9244e8ee985 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinatorStateHandler.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinatorStateHandler.java
@@ -30,6 +30,7 @@ import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.types.Either;
@@ -50,7 +51,11 @@ public class CompactCoordinatorStateHandler
private final SimpleVersionedSerializer<FileSinkCommittable> committableSerializer;
public CompactCoordinatorStateHandler(
+ StreamOperatorParameters<
+ Either<CommittableMessage<FileSinkCommittable>, CompactorRequest>>
+ parameters,
SimpleVersionedSerializer<FileSinkCommittable> committableSerializer) {
+ super(parameters);
this.committableSerializer = checkNotNull(committableSerializer);
}
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinatorStateHandlerFactory.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinatorStateHandlerFactory.java
index d4ebe11dcf0..df77d554ee6 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinatorStateHandlerFactory.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactCoordinatorStateHandlerFactory.java
@@ -66,11 +66,8 @@ public class CompactCoordinatorStateHandlerFactory
parameters) {
try {
final CompactCoordinatorStateHandler handler =
- new CompactCoordinatorStateHandler(committableSerializerSupplier.get());
- handler.setup(
- parameters.getContainingTask(),
- parameters.getStreamConfig(),
- parameters.getOutput());
+ new CompactCoordinatorStateHandler(
+ parameters, committableSerializerSupplier.get());
return (T) handler;
} catch (Exception e) {
throw new IllegalStateException(
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperator.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperator.java
index 0cb573d466f..fa4e2b3d54b 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperator.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperator.java
@@ -43,6 +43,7 @@ import org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -107,10 +108,12 @@ public class CompactorOperator
private ListState<Map<Long, List<CompactorRequest>>> remainingRequestsState;
public CompactorOperator(
+ StreamOperatorParameters<CommittableMessage<FileSinkCommittable>> parameters,
FileCompactStrategy strategy,
SimpleVersionedSerializer<FileSinkCommittable> committableSerializer,
FileCompactor fileCompactor,
BucketWriter<?, String> bucketWriter) {
+ super(parameters);
this.strategy = strategy;
this.committableSerializer = committableSerializer;
this.fileCompactor = fileCompactor;
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorFactory.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorFactory.java
index edbac271a62..0d42a16dda0 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorFactory.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorFactory.java
@@ -69,14 +69,11 @@ public class CompactorOperatorFactory
try {
final CompactorOperator compactOperator =
new CompactorOperator(
+ parameters,
strategy,
committableSerializerSupplier.get(),
fileCompactor,
bucketWriterProvider.get());
- compactOperator.setup(
- parameters.getContainingTask(),
- parameters.getStreamConfig(),
- parameters.getOutput());
return (T) compactOperator;
} catch (Exception e) {
throw new IllegalStateException(
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorStateHandler.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorStateHandler.java
index 02bcbbdcac7..3b406534c28 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorStateHandler.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorStateHandler.java
@@ -34,6 +34,7 @@ import org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.types.Either;
@@ -71,8 +72,10 @@ public class CompactorOperatorStateHandler
private Iterable<Map<Long, List<CompactorRequest>>> stateRemaining;
public CompactorOperatorStateHandler(
+ StreamOperatorParameters<CommittableMessage<FileSinkCommittable>> parameters,
SimpleVersionedSerializer<FileSinkCommittable> committableSerializer,
BucketWriter<?, String> bucketWriter) {
+ super(parameters);
this.committableSerializer = committableSerializer;
this.bucketWriter = bucketWriter;
}
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorStateHandlerFactory.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorStateHandlerFactory.java
index 2d36fee74e8..d21e49cf771 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorStateHandlerFactory.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorStateHandlerFactory.java
@@ -63,11 +63,9 @@ public class CompactorOperatorStateHandlerFactory
try {
final CompactorOperatorStateHandler handler =
new CompactorOperatorStateHandler(
- committableSerializerSupplier.get(), bucketWriterProvider.get());
- handler.setup(
- parameters.getContainingTask(),
- parameters.getStreamConfig(),
- parameters.getOutput());
+ parameters,
+ committableSerializerSupplier.get(),
+ bucketWriterProvider.get());
return (T) handler;
} catch (Exception e) {
throw new IllegalStateException(
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/batch/compact/BatchFileWriter.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/batch/compact/BatchFileWriter.java
index 87996e751e1..dd8e295e49b 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/batch/compact/BatchFileWriter.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/batch/compact/BatchFileWriter.java
@@ -31,7 +31,6 @@ import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.api.TableException;
@@ -75,7 +74,6 @@ public class BatchFileWriter<T> extends AbstractStreamOperator<CoordinatorInput>
this.formatFactory = formatFactory;
this.computer = computer;
this.outputFileConfig = outputFileConfig;
- setChainingStrategy(ChainingStrategy.ALWAYS);
}
@Override
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/AbstractStreamingWriter.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/AbstractStreamingWriter.java
index f1809f58561..9265ddf81ae 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/AbstractStreamingWriter.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/AbstractStreamingWriter.java
@@ -31,7 +31,6 @@ import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSin
import org.apache.flink.streaming.api.functions.sink.filesystem.legacy.StreamingFileSink;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -84,7 +83,6 @@ public abstract class AbstractStreamingWriter<IN, OUT> extends AbstractStreamOpe
bucketsBuilder) {
this.bucketCheckInterval = bucketCheckInterval;
this.bucketsBuilder = bucketsBuilder;
- setChainingStrategy(ChainingStrategy.ALWAYS);
}
/** Notifies a partition created. */
diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/compactor/CompactCoordinatorTest.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/compactor/CompactCoordinatorTest.java
index a1e9bb6d444..cab9f8c76c0 100644
--- a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/compactor/CompactCoordinatorTest.java
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/compactor/CompactCoordinatorTest.java
@@ -54,7 +54,7 @@ class CompactCoordinatorTest extends AbstractCompactTestBase {
void testSizeThreshold() throws Exception {
FileCompactStrategy strategy = Builder.newBuilder().setSizeThreshold(10).build();
CompactCoordinator coordinator =
- new CompactCoordinator(strategy, getTestCommittableSerializer());
+ new CompactCoordinator(null, strategy, getTestCommittableSerializer());
try (OneInputStreamOperatorTestHarness<
CommittableMessage<FileSinkCommittable>, CompactorRequest>
@@ -84,7 +84,7 @@ class CompactCoordinatorTest extends AbstractCompactTestBase {
void testCompactOnCheckpoint() throws Exception {
FileCompactStrategy strategy = Builder.newBuilder().enableCompactionOnCheckpoint(1).build();
CompactCoordinator coordinator =
- new CompactCoordinator(strategy, getTestCommittableSerializer());
+ new CompactCoordinator(null, strategy, getTestCommittableSerializer());
try (OneInputStreamOperatorTestHarness<
CommittableMessage<FileSinkCommittable>, CompactorRequest>
@@ -130,7 +130,7 @@ class CompactCoordinatorTest extends AbstractCompactTestBase {
void testCompactOverMultipleCheckpoints() throws Exception {
FileCompactStrategy strategy = Builder.newBuilder().enableCompactionOnCheckpoint(3).build();
CompactCoordinator coordinator =
- new CompactCoordinator(strategy, getTestCommittableSerializer());
+ new CompactCoordinator(null, strategy, getTestCommittableSerializer());
try (OneInputStreamOperatorTestHarness<
CommittableMessage<FileSinkCommittable>, CompactorRequest>
@@ -166,7 +166,7 @@ class CompactCoordinatorTest extends AbstractCompactTestBase {
void testCompactOnEndOfInput() throws Exception {
FileCompactStrategy strategy = Builder.newBuilder().setSizeThreshold(10).build();
CompactCoordinator coordinator =
- new CompactCoordinator(strategy, getTestCommittableSerializer());
+ new CompactCoordinator(null, strategy, getTestCommittableSerializer());
try (OneInputStreamOperatorTestHarness<
CommittableMessage<FileSinkCommittable>, CompactorRequest>
@@ -197,7 +197,7 @@ class CompactCoordinatorTest extends AbstractCompactTestBase {
void testPassthrough() throws Exception {
FileCompactStrategy strategy = Builder.newBuilder().setSizeThreshold(10).build();
CompactCoordinator coordinator =
- new CompactCoordinator(strategy, getTestCommittableSerializer());
+ new CompactCoordinator(null, strategy, getTestCommittableSerializer());
try (OneInputStreamOperatorTestHarness<
CommittableMessage<FileSinkCommittable>, CompactorRequest>
@@ -230,7 +230,7 @@ class CompactCoordinatorTest extends AbstractCompactTestBase {
void testRestore() throws Exception {
FileCompactStrategy strategy = Builder.newBuilder().setSizeThreshold(10).build();
CompactCoordinator coordinator =
- new CompactCoordinator(strategy, getTestCommittableSerializer());
+ new CompactCoordinator(null, strategy, getTestCommittableSerializer());
FileSinkCommittable committable0 = committable("0", ".0", 5);
FileSinkCommittable committable1 = committable("0", ".1", 6);
@@ -251,7 +251,7 @@ class CompactCoordinatorTest extends AbstractCompactTestBase {
state = harness.snapshot(1, 1);
}
- coordinator = new CompactCoordinator(strategy, getTestCommittableSerializer());
+ coordinator = new CompactCoordinator(null, strategy, getTestCommittableSerializer());
try (OneInputStreamOperatorTestHarness<
CommittableMessage<FileSinkCommittable>, CompactorRequest>
harness = new OneInputStreamOperatorTestHarness<>(coordinator)) {
@@ -282,7 +282,7 @@ class CompactCoordinatorTest extends AbstractCompactTestBase {
void testRestoreWithChangedStrategy() throws Exception {
FileCompactStrategy strategy = Builder.newBuilder().setSizeThreshold(100).build();
CompactCoordinator coordinator =
- new CompactCoordinator(strategy, getTestCommittableSerializer());
+ new CompactCoordinator(null, strategy, getTestCommittableSerializer());
FileSinkCommittable committable0 = committable("0", ".0", 5);
FileSinkCommittable committable1 = committable("0", ".1", 6);
@@ -312,7 +312,7 @@ class CompactCoordinatorTest extends AbstractCompactTestBase {
FileCompactStrategy changedStrategy = Builder.newBuilder().setSizeThreshold(10).build();
CompactCoordinator changedCoordinator =
- new CompactCoordinator(changedStrategy, getTestCommittableSerializer());
+ new CompactCoordinator(null, changedStrategy, getTestCommittableSerializer());
try (OneInputStreamOperatorTestHarness<
CommittableMessage<FileSinkCommittable>, CompactorRequest>
harness = new OneInputStreamOperatorTestHarness<>(changedCoordinator)) {
@@ -336,7 +336,7 @@ class CompactCoordinatorTest extends AbstractCompactTestBase {
void testStateHandler() throws Exception {
FileCompactStrategy strategy = Builder.newBuilder().setSizeThreshold(10).build();
CompactCoordinator coordinator =
- new CompactCoordinator(strategy, getTestCommittableSerializer());
+ new CompactCoordinator(null, strategy, getTestCommittableSerializer());
// with . prefix
FileSinkCommittable committable0 = committable("0", ".0", 5);
@@ -365,7 +365,7 @@ class CompactCoordinatorTest extends AbstractCompactTestBase {
}
CompactCoordinatorStateHandler handler =
- new CompactCoordinatorStateHandler(getTestCommittableSerializer());
+ new CompactCoordinatorStateHandler(null, getTestCommittableSerializer());
try (OneInputStreamOperatorTestHarness<
CommittableMessage<FileSinkCommittable>,
Either<CommittableMessage<FileSinkCommittable>, CompactorRequest>>
diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/compactor/CompactorOperatorTest.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/compactor/CompactorOperatorTest.java
index 51be721418e..3199ddd2ba6 100644
--- a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/compactor/CompactorOperatorTest.java
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/compactor/CompactorOperatorTest.java
@@ -263,7 +263,7 @@ class CompactorOperatorTest extends AbstractCompactTestBase {
CompactorOperatorStateHandler handler =
new CompactorOperatorStateHandler(
- getTestCommittableSerializer(), createTestBucketWriter());
+ null, getTestCommittableSerializer(), createTestBucketWriter());
try (OneInputStreamOperatorTestHarness<
Either<CommittableMessage<FileSinkCommittable>, CompactorRequest>,
CommittableMessage<FileSinkCommittable>>
@@ -363,6 +363,7 @@ class CompactorOperatorTest extends AbstractCompactTestBase {
harness =
new OneInputStreamOperatorTestHarness<>(
new CompactorOperatorStateHandler(
+ null,
getTestCommittableSerializer(),
createTestBucketWriter()))) {
harness.setup();
@@ -402,6 +403,7 @@ class CompactorOperatorTest extends AbstractCompactTestBase {
harness =
new OneInputStreamOperatorTestHarness<>(
new CompactorOperatorStateHandler(
+ null,
getTestCommittableSerializer(),
createTestBucketWriter()))) {
harness.setup();
@@ -436,6 +438,7 @@ class CompactorOperatorTest extends AbstractCompactTestBase {
harness =
new OneInputStreamOperatorTestHarness<>(
new CompactorOperatorStateHandler(
+ null,
getTestCommittableSerializer(),
createTestBucketWriter()))) {
harness.setup();
@@ -497,6 +500,7 @@ class CompactorOperatorTest extends AbstractCompactTestBase {
private CompactorOperator createTestOperator(FileCompactor compactor) {
return new CompactorOperator(
+ null,
FileCompactStrategy.Builder.newBuilder()
.setNumCompactThreads(2)
.enableCompactionOnCheckpoint(1)
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/ProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/ProcessOperator.java
index 2511b5db740..5f35a570214 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/ProcessOperator.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/ProcessOperator.java
@@ -30,7 +30,6 @@ import org.apache.flink.datastream.impl.context.DefaultRuntimeContext;
import org.apache.flink.datastream.impl.context.UnsupportedProcessingTimeManager;
import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -50,8 +49,6 @@ public class ProcessOperator<IN, OUT>
public ProcessOperator(OneInputStreamProcessFunction<IN, OUT> userFunction) {
super(userFunction);
-
- chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputBroadcastProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputBroadcastProcessOperator.java
index a6e34a1daca..df3d7465ee4 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputBroadcastProcessOperator.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputBroadcastProcessOperator.java
@@ -30,7 +30,6 @@ import org.apache.flink.datastream.impl.context.DefaultRuntimeContext;
import org.apache.flink.datastream.impl.context.UnsupportedProcessingTimeManager;
import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedMultiInput;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -54,7 +53,6 @@ public class TwoInputBroadcastProcessOperator<IN1, IN2, OUT>
public TwoInputBroadcastProcessOperator(
TwoInputBroadcastStreamProcessFunction<IN1, IN2, OUT> userFunction) {
super(userFunction);
- chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputNonBroadcastProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputNonBroadcastProcessOperator.java
index 31cc7d48dab..9501aaacf08 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputNonBroadcastProcessOperator.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputNonBroadcastProcessOperator.java
@@ -31,7 +31,6 @@ import org.apache.flink.datastream.impl.context.UnsupportedProcessingTimeManager
import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedMultiInput;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -55,7 +54,6 @@ public class TwoInputNonBroadcastProcessOperator<IN1, IN2, OUT>
public TwoInputNonBroadcastProcessOperator(
TwoInputNonBroadcastStreamProcessFunction<IN1, IN2, OUT> userFunction) {
super(userFunction);
- chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoOutputProcessOperator.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoOutputProcessOperator.java
index af24d1839cd..bae23887c24 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoOutputProcessOperator.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoOutputProcessOperator.java
@@ -31,7 +31,6 @@ import org.apache.flink.datastream.impl.context.DefaultTwoOutputNonPartitionedCo
import org.apache.flink.datastream.impl.context.UnsupportedProcessingTimeManager;
import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.Output;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
@@ -65,7 +64,6 @@ public class TwoOutputProcessOperator<IN, OUT_MAIN, OUT_SIDE>
super(userFunction);
this.outputTag = outputTag;
- chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java
index 49b084925c6..afc7e99a95d 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java
@@ -166,7 +166,7 @@ public class CepOperator<IN, KEY, OUT>
}
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<OUT>> output) {
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/KeyedStateTransformation.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/KeyedStateTransformation.java
index 8bcef815b0c..68228b96e45 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/KeyedStateTransformation.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/KeyedStateTransformation.java
@@ -24,6 +24,7 @@ import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.state.api.functions.KeyedStateBootstrapFunction;
import org.apache.flink.state.api.output.operators.KeyedStateBootstrapOperator;
import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.windows.Window;
@@ -77,7 +78,9 @@ public class KeyedStateTransformation<K, T> {
KeyedStateBootstrapFunction<K, T> processFunction) {
SavepointWriterOperatorFactory factory =
(timestamp, path) ->
- new KeyedStateBootstrapOperator<>(timestamp, path, processFunction);
+ SimpleOperatorFactory.of(
+ new KeyedStateBootstrapOperator<>(
+ timestamp, path, processFunction));
return transform(factory);
}
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/OneInputStateTransformation.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/OneInputStateTransformation.java
index ce9d3977005..a1e1191a148 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/OneInputStateTransformation.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/OneInputStateTransformation.java
@@ -31,6 +31,7 @@ import org.apache.flink.state.api.functions.StateBootstrapFunction;
import org.apache.flink.state.api.output.operators.BroadcastStateBootstrapOperator;
import org.apache.flink.state.api.output.operators.StateBootstrapOperator;
import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
import org.apache.flink.streaming.util.keys.KeySelectorUtil;
import java.util.OptionalInt;
@@ -81,7 +82,9 @@ public class OneInputStateTransformation<T> {
*/
public StateBootstrapTransformation<T> transform(StateBootstrapFunction<T> processFunction) {
SavepointWriterOperatorFactory factory =
- (timestamp, path) -> new StateBootstrapOperator<>(timestamp, path, processFunction);
+ (timestamp, path) ->
+ SimpleOperatorFactory.of(
+ new StateBootstrapOperator<>(timestamp, path, processFunction));
return transform(factory);
}
@@ -100,7 +103,9 @@ public class OneInputStateTransformation<T> {
BroadcastStateBootstrapFunction<T> processFunction) {
SavepointWriterOperatorFactory factory =
(timestamp, path) ->
- new BroadcastStateBootstrapOperator<>(timestamp, path, processFunction);
+ SimpleOperatorFactory.of(
+ new BroadcastStateBootstrapOperator<>(
+ timestamp, path, processFunction));
return transform(factory);
}
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/SavepointWriterOperatorFactory.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/SavepointWriterOperatorFactory.java
index c5a0b415e04..a13295bc34e 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/SavepointWriterOperatorFactory.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/SavepointWriterOperatorFactory.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.core.fs.Path;
import org.apache.flink.state.api.output.TaggedOperatorSubtaskState;
import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
/** Creates a savepoint writing operator from a savepoint path. */
@PublicEvolving
@@ -35,6 +36,6 @@ public interface SavepointWriterOperatorFactory {
* @param savepointPath the path to write the savepoint to.
* @return a stream operator for writing the savepoint.
*/
- StreamOperator<TaggedOperatorSubtaskState> createOperator(
+ StreamOperatorFactory<TaggedOperatorSubtaskState> createOperator(
long savepointTimestamp, Path savepointPath);
}
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateBootstrapTransformation.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateBootstrapTransformation.java
index 6e6f55bcc62..fa59485a878 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateBootstrapTransformation.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateBootstrapTransformation.java
@@ -39,7 +39,9 @@ import org.apache.flink.state.api.runtime.MutableConfig;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.SimpleUdfStreamOperatorFactory;
import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
@@ -164,7 +166,7 @@ public class StateBootstrapTransformation<T> {
Configuration additionalConfig,
Path savepointPath,
int localMaxParallelism) {
- StreamOperator<TaggedOperatorSubtaskState> operator =
+ StreamOperatorFactory<TaggedOperatorSubtaskState> operator =
factory.createOperator(System.currentTimeMillis(), savepointPath);
operator = stream.getExecutionEnvironment().clean(operator);
@@ -187,7 +189,9 @@ public class StateBootstrapTransformation<T> {
operatorRunner)
.setMaxParallelism(localMaxParallelism);
- if (operator instanceof BroadcastStateBootstrapOperator) {
+ if (operator instanceof SimpleUdfStreamOperatorFactory
+ && ((SimpleUdfStreamOperatorFactory<?>) operator).getOperator()
+ instanceof BroadcastStateBootstrapOperator) {
subtaskStates = subtaskStates.setParallelism(1);
} else {
int currentParallelism = getParallelism(subtaskStates);
@@ -203,7 +207,7 @@ public class StateBootstrapTransformation<T> {
OperatorIdentifier operatorIdentifier,
StateBackend stateBackend,
Configuration additionalConfig,
- StreamOperator<TaggedOperatorSubtaskState> operator) {
+ StreamOperatorFactory<TaggedOperatorSubtaskState> operator) {
// Eagerly perform a deep copy of the configuration, otherwise it will result in undefined
// behavior when deploying with multiple bootstrap transformations.
Configuration deepCopy =
@@ -225,7 +229,7 @@ public class StateBootstrapTransformation<T> {
config.setStatePartitioner(0, keySelector);
}
- config.setStreamOperator(operator);
+ config.setStreamOperatorFactory(operator);
config.setOperatorName(operatorIdentifier.getOperatorId().toHexString());
config.setOperatorID(operatorIdentifier.getOperatorId());
config.setStateBackend(stateBackend);
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/WindowedStateTransformation.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/WindowedStateTransformation.java
index 5d42ba7811c..6a59024d487 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/WindowedStateTransformation.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/WindowedStateTransformation.java
@@ -26,7 +26,7 @@ import org.apache.flink.api.common.functions.RichFunction;
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.state.api.output.operators.StateBootstrapWrapperOperator;
+import org.apache.flink.state.api.output.operators.StateBootstrapWrapperOperatorFactory;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.windowing.PassThroughWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
@@ -156,7 +156,8 @@ public class WindowedStateTransformation<T, K, W extends Window> {
WindowOperator<K, T, ?, R, W> operator = builder.reduce(reduceFunction, function);
SavepointWriterOperatorFactory factory =
- (timestamp, path) -> new StateBootstrapWrapperOperator<>(timestamp, path, operator);
+ (timestamp, path) ->
+ new StateBootstrapWrapperOperatorFactory<>(timestamp, path, operator);
return new StateBootstrapTransformation<>(
input, operatorMaxParallelism, factory, keySelector, keyType);
}
@@ -182,7 +183,8 @@ public class WindowedStateTransformation<T, K, W extends Window> {
WindowOperator<K, T, ?, R, W> operator = builder.reduce(reduceFunction, function);
SavepointWriterOperatorFactory factory =
- (timestamp, path) -> new StateBootstrapWrapperOperator<>(timestamp, path, operator);
+ (timestamp, path) ->
+ new StateBootstrapWrapperOperatorFactory<>(timestamp, path, operator);
return new StateBootstrapTransformation<>(
input, operatorMaxParallelism, factory, keySelector, keyType);
}
@@ -317,7 +319,8 @@ public class WindowedStateTransformation<T, K, W extends Window> {
builder.aggregate(aggregateFunction, windowFunction, accumulatorType);
SavepointWriterOperatorFactory factory =
- (timestamp, path) -> new StateBootstrapWrapperOperator<>(timestamp, path, operator);
+ (timestamp, path) ->
+ new StateBootstrapWrapperOperatorFactory<>(timestamp, path, operator);
return new StateBootstrapTransformation<>(
input, operatorMaxParallelism, factory, keySelector, keyType);
}
@@ -394,7 +397,8 @@ public class WindowedStateTransformation<T, K, W extends Window> {
builder.aggregate(aggregateFunction, windowFunction, accumulatorType);
SavepointWriterOperatorFactory factory =
- (timestamp, path) -> new StateBootstrapWrapperOperator<>(timestamp, path, operator);
+ (timestamp, path) ->
+ new StateBootstrapWrapperOperatorFactory<>(timestamp, path, operator);
return new StateBootstrapTransformation<>(
input, operatorMaxParallelism, factory, keySelector, keyType);
}
@@ -418,7 +422,8 @@ public class WindowedStateTransformation<T, K, W extends Window> {
WindowOperator<K, T, ?, R, W> operator = builder.apply(function);
SavepointWriterOperatorFactory factory =
- (timestamp, path) -> new StateBootstrapWrapperOperator<>(timestamp, path, operator);
+ (timestamp, path) ->
+ new StateBootstrapWrapperOperatorFactory<>(timestamp, path, operator);
return new StateBootstrapTransformation<>(
input, operatorMaxParallelism, factory, keySelector, keyType);
}
@@ -442,7 +447,8 @@ public class WindowedStateTransformation<T, K, W extends Window> {
WindowOperator<K, T, ?, R, W> operator = builder.apply(function);
SavepointWriterOperatorFactory factory =
- (timestamp, path) -> new StateBootstrapWrapperOperator<>(timestamp, path, operator);
+ (timestamp, path) ->
+ new StateBootstrapWrapperOperatorFactory<>(timestamp, path, operator);
return new StateBootstrapTransformation<>(
input, operatorMaxParallelism, factory, keySelector, keyType);
}
@@ -463,7 +469,8 @@ public class WindowedStateTransformation<T, K, W extends Window> {
WindowOperator<K, T, ?, R, W> operator = builder.process(function);
SavepointWriterOperatorFactory factory =
- (timestamp, path) -> new StateBootstrapWrapperOperator<>(timestamp, path, operator);
+ (timestamp, path) ->
+ new StateBootstrapWrapperOperatorFactory<>(timestamp, path, operator);
return new StateBootstrapTransformation<>(
input, operatorMaxParallelism, factory, keySelector, keyType);
}
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/operators/GroupReduceOperator.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/operators/GroupReduceOperator.java
index cc919f2391b..8be8aed843a 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/operators/GroupReduceOperator.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/operators/GroupReduceOperator.java
@@ -21,7 +21,6 @@ package org.apache.flink.state.api.output.operators;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.TimestampedCollector;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -44,7 +43,6 @@ public class GroupReduceOperator<IN, OUT>
public GroupReduceOperator(GroupReduceFunction<IN, OUT> userFunction) {
super(userFunction);
- super.chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/operators/StateBootstrapWrapperOperator.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/operators/StateBootstrapWrapperOperator.java
index ccbf588b81c..7d698e9d391 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/operators/StateBootstrapWrapperOperator.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/operators/StateBootstrapWrapperOperator.java
@@ -30,13 +30,12 @@ import org.apache.flink.state.api.runtime.NeverFireProcessingTimeService;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
import org.apache.flink.streaming.api.operators.Output;
-import org.apache.flink.streaming.api.operators.SetupableStreamOperator;
import org.apache.flink.streaming.api.operators.StreamTaskStateInitializer;
import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.operators.windowing.WindowOperator;
import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -49,9 +48,7 @@ import org.apache.flink.util.OutputTag;
@SuppressWarnings({"unchecked", "deprecation", "rawtypes"})
public final class StateBootstrapWrapperOperator<
IN, OUT, OP extends AbstractStreamOperator<OUT> & OneInputStreamOperator<IN, OUT>>
- implements OneInputStreamOperator<IN, TaggedOperatorSubtaskState>,
- SetupableStreamOperator<TaggedOperatorSubtaskState>,
- BoundedOneInput {
+ implements OneInputStreamOperator<IN, TaggedOperatorSubtaskState>, BoundedOneInput {
private static final long serialVersionUID = 1L;
@@ -61,21 +58,20 @@ public final class StateBootstrapWrapperOperator<
private Output<StreamRecord<TaggedOperatorSubtaskState>> output;
- private final OP operator;
-
- public StateBootstrapWrapperOperator(long timestamp, Path savepointPath, OP operator) {
+ private final WindowOperator<?, IN, ?, ?, ?> operator;
+ public StateBootstrapWrapperOperator(
+ long timestamp, Path savepointPath, WindowOperator<?, IN, ?, ?, ?> operator) {
this.timestamp = timestamp;
this.savepointPath = savepointPath;
this.operator = operator;
}
- @Override
public void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<TaggedOperatorSubtaskState>> output) {
- ((SetupableStreamOperator) operator).setup(containingTask, config, new VoidOutput<>());
+ operator.setup(containingTask, config, new VoidOutput<>());
operator.setProcessingTimeService(new NeverFireProcessingTimeService());
this.output = output;
}
@@ -146,16 +142,6 @@ public final class StateBootstrapWrapperOperator<
operator.setKeyContextElement2(record);
}
- @Override
- public ChainingStrategy getChainingStrategy() {
- return operator.getChainingStrategy();
- }
-
- @Override
- public void setChainingStrategy(ChainingStrategy strategy) {
- operator.setChainingStrategy(strategy);
- }
-
@Override
public OperatorMetricGroup getMetricGroup() {
return operator.getMetricGroup();
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/operators/StateBootstrapWrapperOperatorFactory.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/operators/StateBootstrapWrapperOperatorFactory.java
new file mode 100644
index 00000000000..91062e7f518
--- /dev/null
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/operators/StateBootstrapWrapperOperatorFactory.java
@@ -0,0 +1,105 @@
+/*
+ * 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.state.api.output.operators;
+
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.state.api.output.TaggedOperatorSubtaskState;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.operators.windowing.WindowOperator;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+import org.apache.flink.util.OutputTag;
+
+/**
+ * {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * StateBootstrapWrapperOperator}.
+ */
+public class StateBootstrapWrapperOperatorFactory<
+ IN,
+ OUT,
+ OP extends AbstractStreamOperator<OUT> & OneInputStreamOperator<IN, OUT>,
+ OPF extends OneInputStreamOperatorFactory<IN, OUT>>
+ extends AbstractStreamOperatorFactory<TaggedOperatorSubtaskState> {
+
+ private final long timestamp;
+
+ private final Path savepointPath;
+
+ private Output<StreamRecord<TaggedOperatorSubtaskState>> output;
+
+ private final WindowOperator<?, IN, ?, ?, ?> operator;
+
+ public StateBootstrapWrapperOperatorFactory(
+ long timestamp, Path savepointPath, WindowOperator<?, IN, ?, ?, ?> operator) {
+ this.timestamp = timestamp;
+ this.savepointPath = savepointPath;
+ this.operator = operator;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends StreamOperator<TaggedOperatorSubtaskState>> T createStreamOperator(
+ StreamOperatorParameters<TaggedOperatorSubtaskState> parameters) {
+ StateBootstrapWrapperOperator<IN, OUT, OP> wrapperOperator =
+ new StateBootstrapWrapperOperator<>(timestamp, savepointPath, operator);
+ wrapperOperator.setup(
+ parameters.getContainingTask(),
+ parameters.getStreamConfig(),
+ parameters.getOutput());
+ return (T) wrapperOperator;
+ }
+
+ @Override
+ public Class<? extends StreamOperator> getStreamOperatorClass(ClassLoader classLoader) {
+ return StateBootstrapWrapperOperator.class;
+ }
+
+ private static class VoidOutput<T> implements Output<T> {
+
+ @Override
+ public void emitWatermark(Watermark mark) {}
+
+ @Override
+ public void emitWatermarkStatus(WatermarkStatus watermarkStatus) {}
+
+ @Override
+ public <X> void collect(OutputTag<X> outputTag, StreamRecord<X> record) {}
+
+ @Override
+ public void emitLatencyMarker(LatencyMarker latencyMarker) {}
+
+ @Override
+ public void emitRecordAttributes(RecordAttributes recordAttributes) {}
+
+ @Override
+ public void collect(T record) {}
+
+ @Override
+ public void close() {}
+ }
+}
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
index abd92e9eac6..1a6e55f62f3 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
@@ -24,7 +24,6 @@ import org.apache.flink.python.env.PythonEnvironmentManager;
import org.apache.flink.python.metric.process.FlinkMetricContainer;
import org.apache.flink.runtime.state.KeyedStateBackend;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.InternalTimeServiceManager;
import org.apache.flink.streaming.api.operators.sorted.state.BatchExecutionInternalTimeServiceManager;
import org.apache.flink.streaming.api.operators.sorted.state.BatchExecutionKeyedStateBackend;
@@ -74,7 +73,6 @@ public abstract class AbstractPythonFunctionOperator<OUT> extends AbstractStream
public AbstractPythonFunctionOperator(Configuration config) {
this.config = Preconditions.checkNotNull(config);
- this.chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateUdfStreamOperator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateUdfStreamOperator.java
index 783805440cc..f8dd9ad235b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateUdfStreamOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateUdfStreamOperator.java
@@ -80,7 +80,7 @@ public abstract class AbstractAsyncStateUdfStreamOperator<OUT, F extends Functio
// ------------------------------------------------------------------------
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<OUT>> output) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperator.java
index a0e750a8673..79f97aab9e2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperator.java
@@ -81,7 +81,7 @@ class GlobalCommitterOperator<CommT, GlobalCommT> extends AbstractStreamOperator
}
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<Void>> output) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java
index af2d1eb931f..0a21a4451db 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java
@@ -41,14 +41,14 @@ import org.apache.flink.streaming.api.functions.windowing.PassThroughAllWindowFu
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ReduceApplyAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ReduceApplyProcessAllWindowFunction;
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
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.evictors.Evictor;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.windows.Window;
-import org.apache.flink.streaming.runtime.operators.windowing.EvictingWindowOperator;
-import org.apache.flink.streaming.runtime.operators.windowing.WindowOperator;
+import org.apache.flink.streaming.runtime.operators.windowing.EvictingWindowOperatorFactory;
+import org.apache.flink.streaming.runtime.operators.windowing.WindowOperatorFactory;
import org.apache.flink.streaming.runtime.operators.windowing.functions.InternalAggregateProcessAllWindowFunction;
import org.apache.flink.streaming.runtime.operators.windowing.functions.InternalIterableAllWindowFunction;
import org.apache.flink.streaming.runtime.operators.windowing.functions.InternalIterableProcessAllWindowFunction;
@@ -264,7 +264,7 @@ public class AllWindowedStream<T, W extends Window> {
String opDescription;
KeySelector<T, Byte> keySel = input.getKeySelector();
- OneInputStreamOperator<T, R> operator;
+ OneInputStreamOperatorFactory<T, R> operator;
if (evictor != null) {
@SuppressWarnings({"unchecked", "rawtypes"})
@@ -294,7 +294,7 @@ public class AllWindowedStream<T, W extends Window> {
+ ")";
operator =
- new EvictingWindowOperator<>(
+ new EvictingWindowOperatorFactory<>(
windowAssigner,
windowAssigner.getWindowSerializer(
getExecutionEnvironment().getConfig()),
@@ -335,7 +335,7 @@ public class AllWindowedStream<T, W extends Window> {
+ ")";
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
windowAssigner,
windowAssigner.getWindowSerializer(
getExecutionEnvironment().getConfig()),
@@ -410,7 +410,7 @@ public class AllWindowedStream<T, W extends Window> {
String opName;
KeySelector<T, Byte> keySel = input.getKeySelector();
- OneInputStreamOperator<T, R> operator;
+ OneInputStreamOperatorFactory<T, R> operator;
if (evictor != null) {
@SuppressWarnings({"unchecked", "rawtypes"})
@@ -440,7 +440,7 @@ public class AllWindowedStream<T, W extends Window> {
+ ")";
operator =
- new EvictingWindowOperator<>(
+ new EvictingWindowOperatorFactory<>(
windowAssigner,
windowAssigner.getWindowSerializer(
getExecutionEnvironment().getConfig()),
@@ -482,7 +482,7 @@ public class AllWindowedStream<T, W extends Window> {
+ ")";
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
windowAssigner,
windowAssigner.getWindowSerializer(
getExecutionEnvironment().getConfig()),
@@ -670,7 +670,7 @@ public class AllWindowedStream<T, W extends Window> {
final String opName;
final KeySelector<T, Byte> keySel = input.getKeySelector();
- OneInputStreamOperator<T, R> operator;
+ OneInputStreamOperatorFactory<T, R> operator;
if (evictor != null) {
@SuppressWarnings({"unchecked", "rawtypes"})
@@ -700,7 +700,7 @@ public class AllWindowedStream<T, W extends Window> {
+ ")";
operator =
- new EvictingWindowOperator<>(
+ new EvictingWindowOperatorFactory<>(
windowAssigner,
windowAssigner.getWindowSerializer(
getExecutionEnvironment().getConfig()),
@@ -739,7 +739,7 @@ public class AllWindowedStream<T, W extends Window> {
+ ")";
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
windowAssigner,
windowAssigner.getWindowSerializer(
getExecutionEnvironment().getConfig()),
@@ -846,7 +846,7 @@ public class AllWindowedStream<T, W extends Window> {
final String opDescription;
final KeySelector<T, Byte> keySel = input.getKeySelector();
- OneInputStreamOperator<T, R> operator;
+ OneInputStreamOperatorFactory<T, R> operator;
if (evictor != null) {
@SuppressWarnings({"unchecked", "rawtypes"})
@@ -876,7 +876,7 @@ public class AllWindowedStream<T, W extends Window> {
+ ")";
operator =
- new EvictingWindowOperator<>(
+ new EvictingWindowOperatorFactory<>(
windowAssigner,
windowAssigner.getWindowSerializer(
getExecutionEnvironment().getConfig()),
@@ -914,7 +914,7 @@ public class AllWindowedStream<T, W extends Window> {
+ ")";
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
windowAssigner,
windowAssigner.getWindowSerializer(
getExecutionEnvironment().getConfig()),
@@ -1027,7 +1027,7 @@ public class AllWindowedStream<T, W extends Window> {
String opName;
KeySelector<T, Byte> keySel = input.getKeySelector();
- WindowOperator<Byte, T, Iterable<T>, R, W> operator;
+ WindowOperatorFactory<Byte, T, Iterable<T>, R, W> operator;
if (evictor != null) {
@SuppressWarnings({"unchecked", "rawtypes"})
@@ -1057,7 +1057,7 @@ public class AllWindowedStream<T, W extends Window> {
+ ")";
operator =
- new EvictingWindowOperator<>(
+ new EvictingWindowOperatorFactory<>(
windowAssigner,
windowAssigner.getWindowSerializer(
getExecutionEnvironment().getConfig()),
@@ -1096,7 +1096,7 @@ public class AllWindowedStream<T, W extends Window> {
+ ")";
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
windowAssigner,
windowAssigner.getWindowSerializer(
getExecutionEnvironment().getConfig()),
@@ -1170,7 +1170,7 @@ public class AllWindowedStream<T, W extends Window> {
String opName;
KeySelector<T, Byte> keySel = input.getKeySelector();
- OneInputStreamOperator<T, R> operator;
+ OneInputStreamOperatorFactory<T, R> operator;
if (evictor != null) {
@SuppressWarnings({"unchecked", "rawtypes"})
@@ -1200,7 +1200,7 @@ public class AllWindowedStream<T, W extends Window> {
+ ")";
operator =
- new EvictingWindowOperator<>(
+ new EvictingWindowOperatorFactory<>(
windowAssigner,
windowAssigner.getWindowSerializer(
getExecutionEnvironment().getConfig()),
@@ -1241,7 +1241,7 @@ public class AllWindowedStream<T, W extends Window> {
+ ")";
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
windowAssigner,
windowAssigner.getWindowSerializer(
getExecutionEnvironment().getConfig()),
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamSource.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamSource.java
index 5e1a9ad59a1..709005677d3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamSource.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamSource.java
@@ -25,6 +25,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.Source;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.StreamSource;
import org.apache.flink.streaming.api.transformations.LegacySourceTransformation;
import org.apache.flink.streaming.api.transformations.SourceTransformation;
@@ -64,18 +65,32 @@ public class DataStreamSource<T> extends SingleOutputStreamOperator<T> {
Boundedness boundedness) {
super(
environment,
+ createSourceTransformation(
+ environment, outTypeInfo, operator, isParallel, sourceName, boundedness));
+
+ this.isParallel = isParallel;
+ if (!isParallel) {
+ setParallelism(1);
+ }
+ }
+
+ private static <T> LegacySourceTransformation<T> createSourceTransformation(
+ StreamExecutionEnvironment environment,
+ TypeInformation<T> outTypeInfo,
+ StreamSource<T, ?> operator,
+ boolean isParallel,
+ String sourceName,
+ Boundedness boundedness) {
+ LegacySourceTransformation<T> transformation =
new LegacySourceTransformation<>(
sourceName,
operator,
outTypeInfo,
environment.getParallelism(),
boundedness,
- false));
-
- this.isParallel = isParallel;
- if (!isParallel) {
- setParallelism(1);
- }
+ false);
+ transformation.setChainingStrategy(ChainingStrategy.HEAD);
+ return transformation;
}
/**
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/NonKeyedPartitionWindowedStream.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/NonKeyedPartitionWindowedStream.java
index 6efeacc7ebc..90a456ccdfa 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/NonKeyedPartitionWindowedStream.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/NonKeyedPartitionWindowedStream.java
@@ -65,7 +65,11 @@ public class NonKeyedPartitionWindowedStream<T> implements PartitionWindowedStre
TypeInformation<R> resultType =
TypeExtractor.getMapPartitionReturnTypes(
mapPartitionFunction, input.getType(), opName, true);
+
+ // This operator is set to be non-chained as it doesn't use task main thread to write
+ // records to output, which may introduce risks to downstream chained operators.
return input.transform(opName, resultType, new MapPartitionOperator<>(mapPartitionFunction))
+ .disableChaining()
.setParallelism(input.getParallelism());
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperator.java
index e9f559dfa60..02baed7602d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperator.java
@@ -37,6 +37,7 @@ import org.apache.flink.streaming.api.functions.source.legacy.SourceFunction;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.OutputTypeConfigurable;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.StreamSourceContexts;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -247,10 +248,11 @@ public class ContinuousFileReaderOperator<OUT, T extends TimestampedInputSplit>
};
ContinuousFileReaderOperator(
+ StreamOperatorParameters<OUT> parameters,
InputFormat<OUT, ? super T> format,
ProcessingTimeService processingTimeService,
MailboxExecutor mailboxExecutor) {
-
+ super(parameters);
this.format = checkNotNull(format);
this.processingTimeService = checkNotNull(processingTimeService);
this.executor = (MailboxExecutorImpl) checkNotNull(mailboxExecutor);
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperatorFactory.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperatorFactory.java
index aba67181908..7a87eb68606 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperatorFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperatorFactory.java
@@ -55,11 +55,7 @@ public class ContinuousFileReaderOperatorFactory<OUT, T extends TimestampedInput
StreamOperatorParameters<OUT> parameters) {
ContinuousFileReaderOperator<OUT, T> operator =
new ContinuousFileReaderOperator<>(
- inputFormat, processingTimeService, getMailboxExecutor());
- operator.setup(
- parameters.getContainingTask(),
- parameters.getStreamConfig(),
- parameters.getOutput());
+ parameters, inputFormat, processingTimeService, getMailboxExecutor());
operator.setOutputType(type, executionConfig);
return (O) operator;
}
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 3f39db26c63..6b298a9c978 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
@@ -95,7 +95,6 @@ import static org.apache.flink.util.Preconditions.checkState;
@PublicEvolving
public abstract class AbstractStreamOperator<OUT>
implements StreamOperator<OUT>,
- SetupableStreamOperator<OUT>,
YieldingOperator<OUT>,
CheckpointedStreamOperator,
KeyContextHandler,
@@ -105,11 +104,6 @@ public abstract class AbstractStreamOperator<OUT>
/** The logger used by the operator class and its subclasses. */
protected static final Logger LOG = LoggerFactory.getLogger(AbstractStreamOperator.class);
- // ----------- configuration properties -------------
-
- // A sane default for most operators
- protected ChainingStrategy chainingStrategy = ChainingStrategy.HEAD;
-
// ---------------- runtime fields ------------------
/** The task that contains this operator (and other operators in the same chain). */
@@ -164,12 +158,24 @@ public abstract class AbstractStreamOperator<OUT>
protected transient RecordAttributes lastRecordAttributes1;
protected transient RecordAttributes lastRecordAttributes2;
+ public AbstractStreamOperator() {}
+
+ public AbstractStreamOperator(StreamOperatorParameters<OUT> parameters) {
+ if (parameters != null) {
+ setup(
+ parameters.getContainingTask(),
+ parameters.getStreamConfig(),
+ parameters.getOutput());
+ this.processingTimeService =
+ Preconditions.checkNotNull(parameters.getProcessingTimeService());
+ }
+ }
+
// ------------------------------------------------------------------------
// Life Cycle
// ------------------------------------------------------------------------
- @Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<OUT>> output) {
@@ -246,12 +252,7 @@ public abstract class AbstractStreamOperator<OUT>
lastRecordAttributes2 = RecordAttributes.EMPTY_RECORD_ATTRIBUTES;
}
- /**
- * @deprecated The {@link ProcessingTimeService} instance should be passed by the operator
- * constructor and this method will be removed along with {@link SetupableStreamOperator}.
- */
- @Deprecated
- public void setProcessingTimeService(ProcessingTimeService processingTimeService) {
+ protected void setProcessingTimeService(ProcessingTimeService processingTimeService) {
this.processingTimeService = Preconditions.checkNotNull(processingTimeService);
}
@@ -591,16 +592,6 @@ public abstract class AbstractStreamOperator<OUT>
// Context and chaining properties
// ------------------------------------------------------------------------
- @Override
- public final void setChainingStrategy(ChainingStrategy strategy) {
- this.chainingStrategy = strategy;
- }
-
- @Override
- public final ChainingStrategy getChainingStrategy() {
- return chainingStrategy;
- }
-
// ------------------------------------------------------------------------
// Metrics
// ------------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
index 61660d151ac..656bc2ed941 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
@@ -57,6 +57,11 @@ public abstract class AbstractUdfStreamOperator<OUT, F extends Function>
protected final F userFunction;
public AbstractUdfStreamOperator(F userFunction) {
+ this(null, userFunction);
+ }
+
+ protected AbstractUdfStreamOperator(StreamOperatorParameters<OUT> parameters, F userFunction) {
+ super(parameters);
this.userFunction = requireNonNull(userFunction);
checkUdfCheckpointingPreconditions();
}
@@ -75,7 +80,7 @@ public abstract class AbstractUdfStreamOperator<OUT, F extends Function>
// ------------------------------------------------------------------------
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<OUT>> output) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/KeyedProcessOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/KeyedProcessOperator.java
index d4bfb5cb7c7..ad84ecdf291 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/KeyedProcessOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/KeyedProcessOperator.java
@@ -46,8 +46,6 @@ public class KeyedProcessOperator<K, IN, OUT>
public KeyedProcessOperator(KeyedProcessFunction<K, IN, OUT> function) {
super(function);
-
- chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/LegacyKeyedProcessOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/LegacyKeyedProcessOperator.java
index fd301e19cdc..fc26e880455 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/LegacyKeyedProcessOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/LegacyKeyedProcessOperator.java
@@ -51,8 +51,6 @@ public class LegacyKeyedProcessOperator<K, IN, OUT>
public LegacyKeyedProcessOperator(ProcessFunction<IN, OUT> function) {
super(function);
-
- chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/MapPartitionOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/MapPartitionOperator.java
index df814912da3..0e65e42d4ed 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/MapPartitionOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/MapPartitionOperator.java
@@ -39,9 +39,6 @@ public class MapPartitionOperator<IN, OUT>
public MapPartitionOperator(MapPartitionFunction<IN, OUT> function) {
super(function);
this.function = function;
- // This operator is set to be non-chained as it doesn't use task main thread to write
- // records to output, which may introduce risks to downstream chained operators.
- this.chainingStrategy = ChainingStrategy.NEVER;
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/ProcessOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/ProcessOperator.java
index f255497cfbc..8bf26b070a8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/ProcessOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/ProcessOperator.java
@@ -47,8 +47,6 @@ public class ProcessOperator<IN, OUT>
public ProcessOperator(ProcessFunction<IN, OUT> function) {
super(function);
-
- chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SimpleOperatorFactory.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SimpleOperatorFactory.java
index c2b3d23b3be..a20eb643099 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SimpleOperatorFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SimpleOperatorFactory.java
@@ -63,9 +63,6 @@ public class SimpleOperatorFactory<OUT> extends AbstractStreamOperatorFactory<OU
protected SimpleOperatorFactory(StreamOperator<OUT> operator) {
this.operator = checkNotNull(operator);
- if (operator instanceof SetupableStreamOperator) {
- this.chainingStrategy = ((SetupableStreamOperator) operator).getChainingStrategy();
- }
}
public StreamOperator<OUT> getOperator() {
@@ -77,10 +74,9 @@ public class SimpleOperatorFactory<OUT> extends AbstractStreamOperatorFactory<OU
public <T extends StreamOperator<OUT>> T createStreamOperator(
StreamOperatorParameters<OUT> parameters) {
if (operator instanceof AbstractStreamOperator) {
- ((AbstractStreamOperator) operator).setProcessingTimeService(processingTimeService);
- }
- if (operator instanceof SetupableStreamOperator) {
- ((SetupableStreamOperator) operator)
+ ((AbstractStreamOperator) operator)
+ .setProcessingTimeService(parameters.getProcessingTimeService());
+ ((AbstractStreamOperator) operator)
.setup(
parameters.getContainingTask(),
parameters.getStreamConfig(),
@@ -89,14 +85,6 @@ public class SimpleOperatorFactory<OUT> extends AbstractStreamOperatorFactory<OU
return (T) operator;
}
- @Override
- public void setChainingStrategy(ChainingStrategy strategy) {
- this.chainingStrategy = strategy;
- if (operator instanceof SetupableStreamOperator) {
- ((SetupableStreamOperator) operator).setChainingStrategy(strategy);
- }
- }
-
@Override
public boolean isStreamSource() {
return operator instanceof StreamSource;
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
index 3c7c1964a37..6491a055b1a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
@@ -206,6 +206,7 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
private transient PausableRelativeClock mainInputActivityClock;
public SourceOperator(
+ StreamOperatorParameters<OUT> parameters,
FunctionWithException<SourceReaderContext, SourceReader<OUT, SplitT>, Exception>
readerFactory,
OperatorEventGateway operatorEventGateway,
@@ -216,7 +217,7 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
String localHostname,
boolean emitProgressiveWatermarks,
CanEmitBatchOfRecordsChecker canEmitBatchOfRecords) {
-
+ super(parameters);
this.readerFactory = checkNotNull(readerFactory);
this.operatorEventGateway = checkNotNull(operatorEventGateway);
this.splitSerializer = checkNotNull(splitSerializer);
@@ -232,7 +233,7 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
}
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<OUT>> output) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperatorFactory.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperatorFactory.java
index 8ec3f3c9a9d..14c53fc6dd4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperatorFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperatorFactory.java
@@ -101,6 +101,7 @@ public class SourceOperatorFactory<OUT> extends AbstractStreamOperatorFactory<OU
final SourceOperator<OUT, ?> sourceOperator =
instantiateSourceOperator(
+ parameters,
source::createReader,
gateway,
source.getSplitSerializer(),
@@ -119,10 +120,6 @@ public class SourceOperatorFactory<OUT> extends AbstractStreamOperatorFactory<OU
emitProgressiveWatermarks,
parameters.getContainingTask().getCanEmitBatchOfRecords());
- sourceOperator.setup(
- parameters.getContainingTask(),
- parameters.getStreamConfig(),
- parameters.getOutput());
parameters.getOperatorEventDispatcher().registerEventHandler(operatorId, sourceOperator);
// today's lunch is generics spaghetti
@@ -177,6 +174,7 @@ public class SourceOperatorFactory<OUT> extends AbstractStreamOperatorFactory<OU
@SuppressWarnings("unchecked")
private static <T, SplitT extends SourceSplit>
SourceOperator<T, SplitT> instantiateSourceOperator(
+ StreamOperatorParameters<T> parameters,
FunctionWithException<SourceReaderContext, SourceReader<T, ?>, Exception>
readerFactory,
OperatorEventGateway eventGateway,
@@ -200,6 +198,7 @@ public class SourceOperatorFactory<OUT> extends AbstractStreamOperatorFactory<OU
(SimpleVersionedSerializer<SplitT>) splitSerializer;
return new SourceOperator<>(
+ parameters,
typedReaderFactory,
eventGateway,
typedSplitSerializer,
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamFilter.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamFilter.java
index 9bbe0690126..d31fa290cfa 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamFilter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamFilter.java
@@ -30,7 +30,6 @@ public class StreamFilter<IN> extends AbstractUdfStreamOperator<IN, FilterFuncti
public StreamFilter(FilterFunction<IN> filterFunction) {
super(filterFunction);
- chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java
index 82d0514c84f..f40d2b73281 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java
@@ -32,7 +32,6 @@ public class StreamFlatMap<IN, OUT> extends AbstractUdfStreamOperator<OUT, FlatM
public StreamFlatMap(FlatMapFunction<IN, OUT> flatMapper) {
super(flatMapper);
- chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamMap.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamMap.java
index 1e2697cdcb1..27b49a4145f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamMap.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamMap.java
@@ -30,7 +30,6 @@ public class StreamMap<IN, OUT> extends AbstractUdfStreamOperator<OUT, MapFuncti
public StreamMap(MapFunction<IN, OUT> mapper) {
super(mapper);
- chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SetupableStreamOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorUtils.java
similarity index 55%
rename from flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SetupableStreamOperator.java
rename to flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorUtils.java
index 5c59d4da984..347d6792f26 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SetupableStreamOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorUtils.java
@@ -17,28 +17,27 @@
package org.apache.flink.streaming.api.operators;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
import org.apache.flink.streaming.runtime.tasks.StreamTask;
-/**
- * Stream operators can implement this interface if they need access to the context and the output.
- *
- * @param <OUT> The output type of the operator
- * @deprecated This class is deprecated in favour of using {@link StreamOperatorFactory} and it's
- * {@link StreamOperatorFactory#createStreamOperator} and passing the required parameters to the
- * Operator's constructor in create method.
- */
-@Deprecated
-@PublicEvolving
-public interface SetupableStreamOperator<OUT> {
-
- /** Initializes the operator. Sets access to the context and the output. */
- void setup(
- StreamTask<?, ?> containingTask, StreamConfig config, Output<StreamRecord<OUT>> output);
-
- ChainingStrategy getChainingStrategy();
+@Internal
+public class StreamOperatorUtils {
+ @VisibleForTesting
+ public static <OUT> void setupStreamOperator(
+ AbstractStreamOperator<OUT> operator,
+ StreamTask<?, ?> containingTask,
+ StreamConfig config,
+ Output<StreamRecord<OUT>> output) {
+ operator.setup(containingTask, config, output);
+ }
- void setChainingStrategy(ChainingStrategy strategy);
+ @VisibleForTesting
+ public static void setProcessingTimeService(
+ AbstractStreamOperator<?> operator, ProcessingTimeService processingTimeService) {
+ operator.setProcessingTimeService(processingTimeService);
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamProject.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamProject.java
index a4665324344..eac3c678536 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamProject.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamProject.java
@@ -39,8 +39,6 @@ public class StreamProject<IN, OUT extends Tuple> extends AbstractStreamOperator
this.fields = fields;
this.numFields = this.fields.length;
this.outSerializer = outSerializer;
-
- chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java
index 7aa6c33c035..babbfa7344a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java
@@ -38,7 +38,6 @@ public class StreamSink<IN> extends AbstractUdfStreamOperator<Object, SinkFuncti
public StreamSink(SinkFunction<IN> sinkFunction) {
super(sinkFunction);
- chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
index 11c769531d9..d40f6e6f1c8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
@@ -51,7 +51,6 @@ public class StreamSource<OUT, SRC extends SourceFunction<OUT>>
public StreamSource(SRC sourceFunction, boolean emitProgressiveWatermarks) {
super(sourceFunction);
- this.chainingStrategy = ChainingStrategy.HEAD;
this.emitProgressiveWatermarks = emitProgressiveWatermarks;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkOperatorFactory.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkOperatorFactory.java
index aa0d4c21485..755bd09afa1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkOperatorFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkOperatorFactory.java
@@ -62,14 +62,11 @@ public class CollectSinkOperatorFactory<IN> extends SimpleUdfStreamOperatorFacto
@SuppressWarnings("unchecked")
public <T extends StreamOperator<Object>> T createStreamOperator(
StreamOperatorParameters<Object> parameters) {
+ CollectSinkOperator<IN> operator = super.createStreamOperator(parameters);
final OperatorID operatorId = parameters.getStreamConfig().getOperatorID();
final OperatorEventDispatcher eventDispatcher = parameters.getOperatorEventDispatcher();
operator.setOperatorEventGateway(eventDispatcher.getOperatorEventGateway(operatorId));
- operator.setup(
- parameters.getContainingTask(),
- parameters.getStreamConfig(),
- parameters.getOutput());
eventDispatcher.registerEventHandler(operatorId, operator);
return (T) operator;
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sortpartition/KeyedSortPartitionOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sortpartition/KeyedSortPartitionOperator.java
index 746a69285dc..e986efffee3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sortpartition/KeyedSortPartitionOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sortpartition/KeyedSortPartitionOperator.java
@@ -135,7 +135,7 @@ public class KeyedSortPartitionOperator<INPUT, KEY> extends AbstractStreamOperat
}
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<INPUT>> output) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sortpartition/SortPartitionOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sortpartition/SortPartitionOperator.java
index 6946baaf437..a361c7fd484 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sortpartition/SortPartitionOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sortpartition/SortPartitionOperator.java
@@ -124,7 +124,7 @@ public class SortPartitionOperator<INPUT> extends AbstractStreamOperator<INPUT>
}
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<INPUT>> output) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperator.java
index a636ea9ab82..3a06dba9579 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperator.java
@@ -26,9 +26,9 @@ import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.util.PausableRelativeClock;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
@@ -74,10 +74,12 @@ public class TimestampsAndWatermarksOperator<T> extends AbstractStreamOperator<T
private transient PausableRelativeClock inputActivityClock;
public TimestampsAndWatermarksOperator(
- WatermarkStrategy<T> watermarkStrategy, boolean emitProgressiveWatermarks) {
+ StreamOperatorParameters<T> parameters,
+ WatermarkStrategy<T> watermarkStrategy,
+ boolean emitProgressiveWatermarks) {
+ super(parameters);
this.watermarkStrategy = checkNotNull(watermarkStrategy);
this.emitProgressiveWatermarks = emitProgressiveWatermarks;
- this.chainingStrategy = ChainingStrategy.DEFAULT_CHAINING_STRATEGY;
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperatorFactory.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperatorFactory.java
new file mode 100644
index 00000000000..627fc7d9830
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperatorFactory.java
@@ -0,0 +1,60 @@
+/*
+ * 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.streaming.runtime.operators;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * TimestampsAndWatermarksOperator}.
+ */
+public class TimestampsAndWatermarksOperatorFactory<T> extends AbstractStreamOperatorFactory<T>
+ implements OneInputStreamOperatorFactory<T, T> {
+
+ private final WatermarkStrategy<T> watermarkStrategy;
+
+ /** Whether to emit intermediate watermarks or only one final watermark at the end of input. */
+ private final boolean emitProgressiveWatermarks;
+
+ public TimestampsAndWatermarksOperatorFactory(
+ WatermarkStrategy<T> watermarkStrategy, boolean emitProgressiveWatermarks) {
+ this.watermarkStrategy = checkNotNull(watermarkStrategy);
+ this.emitProgressiveWatermarks = emitProgressiveWatermarks;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T1 extends StreamOperator<T>> T1 createStreamOperator(
+ StreamOperatorParameters<T> parameters) {
+ return (T1)
+ new TimestampsAndWatermarksOperator<>(
+ parameters, watermarkStrategy, emitProgressiveWatermarks);
+ }
+
+ @Override
+ public Class<? extends StreamOperator> getStreamOperatorClass(ClassLoader classLoader) {
+ return TimestampsAndWatermarksOperator.class;
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
index f94c55df941..f0e717797de 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
@@ -34,6 +34,7 @@ import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.runtime.operators.sink.committables.CheckpointCommittableManager;
@@ -89,6 +90,7 @@ class CommitterOperator<CommT> extends AbstractStreamOperator<CommittableMessage
private ListState<CommittableCollector<CommT>> committableCollectorState;
public CommitterOperator(
+ StreamOperatorParameters<CommittableMessage<CommT>> parameters,
ProcessingTimeService processingTimeService,
SimpleVersionedSerializer<CommT> committableSerializer,
FunctionWithException<CommitterInitContext, Committer<CommT>, IOException>
@@ -96,6 +98,7 @@ class CommitterOperator<CommT> extends AbstractStreamOperator<CommittableMessage
boolean emitDownstream,
boolean isBatchMode,
boolean isCheckpointingEnabled) {
+ super(parameters);
this.emitDownstream = emitDownstream;
this.isBatchMode = isBatchMode;
this.isCheckpointingEnabled = isCheckpointingEnabled;
@@ -105,7 +108,7 @@ class CommitterOperator<CommT> extends AbstractStreamOperator<CommittableMessage
}
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<CommittableMessage<CommT>>> output) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorFactory.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorFactory.java
index 162f8d04181..08d2c2181ae 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorFactory.java
@@ -60,16 +60,13 @@ public final class CommitterOperatorFactory<CommT>
try {
final CommitterOperator<CommT> committerOperator =
new CommitterOperator<>(
+ parameters,
processingTimeService,
sink.getCommittableSerializer(),
context -> sink.createCommitter(context),
sink instanceof SupportsPostCommitTopology,
isBatchMode,
isCheckpointingEnabled);
- committerOperator.setup(
- parameters.getContainingTask(),
- parameters.getStreamConfig(),
- parameters.getOutput());
return (T) committerOperator;
} catch (Exception e) {
throw new IllegalStateException(
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
index 98702a463f0..2e89fdf2dfc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
@@ -45,6 +45,7 @@ import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.InternalTimerService;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.api.watermark.Watermark;
@@ -122,9 +123,11 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
@Nullable private ListState<Boolean> endOfInputState;
SinkWriterOperator(
+ StreamOperatorParameters<CommittableMessage<CommT>> parameters,
Sink<InputT> sink,
ProcessingTimeService processingTimeService,
MailboxExecutor mailboxExecutor) {
+ super(parameters);
this.processingTimeService = checkNotNull(processingTimeService);
this.mailboxExecutor = checkNotNull(mailboxExecutor);
this.context = new Context<>();
@@ -145,7 +148,7 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
}
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<CommittableMessage<CommT>>> output) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperatorFactory.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperatorFactory.java
index 4ad16c68f73..9d1798135ae 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperatorFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperatorFactory.java
@@ -54,11 +54,8 @@ public final class SinkWriterOperatorFactory<InputT, CommT>
StreamOperatorParameters<CommittableMessage<CommT>> parameters) {
try {
final SinkWriterOperator<InputT, CommT> writerOperator =
- new SinkWriterOperator<>(sink, processingTimeService, getMailboxExecutor());
- writerOperator.setup(
- parameters.getContainingTask(),
- parameters.getStreamConfig(),
- parameters.getOutput());
+ new SinkWriterOperator<>(
+ parameters, sink, processingTimeService, getMailboxExecutor());
return (T) writerOperator;
} catch (Exception e) {
throw new IllegalStateException(
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperatorFactory.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperatorFactory.java
new file mode 100644
index 00000000000..f0ec136eb7d
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperatorFactory.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.streaming.runtime.operators.windowing;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.state.AppendingState;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+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.streaming.runtime.operators.windowing.functions.InternalWindowFunction;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * EvictingWindowOperator}.
+ */
+@Internal
+public class EvictingWindowOperatorFactory<K, IN, OUT, W extends Window>
+ extends WindowOperatorFactory<K, IN, Iterable<IN>, OUT, W> {
+
+ private final Evictor<? super IN, ? super W> evictor;
+
+ private final StateDescriptor<? extends ListState<StreamRecord<IN>>, ?>
+ evictingWindowStateDescriptor;
+
+ public EvictingWindowOperatorFactory(
+ WindowAssigner<? super IN, W> windowAssigner,
+ TypeSerializer<W> windowSerializer,
+ KeySelector<IN, K> keySelector,
+ TypeSerializer<K> keySerializer,
+ StateDescriptor<? extends ListState<StreamRecord<IN>>, ?> windowStateDescriptor,
+ InternalWindowFunction<Iterable<IN>, OUT, K, W> windowFunction,
+ Trigger<? super IN, ? super W> trigger,
+ Evictor<? super IN, ? super W> evictor,
+ long allowedLateness,
+ OutputTag<IN> lateDataOutputTag) {
+ super(
+ windowAssigner,
+ windowSerializer,
+ keySelector,
+ keySerializer,
+ null,
+ windowFunction,
+ trigger,
+ allowedLateness,
+ lateDataOutputTag);
+
+ this.evictor = checkNotNull(evictor);
+ this.evictingWindowStateDescriptor = checkNotNull(windowStateDescriptor);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends StreamOperator<OUT>> T createStreamOperator(
+ StreamOperatorParameters<OUT> parameters) {
+ EvictingWindowOperator<K, IN, OUT, W> operator =
+ new EvictingWindowOperator<>(
+ windowAssigner,
+ windowSerializer,
+ keySelector,
+ keySerializer,
+ evictingWindowStateDescriptor,
+ windowFunction,
+ trigger,
+ evictor,
+ allowedLateness,
+ lateDataOutputTag);
+ operator.setup(
+ parameters.getContainingTask(),
+ parameters.getStreamConfig(),
+ parameters.getOutput());
+ operator.setProcessingTimeService(parameters.getProcessingTimeService());
+ return (T) operator;
+ }
+
+ @Override
+ public Class<? extends StreamOperator> getStreamOperatorClass(ClassLoader classLoader) {
+ return EvictingWindowOperator.class;
+ }
+
+ @Override
+ @VisibleForTesting
+ @SuppressWarnings("unchecked, rawtypes")
+ public StateDescriptor<? extends AppendingState<IN, Iterable<IN>>, ?> getStateDescriptor() {
+ return (StateDescriptor<? extends AppendingState<IN, Iterable<IN>>, ?>)
+ evictingWindowStateDescriptor;
+ }
+
+ @VisibleForTesting
+ public Evictor<? super IN, ? super W> getEvictor() {
+ return evictor;
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
index 35609ef9ac3..4bdc82c5e5b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
@@ -51,13 +51,14 @@ import org.apache.flink.runtime.state.VoidNamespaceSerializer;
import org.apache.flink.runtime.state.internal.InternalAppendingState;
import org.apache.flink.runtime.state.internal.InternalListState;
import org.apache.flink.runtime.state.internal.InternalMergingState;
+import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.InternalTimer;
import org.apache.flink.streaming.api.operators.InternalTimerService;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.OperatorAttributes;
import org.apache.flink.streaming.api.operators.OperatorAttributesBuilder;
+import org.apache.flink.streaming.api.operators.Output;
import org.apache.flink.streaming.api.operators.TimestampedCollector;
import org.apache.flink.streaming.api.operators.Triggerable;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
@@ -68,6 +69,8 @@ import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.streaming.runtime.operators.windowing.functions.InternalWindowFunction;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
import org.apache.flink.util.OutputTag;
import java.util.Collection;
@@ -204,8 +207,19 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
this.trigger = checkNotNull(trigger);
this.allowedLateness = allowedLateness;
this.lateDataOutputTag = lateDataOutputTag;
+ }
- setChainingStrategy(ChainingStrategy.ALWAYS);
+ @Override
+ public void setup(
+ StreamTask<?, ?> containingTask,
+ StreamConfig config,
+ Output<StreamRecord<OUT>> output) {
+ super.setup(containingTask, config, output);
+ }
+
+ @Override
+ public void setProcessingTimeService(ProcessingTimeService processingTimeService) {
+ super.setProcessingTimeService(processingTimeService);
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorFactory.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorFactory.java
new file mode 100644
index 00000000000..f61fafee481
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorFactory.java
@@ -0,0 +1,145 @@
+/*
+ * 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.streaming.runtime.operators.windowing;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.state.AppendingState;
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+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.streaming.runtime.operators.windowing.functions.InternalWindowFunction;
+import org.apache.flink.util.OutputTag;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * WindowOperator}.
+ */
+@Internal
+public class WindowOperatorFactory<K, IN, ACC, OUT, W extends Window>
+ extends AbstractStreamOperatorFactory<OUT>
+ implements OneInputStreamOperatorFactory<IN, OUT> {
+
+ protected final WindowAssigner<? super IN, W> windowAssigner;
+
+ protected final KeySelector<IN, K> keySelector;
+
+ protected final Trigger<? super IN, ? super W> trigger;
+
+ protected final StateDescriptor<? extends AppendingState<IN, ACC>, ?> windowStateDescriptor;
+
+ protected final InternalWindowFunction<ACC, OUT, K, W> windowFunction;
+
+ protected final TypeSerializer<K> keySerializer;
+
+ protected final TypeSerializer<W> windowSerializer;
+
+ protected final long allowedLateness;
+
+ protected final OutputTag<IN> lateDataOutputTag;
+
+ public WindowOperatorFactory(
+ WindowAssigner<? super IN, W> windowAssigner,
+ TypeSerializer<W> windowSerializer,
+ KeySelector<IN, K> keySelector,
+ TypeSerializer<K> keySerializer,
+ StateDescriptor<? extends AppendingState<IN, ACC>, ?> windowStateDescriptor,
+ InternalWindowFunction<ACC, OUT, K, W> windowFunction,
+ Trigger<? super IN, ? super W> trigger,
+ long allowedLateness,
+ OutputTag<IN> lateDataOutputTag) {
+ checkArgument(allowedLateness >= 0);
+
+ checkArgument(
+ windowStateDescriptor == null || windowStateDescriptor.isSerializerInitialized(),
+ "window state serializer is not properly initialized");
+
+ this.windowAssigner = checkNotNull(windowAssigner);
+ this.windowSerializer = checkNotNull(windowSerializer);
+ this.keySelector = checkNotNull(keySelector);
+ this.keySerializer = checkNotNull(keySerializer);
+ this.windowStateDescriptor = windowStateDescriptor;
+ this.windowFunction = windowFunction;
+ this.trigger = checkNotNull(trigger);
+ this.allowedLateness = allowedLateness;
+ this.lateDataOutputTag = lateDataOutputTag;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends StreamOperator<OUT>> T createStreamOperator(
+ StreamOperatorParameters<OUT> parameters) {
+ WindowOperator<K, IN, ACC, OUT, W> operator =
+ new WindowOperator<>(
+ windowAssigner,
+ windowSerializer,
+ keySelector,
+ keySerializer,
+ windowStateDescriptor,
+ windowFunction,
+ trigger,
+ allowedLateness,
+ lateDataOutputTag);
+ operator.setup(
+ parameters.getContainingTask(),
+ parameters.getStreamConfig(),
+ parameters.getOutput());
+ operator.setProcessingTimeService(parameters.getProcessingTimeService());
+ return (T) operator;
+ }
+
+ @Override
+ public Class<? extends StreamOperator> getStreamOperatorClass(ClassLoader classLoader) {
+ return WindowOperator.class;
+ }
+
+ // ------------------------------------------------------------------------
+ // Getters for testing
+ // ------------------------------------------------------------------------
+
+ @VisibleForTesting
+ public Trigger<? super IN, ? super W> getTrigger() {
+ return trigger;
+ }
+
+ @VisibleForTesting
+ public KeySelector<IN, K> getKeySelector() {
+ return keySelector;
+ }
+
+ @VisibleForTesting
+ public WindowAssigner<? super IN, W> getWindowAssigner() {
+ return windowAssigner;
+ }
+
+ @VisibleForTesting
+ public StateDescriptor<? extends AppendingState<IN, ACC>, ?> getStateDescriptor() {
+ return windowStateDescriptor;
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationTail.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationTail.java
index 928ef2ac946..b8455250d7e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationTail.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationTail.java
@@ -22,6 +22,7 @@ import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.io.BlockingQueueBroker;
import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
@@ -74,11 +75,15 @@ public class StreamIterationTail<IN> extends OneInputStreamTask<IN, IN> {
LOG.info("Iteration tail {} acquired feedback queue {}", getName(), brokerID);
- RecordPusher<IN> headOperator = new RecordPusher<>();
- headOperator.setup(
- this,
- getConfiguration(),
- new IterationTailOutput<>(dataChannel, iterationWaitTime));
+ RecordPusher<IN> headOperator =
+ new RecordPusher<>(
+ new StreamOperatorParameters<>(
+ this,
+ getConfiguration(),
+ new IterationTailOutput<>(dataChannel, iterationWaitTime),
+ () -> new SystemProcessingTimeService(ex -> {}),
+ null,
+ null));
this.mainOperator = headOperator;
// call super.init() last because that needs this.headOperator to be set up
@@ -90,6 +95,10 @@ public class StreamIterationTail<IN> extends OneInputStreamTask<IN, IN> {
private static final long serialVersionUID = 1L;
+ private RecordPusher(StreamOperatorParameters<IN> parameters) {
+ super(parameters);
+ }
+
@Override
public void processElement(StreamRecord<IN> record) throws Exception {
output.collect(record);
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/TimestampsAndWatermarksTransformationTranslator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/TimestampsAndWatermarksTransformationTranslator.java
index d7291203ecb..498e60eadb6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/TimestampsAndWatermarksTransformationTranslator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/TimestampsAndWatermarksTransformationTranslator.java
@@ -20,9 +20,8 @@ package org.apache.flink.streaming.runtime.translators;
import org.apache.flink.annotation.Internal;
import org.apache.flink.streaming.api.graph.TransformationTranslator;
-import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
import org.apache.flink.streaming.api.transformations.TimestampsAndWatermarksTransformation;
-import org.apache.flink.streaming.runtime.operators.TimestampsAndWatermarksOperator;
+import org.apache.flink.streaming.runtime.operators.TimestampsAndWatermarksOperatorFactory;
import java.util.Collection;
@@ -59,10 +58,9 @@ public class TimestampsAndWatermarksTransformationTranslator<IN>
checkNotNull(transformation);
checkNotNull(context);
- TimestampsAndWatermarksOperator<IN> operator =
- new TimestampsAndWatermarksOperator<>(
+ TimestampsAndWatermarksOperatorFactory<IN> operatorFactory =
+ new TimestampsAndWatermarksOperatorFactory<>(
transformation.getWatermarkStrategy(), emitProgressiveWatermarks);
- SimpleOperatorFactory<IN> operatorFactory = SimpleOperatorFactory.of(operator);
operatorFactory.setChainingStrategy(transformation.getChainingStrategy());
return translateInternal(
transformation,
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java
index 623a604cae7..50ff822dfc5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java
@@ -63,8 +63,16 @@ class SourceOperatorSplitWatermarkAlignmentTest {
MockSourceReader sourceReader =
new MockSourceReader(WaitingForSplits.DO_NOT_WAIT_FOR_SPLITS, false, true);
+ Environment env = getTestingEnvironment();
SourceOperator<Integer, MockSourceSplit> operator =
new TestingSourceOperator<>(
+ new StreamOperatorParameters<>(
+ new SourceOperatorStreamTask<Integer>(env),
+ new MockStreamConfig(new Configuration(), 1),
+ new MockOutput<>(new ArrayList<>()),
+ TestProcessingTimeService::new,
+ null,
+ null),
sourceReader,
WatermarkStrategy.forGenerator(ctx -> new TestWatermarkGenerator())
.withTimestampAssigner((r, l) -> r)
@@ -74,11 +82,6 @@ class SourceOperatorSplitWatermarkAlignmentTest {
1,
5,
true);
- Environment env = getTestingEnvironment();
- operator.setup(
- new SourceOperatorStreamTask<Integer>(env),
- new MockStreamConfig(new Configuration(), 1),
- new MockOutput<>(new ArrayList<>()));
operator.initializeState(
new StreamTaskStateInitializerImpl(env, new HashMapStateBackend()));
@@ -262,8 +265,16 @@ class SourceOperatorSplitWatermarkAlignmentTest {
long idleTimeout)
throws Exception {
+ Environment env = getTestingEnvironment();
SourceOperator<Integer, MockSourceSplit> operator =
new TestingSourceOperator<>(
+ new StreamOperatorParameters<>(
+ new SourceOperatorStreamTask<Integer>(env),
+ new MockStreamConfig(new Configuration(), 1),
+ new MockOutput<>(new ArrayList<>()),
+ () -> processingTimeService,
+ null,
+ null),
sourceReader,
WatermarkStrategy.forGenerator(ctx -> new TestWatermarkGenerator())
.withTimestampAssigner((r, l) -> r)
@@ -274,11 +285,6 @@ class SourceOperatorSplitWatermarkAlignmentTest {
1,
5,
true);
- Environment env = getTestingEnvironment();
- operator.setup(
- new SourceOperatorStreamTask<Integer>(env),
- new MockStreamConfig(new Configuration(), 1),
- new MockOutput<>(new ArrayList<>()));
operator.initializeState(
new StreamTaskStateInitializerImpl(env, new HashMapStateBackend()));
operator.open();
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTestContext.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTestContext.java
index 8fd2f56b416..f7e75788eb3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTestContext.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTestContext.java
@@ -93,8 +93,16 @@ public class SourceOperatorTestContext implements AutoCloseable {
usePerSplitOutputs);
mockGateway = new MockOperatorEventGateway();
timeService = new TestProcessingTimeService();
+ Environment env = getTestingEnvironment();
operator =
new TestingSourceOperator<>(
+ new StreamOperatorParameters<>(
+ new SourceOperatorStreamTask<Integer>(env),
+ new MockStreamConfig(new Configuration(), 1),
+ output,
+ () -> timeService,
+ null,
+ null),
mockSourceReader,
watermarkStrategy,
timeService,
@@ -102,11 +110,6 @@ public class SourceOperatorTestContext implements AutoCloseable {
SUBTASK_INDEX,
5,
true);
- Environment env = getTestingEnvironment();
- operator.setup(
- new SourceOperatorStreamTask<Integer>(env),
- new MockStreamConfig(new Configuration(), 1),
- output);
operator.initializeState(
new StreamTaskStateInitializerImpl(env, new HashMapStateBackend()));
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StateDescriptorPassingTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StateDescriptorPassingTest.java
index 40ea2cbf4d3..9ab8c8e0736 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StateDescriptorPassingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StateDescriptorPassingTest.java
@@ -39,6 +39,7 @@ import org.apache.flink.streaming.api.transformations.OneInputTransformation;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.runtime.operators.windowing.WindowOperator;
+import org.apache.flink.streaming.runtime.operators.windowing.WindowOperatorFactory;
import org.apache.flink.util.Collector;
import com.esotericsoftware.kryo.Kryo;
@@ -271,8 +272,15 @@ class StateDescriptorPassingTest {
private void validateStateDescriptorConfigured(SingleOutputStreamOperator<?> result) {
OneInputTransformation<?, ?> transform =
(OneInputTransformation<?, ?>) result.getTransformation();
- WindowOperator<?, ?, ?, ?, ?> op = (WindowOperator<?, ?, ?, ?, ?>) transform.getOperator();
- StateDescriptor<?, ?> descr = op.getStateDescriptor();
+ StreamOperatorFactory<?> factory = transform.getOperatorFactory();
+ StateDescriptor<?, ?> descr;
+ if (factory instanceof WindowOperatorFactory) {
+ descr = ((WindowOperatorFactory<?, ?, ?, ?, ?>) factory).getStateDescriptor();
+ } else {
+ WindowOperator<?, ?, ?, ?, ?> op =
+ (WindowOperator<?, ?, ?, ?, ?>) transform.getOperator();
+ descr = op.getStateDescriptor();
+ }
// this would be the first statement to fail if state descriptors were not properly
// initialized
@@ -289,8 +297,15 @@ class StateDescriptorPassingTest {
private void validateListStateDescriptorConfigured(SingleOutputStreamOperator<?> result) {
OneInputTransformation<?, ?> transform =
(OneInputTransformation<?, ?>) result.getTransformation();
- WindowOperator<?, ?, ?, ?, ?> op = (WindowOperator<?, ?, ?, ?, ?>) transform.getOperator();
- StateDescriptor<?, ?> descr = op.getStateDescriptor();
+ StreamOperatorFactory<?> factory = transform.getOperatorFactory();
+ StateDescriptor<?, ?> descr;
+ if (factory instanceof WindowOperatorFactory) {
+ descr = ((WindowOperatorFactory<?, ?, ?, ?, ?>) factory).getStateDescriptor();
+ } else {
+ WindowOperator<?, ?, ?, ?, ?> op =
+ (WindowOperator<?, ?, ?, ?, ?>) transform.getOperator();
+ descr = op.getStateDescriptor();
+ }
assertThat(descr).isInstanceOf(ListStateDescriptor.class);
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContextTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContextTest.java
index 2efd0f3813b..0cf2009bedd 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContextTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContextTest.java
@@ -434,22 +434,27 @@ class StreamingRuntimeContextTest {
boolean stateV2)
throws Exception {
+ StreamConfig streamConfig = new StreamConfig(new Configuration());
+ streamConfig.setOperatorID(new OperatorID());
AbstractStreamOperator<?> operator =
- new AbstractStreamOperator<Object>() {
+ new AbstractStreamOperator<Object>(
+ new StreamOperatorParameters<>(
+ new MockStreamTaskBuilder(environment)
+ .setExecutionConfig(config)
+ .build(),
+ streamConfig,
+ new CollectorOutput<>(new ArrayList<>()),
+ TestProcessingTimeService::new,
+ null,
+ null)) {
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<Object>> output) {
super.setup(containingTask, config, output);
}
};
- StreamConfig streamConfig = new StreamConfig(new Configuration());
- streamConfig.setOperatorID(new OperatorID());
- operator.setup(
- new MockStreamTaskBuilder(environment).setExecutionConfig(config).build(),
- streamConfig,
- new CollectorOutput<>(new ArrayList<>()));
StreamTaskStateInitializer streamTaskStateManager =
new StreamTaskStateInitializerImpl(environment, new HashMapStateBackend());
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/source/TestingSourceOperator.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/source/TestingSourceOperator.java
index f6130a8ccc1..9727d3ae6ed 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/source/TestingSourceOperator.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/source/TestingSourceOperator.java
@@ -39,6 +39,7 @@ import org.apache.flink.runtime.state.StateInitializationContextImpl;
import org.apache.flink.runtime.state.TestTaskStateManager;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.operators.SourceOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
import org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask;
@@ -60,12 +61,14 @@ public class TestingSourceOperator<T> extends SourceOperator<T, MockSourceSplit>
private final int parallelism;
public TestingSourceOperator(
+ StreamOperatorParameters<T> parameters,
SourceReader<T, MockSourceSplit> reader,
WatermarkStrategy<T> watermarkStrategy,
ProcessingTimeService timeService,
boolean emitProgressiveWatermarks) {
this(
+ parameters,
reader,
watermarkStrategy,
timeService,
@@ -76,6 +79,7 @@ public class TestingSourceOperator<T> extends SourceOperator<T, MockSourceSplit>
}
public TestingSourceOperator(
+ StreamOperatorParameters<T> parameters,
SourceReader<T, MockSourceSplit> reader,
WatermarkStrategy<T> watermarkStrategy,
ProcessingTimeService timeService,
@@ -85,6 +89,7 @@ public class TestingSourceOperator<T> extends SourceOperator<T, MockSourceSplit>
boolean emitProgressiveWatermarks) {
super(
+ parameters,
(context) -> reader,
eventGateway,
new MockSourceSplitSerializer(),
@@ -146,20 +151,25 @@ public class TestingSourceOperator<T> extends SourceOperator<T, MockSourceSplit>
final SourceOperator<T, MockSourceSplit> sourceOperator =
new TestingSourceOperator<>(
- reader, watermarkStrategy, timeService, emitProgressiveWatermarks);
-
- sourceOperator.setup(
- new SourceOperatorStreamTask<Integer>(
- new StreamMockEnvironment(
- new Configuration(),
- new Configuration(),
- new ExecutionConfig(),
- 1L,
- new MockInputSplitProvider(),
- 1,
- new TestTaskStateManager())),
- new MockStreamConfig(new Configuration(), 1),
- new MockOutput<>(new ArrayList<>()));
+ new StreamOperatorParameters<>(
+ new SourceOperatorStreamTask<Integer>(
+ new StreamMockEnvironment(
+ new Configuration(),
+ new Configuration(),
+ new ExecutionConfig(),
+ 1L,
+ new MockInputSplitProvider(),
+ 1,
+ new TestTaskStateManager())),
+ new MockStreamConfig(new Configuration(), 1),
+ new MockOutput<>(new ArrayList<>()),
+ () -> timeService,
+ null,
+ null),
+ reader,
+ watermarkStrategy,
+ timeService,
+ emitProgressiveWatermarks);
sourceOperator.initializeState(stateContext);
sourceOperator.open();
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/StreamOperatorChainingTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/StreamOperatorChainingTest.java
index 115cb8e2740..d425fc61b00 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/StreamOperatorChainingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/StreamOperatorChainingTest.java
@@ -51,6 +51,7 @@ import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
+import static org.apache.flink.streaming.api.operators.StreamOperatorUtils.setupStreamOperator;
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for stream operator chaining behaviour. */
@@ -133,7 +134,8 @@ class StreamOperatorChainingTest {
OperatorChain<Integer, StreamMap<Integer, Integer>> operatorChain =
createOperatorChain(streamConfig, environment, mockTask);
- headOperator.setup(mockTask, streamConfig, operatorChain.getMainOperatorOutput());
+ setupStreamOperator(
+ headOperator, mockTask, streamConfig, operatorChain.getMainOperatorOutput());
operatorChain.initializeStateAndOpenOperators(null);
@@ -259,7 +261,8 @@ class StreamOperatorChainingTest {
OperatorChain<Integer, StreamMap<Integer, Integer>> operatorChain =
createOperatorChain(streamConfig, environment, mockTask);
- headOperator.setup(mockTask, streamConfig, operatorChain.getMainOperatorOutput());
+ setupStreamOperator(
+ headOperator, mockTask, streamConfig, operatorChain.getMainOperatorOutput());
operatorChain.initializeStateAndOpenOperators(null);
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/StreamSourceOperatorLatencyMetricsTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/StreamSourceOperatorLatencyMetricsTest.java
index 8128987428b..529e7f1bc97 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/StreamSourceOperatorLatencyMetricsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/StreamSourceOperatorLatencyMetricsTest.java
@@ -50,6 +50,8 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import static org.apache.flink.streaming.api.operators.StreamOperatorUtils.setProcessingTimeService;
+import static org.apache.flink.streaming.api.operators.StreamOperatorUtils.setupStreamOperator;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
@@ -246,9 +248,10 @@ class StreamSourceOperatorLatencyMetricsTest {
.setTimerService(timerService)
.build();
- operator.setProcessingTimeService(
+ setProcessingTimeService(
+ operator,
mockTask.getProcessingTimeServiceFactory().createProcessingTimeService(null));
- operator.setup(mockTask, cfg, (Output<StreamRecord<T>>) mock(Output.class));
+ setupStreamOperator(operator, mockTask, cfg, (Output<StreamRecord<T>>) mock(Output.class));
}
// ------------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperatorTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperatorTest.java
index 3f05af39cf3..361942ff7c4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperatorTest.java
@@ -246,8 +246,8 @@ class TimestampsAndWatermarksOperatorTest {
void watermarksWithIdlenessUnderBackpressure() throws Exception {
long idleTimeout = 100;
- TimestampsAndWatermarksOperator<Tuple2<Boolean, Long>> operator =
- new TimestampsAndWatermarksOperator<>(
+ TimestampsAndWatermarksOperatorFactory<Tuple2<Boolean, Long>> operator =
+ new TimestampsAndWatermarksOperatorFactory<>(
WatermarkStrategy.forGenerator((ctx) -> new PunctuatedWatermarkGenerator())
.withTimestampAssigner((ctx) -> new TupleExtractor())
.withIdleness(Duration.ofMillis(idleTimeout)),
@@ -285,8 +285,8 @@ class TimestampsAndWatermarksOperatorTest {
private static <T> OneInputStreamOperatorTestHarness<T, T> createTestHarness(
WatermarkStrategy<T> watermarkStrategy) throws Exception {
- final TimestampsAndWatermarksOperator<T> operator =
- new TimestampsAndWatermarksOperator<>(watermarkStrategy, true);
+ final TimestampsAndWatermarksOperatorFactory<T> operator =
+ new TimestampsAndWatermarksOperatorFactory<>(watermarkStrategy, true);
OneInputStreamOperatorTestHarness<T, T> testHarness =
new OneInputStreamOperatorTestHarness<>(operator);
@@ -301,8 +301,8 @@ class TimestampsAndWatermarksOperatorTest {
private static <T> OneInputStreamOperatorTestHarness<T, T> createBatchHarness(
WatermarkStrategy<T> watermarkStrategy) throws Exception {
- final TimestampsAndWatermarksOperator<T> operator =
- new TimestampsAndWatermarksOperator<>(watermarkStrategy, false);
+ final TimestampsAndWatermarksOperatorFactory<T> operator =
+ new TimestampsAndWatermarksOperatorFactory<>(watermarkStrategy, false);
OneInputStreamOperatorTestHarness<T, T> testHarness =
new OneInputStreamOperatorTestHarness<>(operator);
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperatorContractTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperatorContractTest.java
index 3e2f7dd376d..088cd914aa4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperatorContractTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperatorContractTest.java
@@ -69,8 +69,8 @@ class EvictingWindowOperatorContractTest extends WindowOperatorContractTest {
new StreamElementSerializer(IntSerializer.INSTANCE));
@SuppressWarnings("unchecked")
- EvictingWindowOperator<Integer, Integer, OUT, W> operator =
- new EvictingWindowOperator<>(
+ EvictingWindowOperatorFactory<Integer, Integer, OUT, W> operator =
+ new EvictingWindowOperatorFactory<>(
assigner,
assigner.getWindowSerializer(new ExecutionConfig()),
keySelector,
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/RegularWindowOperatorContractTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/RegularWindowOperatorContractTest.java
index f30780fda00..7308dffe253 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/RegularWindowOperatorContractTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/RegularWindowOperatorContractTest.java
@@ -182,8 +182,8 @@ class RegularWindowOperatorContractTest extends WindowOperatorContractTest {
};
@SuppressWarnings("unchecked")
- WindowOperator<Integer, Integer, ACC, OUT, W> operator =
- new WindowOperator<>(
+ WindowOperatorFactory<Integer, Integer, ACC, OUT, W> operator =
+ new WindowOperatorFactory<>(
assigner,
assigner.getWindowSerializer(new ExecutionConfig()),
keySelector,
@@ -222,8 +222,8 @@ class RegularWindowOperatorContractTest extends WindowOperatorContractTest {
new ListStateDescriptor<>("int-list", IntSerializer.INSTANCE);
@SuppressWarnings("unchecked")
- WindowOperator<Integer, Integer, Iterable<Integer>, OUT, W> operator =
- new WindowOperator<>(
+ WindowOperatorFactory<Integer, Integer, Iterable<Integer>, OUT, W> operator =
+ new WindowOperatorFactory<>(
assigner,
assigner.getWindowSerializer(new ExecutionConfig()),
keySelector,
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/OperatorChainTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/OperatorChainTest.java
index bec6d848cda..1a622a3f929 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/OperatorChainTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/OperatorChainTest.java
@@ -26,7 +26,6 @@ import org.apache.flink.runtime.operators.testutils.MockEnvironment;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-import org.apache.flink.streaming.api.operators.SetupableStreamOperator;
import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.runtime.io.RecordWriterOutput;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -39,6 +38,7 @@ import java.util.List;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicInteger;
+import static org.apache.flink.streaming.api.operators.StreamOperatorUtils.setupStreamOperator;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.assertj.core.api.Assertions.assertThat;
@@ -95,8 +95,9 @@ class OperatorChainTest {
for (int i = 0; i < operators.length; i++) {
int operatorIndex = operators.length - i - 1;
OneInputStreamOperator<T, T> op = operators[operatorIndex];
- if (op instanceof SetupableStreamOperator) {
- ((SetupableStreamOperator) op).setup(containingTask, cfg, lastWriter);
+ if (op instanceof AbstractStreamOperator) {
+ setupStreamOperator(
+ (AbstractStreamOperator<T>) op, containingTask, cfg, lastWriter);
}
lastWriter = new ChainingOutput<>(op, null, op.getMetricGroup(), null);
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapperTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapperTest.java
index c036776fb5e..09c5c3100c7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapperTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapperTest.java
@@ -99,7 +99,6 @@ class StreamOperatorWrapperTest {
processingTimeService,
mailboxExecutor,
timerMailController);
- streamOperator.setProcessingTimeService(processingTimeService);
StreamOperatorWrapper<?, ?> operatorWrapper =
new StreamOperatorWrapper<>(
@@ -281,6 +280,7 @@ class StreamOperatorWrapperTest {
processingTimeService.registerTimer(
Long.MAX_VALUE, t2 -> output.add("[" + name + "]: Timer not triggered"));
+ super.setProcessingTimeService(processingTimeService);
}
public String getName() {
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java b/flink-runtime/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java
index 5c8b0e2eeee..4f6791e3268 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java
@@ -66,7 +66,6 @@ import org.apache.flink.streaming.api.operators.KeyContext;
import org.apache.flink.streaming.api.operators.OperatorSnapshotFinalizer;
import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
import org.apache.flink.streaming.api.operators.Output;
-import org.apache.flink.streaming.api.operators.SetupableStreamOperator;
import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
@@ -102,6 +101,8 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;
+import static org.apache.flink.streaming.api.operators.StreamOperatorUtils.setProcessingTimeService;
+import static org.apache.flink.streaming.api.operators.StreamOperatorUtils.setupStreamOperator;
import static org.apache.flink.util.Preconditions.checkState;
/** Base class for {@code AbstractStreamOperator} test harnesses. */
@@ -432,12 +433,20 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
return outputValues;
}
- /** Calls {@link SetupableStreamOperator#setup(StreamTask, StreamConfig, Output)} ()}. */
+ /**
+ * Calls {@link
+ * org.apache.flink.streaming.api.operators.StreamOperatorUtils#setupStreamOperator(AbstractStreamOperator,
+ * StreamTask, StreamConfig, Output)} ()}.
+ */
public void setup() {
setup(null);
}
- /** Calls {@link SetupableStreamOperator#setup(StreamTask, StreamConfig, Output)} ()}. */
+ /**
+ * Calls {@link
+ * org.apache.flink.streaming.api.operators.StreamOperatorUtils#setupStreamOperator(AbstractStreamOperator,
+ * StreamTask, StreamConfig, Output)} ()}.
+ */
public void setup(TypeSerializer<OUT> outputSerializer) {
if (!setupCalled) {
streamTaskStateInitializer =
@@ -458,12 +467,13 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
.f0;
} else {
if (operator instanceof AbstractStreamOperator) {
- ((AbstractStreamOperator) operator)
- .setProcessingTimeService(processingTimeService);
- }
- if (operator instanceof SetupableStreamOperator) {
- ((SetupableStreamOperator) operator)
- .setup(mockTask, config, new MockOutput(outputSerializer));
+ setProcessingTimeService(
+ (AbstractStreamOperator) operator, processingTimeService);
+ setupStreamOperator(
+ (AbstractStreamOperator) operator,
+ mockTask,
+ config,
+ new MockOutput(outputSerializer));
}
}
setupCalled = true;
@@ -475,8 +485,8 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
* Calls {@link
* org.apache.flink.streaming.api.operators.StreamOperator#initializeState(StreamTaskStateInitializer)}.
* Calls {@link
- * org.apache.flink.streaming.api.operators.SetupableStreamOperator#setup(StreamTask,
- * StreamConfig, Output)} if it was not called before.
+ * org.apache.flink.streaming.api.operators.StreamOperatorUtils#setupStreamOperator(AbstractStreamOperator,
+ * StreamTask, StreamConfig, Output)} if it was not called before.
*/
public void initializeState(OperatorSubtaskState operatorStateHandles) throws Exception {
initializeState(operatorStateHandles, null);
@@ -574,8 +584,8 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
/**
* Calls {@link org.apache.flink.streaming.api.operators.StreamOperator#initializeState()}.
* Calls {@link
- * org.apache.flink.streaming.api.operators.SetupableStreamOperator#setup(StreamTask,
- * StreamConfig, Output)} if it was not called before.
+ * org.apache.flink.streaming.api.operators.StreamOperatorUtils#setupStreamOperator(AbstractStreamOperator,
+ * StreamTask, StreamConfig, Output)} if it was not called before.
*
* @param jmOperatorStateHandles the primary state (owned by JM)
* @param tmOperatorStateHandles the (optional) local state (owned by TM) or null.
@@ -684,7 +694,8 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
/**
* Calls {@link StreamOperator#open()}. This also calls {@link
- * SetupableStreamOperator#setup(StreamTask, StreamConfig, Output)} if it was not called before.
+ * org.apache.flink.streaming.api.operators.StreamOperatorUtils#setupStreamOperator(AbstractStreamOperator,
+ * StreamTask, StreamConfig, Output)} if it was not called before.
*/
public void open() throws Exception {
if (!initializeCalled) {
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
index a3fbb768962..2f147190a0a 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
@@ -33,9 +33,9 @@ import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.TimestampedCollector;
import org.apache.flink.streaming.api.operators.async.queue.OrderedStreamElementQueue;
import org.apache.flink.streaming.api.operators.async.queue.StreamElementQueue;
@@ -139,6 +139,7 @@ public class AsyncWaitOperator<IN, OUT>
private transient AtomicBoolean retryDisabledOnFinish;
public AsyncWaitOperator(
+ StreamOperatorParameters<OUT> parameters,
@Nonnull AsyncFunction<IN, OUT> asyncFunction,
long timeout,
int capacity,
@@ -146,9 +147,7 @@ public class AsyncWaitOperator<IN, OUT>
@Nonnull AsyncRetryStrategy<OUT> asyncRetryStrategy,
@Nonnull ProcessingTimeService processingTimeService,
@Nonnull MailboxExecutor mailboxExecutor) {
- super(asyncFunction);
-
- setChainingStrategy(ChainingStrategy.ALWAYS);
+ super(null, asyncFunction);
Preconditions.checkArgument(
capacity > 0, "The number of concurrent async operation should be greater than 0.");
@@ -172,10 +171,12 @@ public class AsyncWaitOperator<IN, OUT>
this.processingTimeService = Preconditions.checkNotNull(processingTimeService);
this.mailboxExecutor = mailboxExecutor;
+
+ setup(parameters.getContainingTask(), parameters.getStreamConfig(), parameters.getOutput());
}
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<OUT>> output) {
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorFactory.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorFactory.java
index e5b2df779bc..f01bedc83a9 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorFactory.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorFactory.java
@@ -70,6 +70,7 @@ public class AsyncWaitOperatorFactory<IN, OUT> extends AbstractStreamOperatorFac
StreamOperatorParameters<OUT> parameters) {
AsyncWaitOperator asyncWaitOperator =
new AsyncWaitOperator(
+ parameters,
asyncFunction,
timeout,
capacity,
@@ -77,10 +78,6 @@ public class AsyncWaitOperatorFactory<IN, OUT> extends AbstractStreamOperatorFac
asyncRetryStrategy,
processingTimeService,
getMailboxExecutor());
- asyncWaitOperator.setup(
- parameters.getContainingTask(),
- parameters.getStreamConfig(),
- parameters.getOutput());
return (T) asyncWaitOperator;
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java
index 3a527b13893..01e00d71a6d 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java
@@ -1050,7 +1050,7 @@ class StreamGraphGeneratorTest {
}
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<Integer>> output) {}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperatorLifecycleTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperatorLifecycleTest.java
index fe97d763c73..89a1104a3b6 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperatorLifecycleTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperatorLifecycleTest.java
@@ -280,7 +280,7 @@ class AbstractUdfStreamOperatorLifecycleTest {
}
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<OUT>> output) {
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/RecordProcessorUtilsTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/RecordProcessorUtilsTest.java
index 69d9703e007..b5cc6771c4c 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/RecordProcessorUtilsTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/RecordProcessorUtilsTest.java
@@ -29,20 +29,25 @@ import org.apache.flink.streaming.api.operators.Input;
import org.apache.flink.streaming.api.operators.KeyContextHandler;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.StreamTaskStateInitializer;
import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
import org.apache.flink.streaming.runtime.tasks.StreamTaskITCase;
import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
import org.apache.flink.streaming.util.MockOutput;
import org.apache.flink.streaming.util.MockStreamConfig;
import org.apache.flink.util.InstantiationUtil;
+import org.apache.flink.util.clock.Clock;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledFuture;
import static org.assertj.core.api.Assertions.assertThat;
@@ -150,7 +155,14 @@ class RecordProcessorUtilsTest {
boolean setCurrentKeyCalled = false;
NoOverrideOperator() throws Exception {
- super();
+ super(
+ new StreamOperatorParameters<>(
+ new StreamTaskITCase.NoOpStreamTask<>(new DummyEnvironment()),
+ new MockStreamConfig(new Configuration(), 1),
+ new MockOutput<>(new ArrayList<>()),
+ NoOpProcessingTimeService::new,
+ null,
+ null));
// For case that "SetKeyContextElement" has not been overridden,
// we can determine whether the "SetKeyContextElement" is called through
// "setCurrentKey". According to the implementation, we need to make the
@@ -161,10 +173,6 @@ class RecordProcessorUtilsTest {
KeySelector keySelector = x -> x;
InstantiationUtil.writeObjectToConfig(keySelector, configuration, "statePartitioner0");
InstantiationUtil.writeObjectToConfig(keySelector, configuration, "statePartitioner1");
- setup(
- new StreamTaskITCase.NoOpStreamTask<>(new DummyEnvironment()),
- new MockStreamConfig(configuration, 1),
- new MockOutput<>(new ArrayList<>()));
}
@Override
@@ -404,4 +412,34 @@ class RecordProcessorUtilsTest {
return hasKeyContext2;
}
}
+
+ private static class NoOpProcessingTimeService implements ProcessingTimeService {
+
+ @Override
+ public Clock getClock() {
+ return null;
+ }
+
+ @Override
+ public ScheduledFuture<?> scheduleAtFixedRate(
+ ProcessingTimeCallback callback, long initialDelay, long period) {
+ return null;
+ }
+
+ @Override
+ public ScheduledFuture<?> scheduleWithFixedDelay(
+ ProcessingTimeCallback callback, long initialDelay, long period) {
+ return null;
+ }
+
+ @Override
+ public CompletableFuture<Void> quiesce() {
+ return null;
+ }
+
+ @Override
+ public ScheduledFuture<?> registerTimer(long timestamp, ProcessingTimeCallback target) {
+ return null;
+ }
+ }
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamSourceOperatorWatermarksTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamSourceOperatorWatermarksTest.java
index 39187fb82a3..eafe91b5940 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamSourceOperatorWatermarksTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamSourceOperatorWatermarksTest.java
@@ -52,6 +52,7 @@ import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
+import static org.apache.flink.streaming.api.operators.StreamOperatorUtils.setupStreamOperator;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
@@ -201,7 +202,7 @@ class StreamSourceOperatorWatermarksTest {
.setTimerService(timeProvider)
.build();
- operator.setup(mockTask, cfg, (Output<StreamRecord<T>>) mock(Output.class));
+ setupStreamOperator(operator, mockTask, cfg, (Output<StreamRecord<T>>) mock(Output.class));
return mockTask;
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskOperatorTimerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskOperatorTimerTest.java
index d2ca5c7fe08..0525fddbdf5 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskOperatorTimerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskOperatorTimerTest.java
@@ -94,12 +94,10 @@ class StreamTaskOperatorTimerTest {
public <Operator extends StreamOperator<String>> Operator createStreamOperator(
StreamOperatorParameters<String> parameters) {
TestOperator operator =
- new TestOperator(parameters.getStreamConfig().getChainIndex(), mailboxExecutor);
- operator.setProcessingTimeService(processingTimeService);
- operator.setup(
- parameters.getContainingTask(),
- parameters.getStreamConfig(),
- parameters.getOutput());
+ new TestOperator(
+ parameters,
+ parameters.getStreamConfig().getChainIndex(),
+ mailboxExecutor);
return (Operator) operator;
}
@@ -119,7 +117,11 @@ class StreamTaskOperatorTimerTest {
private final int chainIndex;
private transient int count;
- TestOperator(int chainIndex, MailboxExecutor mailboxExecutor) {
+ TestOperator(
+ StreamOperatorParameters<String> parameters,
+ int chainIndex,
+ MailboxExecutor mailboxExecutor) {
+ super(parameters);
this.chainIndex = chainIndex;
this.mailboxExecutor = mailboxExecutor;
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AllWindowTranslationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AllWindowTranslationTest.java
index 17952332666..653822d5627 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AllWindowTranslationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AllWindowTranslationTest.java
@@ -35,7 +35,8 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
import org.apache.flink.streaming.api.transformations.OneInputTransformation;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
@@ -208,11 +209,11 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple3<String, String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple3<String, String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple3<String, String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ StreamOperatorFactory<Tuple3<String, String, Integer>> operator =
+ transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(EventTimeTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(EventTimeSessionWindows.class);
assertThat(winOperator.getStateDescriptor()).isInstanceOf(ListStateDescriptor.class);
@@ -245,11 +246,10 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(EventTimeTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(SlidingEventTimeWindows.class);
assertThat(winOperator.getStateDescriptor()).isInstanceOf(ReducingStateDescriptor.class);
@@ -278,11 +278,10 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(ProcessingTimeTrigger.class);
assertThat(winOperator.getWindowAssigner())
.isInstanceOf(SlidingProcessingTimeWindows.class);
@@ -330,17 +329,17 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple3<String, String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple3<String, String, Integer>>)
window.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple3<String, String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ StreamOperatorFactory<Tuple3<String, String, Integer>> operator =
+ transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(EventTimeTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(TumblingEventTimeWindows.class);
assertThat(winOperator.getStateDescriptor()).isInstanceOf(ReducingStateDescriptor.class);
processElementAndEnsureOutput(
- operator,
+ winOperator,
winOperator.getKeySelector(),
BasicTypeInfo.STRING_TYPE_INFO,
new Tuple2<>("hello", 1));
@@ -379,18 +378,18 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple3<String, String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple3<String, String, Integer>>)
window.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple3<String, String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ StreamOperatorFactory<Tuple3<String, String, Integer>> operator =
+ transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(ProcessingTimeTrigger.class);
assertThat(winOperator.getWindowAssigner())
.isInstanceOf(TumblingProcessingTimeWindows.class);
assertThat(winOperator.getStateDescriptor()).isInstanceOf(ReducingStateDescriptor.class);
processElementAndEnsureOutput(
- operator,
+ winOperator,
winOperator.getKeySelector(),
BasicTypeInfo.STRING_TYPE_INFO,
new Tuple2<>("hello", 1));
@@ -431,17 +430,17 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple3<String, String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple3<String, String, Integer>>)
window.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple3<String, String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ StreamOperatorFactory<Tuple3<String, String, Integer>> operator =
+ transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(EventTimeTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(TumblingEventTimeWindows.class);
assertThat(winOperator.getStateDescriptor()).isInstanceOf(ReducingStateDescriptor.class);
processElementAndEnsureOutput(
- operator,
+ winOperator,
winOperator.getKeySelector(),
BasicTypeInfo.STRING_TYPE_INFO,
new Tuple2<>("hello", 1));
@@ -480,18 +479,18 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple3<String, String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple3<String, String, Integer>>)
window.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple3<String, String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ StreamOperatorFactory<Tuple3<String, String, Integer>> operator =
+ transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(ProcessingTimeTrigger.class);
assertThat(winOperator.getWindowAssigner())
.isInstanceOf(TumblingProcessingTimeWindows.class);
assertThat(winOperator.getStateDescriptor()).isInstanceOf(ReducingStateDescriptor.class);
processElementAndEnsureOutput(
- operator,
+ winOperator,
winOperator.getKeySelector(),
BasicTypeInfo.STRING_TYPE_INFO,
new Tuple2<>("hello", 1));
@@ -533,11 +532,10 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(EvictingWindowOperator.class);
- EvictingWindowOperator<String, Tuple2<String, Integer>, ?, ?> winOperator =
- (EvictingWindowOperator<String, Tuple2<String, Integer>, ?, ?>) operator;
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(EvictingWindowOperatorFactory.class);
+ EvictingWindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?> winOperator =
+ (EvictingWindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(EventTimeTrigger.class);
assertThat(winOperator.getEvictor()).isInstanceOf(CountEvictor.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(SlidingEventTimeWindows.class);
@@ -586,17 +584,17 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple3<String, String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple3<String, String, Integer>>)
window.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple3<String, String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ StreamOperatorFactory<Tuple3<String, String, Integer>> operator =
+ transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(EventTimeTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(TumblingEventTimeWindows.class);
assertThat(winOperator.getStateDescriptor()).isInstanceOf(ReducingStateDescriptor.class);
processElementAndEnsureOutput(
- operator,
+ winOperator,
winOperator.getKeySelector(),
BasicTypeInfo.STRING_TYPE_INFO,
new Tuple2<>("hello", 1));
@@ -623,12 +621,11 @@ class AllWindowTranslationTest {
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(EventTimeTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(SlidingEventTimeWindows.class);
@@ -658,12 +655,11 @@ class AllWindowTranslationTest {
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(ProcessingTimeTrigger.class);
assertThat(winOperator.getWindowAssigner())
@@ -692,19 +688,19 @@ class AllWindowTranslationTest {
(OneInputTransformation<Tuple2<String, Integer>, Tuple3<String, String, Integer>>)
window.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple3<String, String, Integer>> operator =
- transform.getOperator();
+ StreamOperatorFactory<Tuple3<String, String, Integer>> operator =
+ transform.getOperatorFactory();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(EventTimeTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(TumblingEventTimeWindows.class);
assertThat(winOperator.getStateDescriptor()).isInstanceOf(AggregatingStateDescriptor.class);
processElementAndEnsureOutput(
- operator,
+ winOperator,
winOperator.getKeySelector(),
BasicTypeInfo.STRING_TYPE_INFO,
new Tuple2<>("hello", 1));
@@ -725,12 +721,12 @@ class AllWindowTranslationTest {
(OneInputTransformation<Tuple2<String, Integer>, Tuple3<String, String, Integer>>)
window.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple3<String, String, Integer>> operator =
- transform.getOperator();
+ StreamOperatorFactory<Tuple3<String, String, Integer>> operator =
+ transform.getOperatorFactory();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(ProcessingTimeTrigger.class);
assertThat(winOperator.getWindowAssigner())
@@ -738,7 +734,7 @@ class AllWindowTranslationTest {
assertThat(winOperator.getStateDescriptor()).isInstanceOf(AggregatingStateDescriptor.class);
processElementAndEnsureOutput(
- operator,
+ winOperator,
winOperator.getKeySelector(),
BasicTypeInfo.STRING_TYPE_INFO,
new Tuple2<>("hello", 1));
@@ -762,12 +758,11 @@ class AllWindowTranslationTest {
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(EventTimeTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(SlidingEventTimeWindows.class);
@@ -814,12 +809,11 @@ class AllWindowTranslationTest {
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(EventTimeTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(SlidingEventTimeWindows.class);
@@ -868,11 +862,10 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(EventTimeTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(TumblingEventTimeWindows.class);
assertThat(winOperator.getStateDescriptor()).isInstanceOf(ListStateDescriptor.class);
@@ -916,11 +909,10 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(ProcessingTimeTrigger.class);
assertThat(winOperator.getWindowAssigner())
.isInstanceOf(TumblingProcessingTimeWindows.class);
@@ -967,11 +959,10 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(EvictingWindowOperator.class);
- EvictingWindowOperator<String, Tuple2<String, Integer>, ?, ?> winOperator =
- (EvictingWindowOperator<String, Tuple2<String, Integer>, ?, ?>) operator;
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(EvictingWindowOperatorFactory.class);
+ EvictingWindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?> winOperator =
+ (EvictingWindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(CountTrigger.class);
assertThat(winOperator.getEvictor()).isInstanceOf(TimeEvictor.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(TumblingEventTimeWindows.class);
@@ -1017,11 +1008,10 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(CountTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(TumblingEventTimeWindows.class);
assertThat(winOperator.getStateDescriptor()).isInstanceOf(ListStateDescriptor.class);
@@ -1069,11 +1059,10 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(EventTimeTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(TumblingEventTimeWindows.class);
assertThat(winOperator.getStateDescriptor()).isInstanceOf(ListStateDescriptor.class);
@@ -1117,11 +1106,10 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(ProcessingTimeTrigger.class);
assertThat(winOperator.getWindowAssigner())
.isInstanceOf(TumblingProcessingTimeWindows.class);
@@ -1154,11 +1142,10 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(CountTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(SlidingEventTimeWindows.class);
assertThat(winOperator.getStateDescriptor()).isInstanceOf(ReducingStateDescriptor.class);
@@ -1203,11 +1190,10 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(WindowOperator.class);
- WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
- (WindowOperator<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(WindowOperatorFactory.class);
+ WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?> winOperator =
+ (WindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(CountTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(TumblingEventTimeWindows.class);
assertThat(winOperator.getStateDescriptor()).isInstanceOf(ListStateDescriptor.class);
@@ -1239,11 +1225,10 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(EvictingWindowOperator.class);
- EvictingWindowOperator<String, Tuple2<String, Integer>, ?, ?> winOperator =
- (EvictingWindowOperator<String, Tuple2<String, Integer>, ?, ?>) operator;
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(EvictingWindowOperatorFactory.class);
+ EvictingWindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?> winOperator =
+ (EvictingWindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(EventTimeTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(SlidingEventTimeWindows.class);
assertThat(winOperator.getEvictor()).isInstanceOf(CountEvictor.class);
@@ -1290,11 +1275,10 @@ class AllWindowTranslationTest {
OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform =
(OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>)
window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- transform.getOperator();
- assertThat(operator).isInstanceOf(EvictingWindowOperator.class);
- EvictingWindowOperator<String, Tuple2<String, Integer>, ?, ?> winOperator =
- (EvictingWindowOperator<String, Tuple2<String, Integer>, ?, ?>) operator;
+ StreamOperatorFactory<Tuple2<String, Integer>> operator = transform.getOperatorFactory();
+ assertThat(operator).isInstanceOf(EvictingWindowOperatorFactory.class);
+ EvictingWindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?> winOperator =
+ (EvictingWindowOperatorFactory<String, Tuple2<String, Integer>, ?, ?>) operator;
assertThat(winOperator.getTrigger()).isInstanceOf(CountTrigger.class);
assertThat(winOperator.getWindowAssigner()).isInstanceOf(TumblingEventTimeWindows.class);
assertThat(winOperator.getEvictor()).isInstanceOf(TimeEvictor.class);
@@ -1312,7 +1296,7 @@ class AllWindowTranslationTest {
* watermark and processing time to {@code Long.MAX_VALUE}.
*/
private static <K, IN, OUT> void processElementAndEnsureOutput(
- OneInputStreamOperator<IN, OUT> operator,
+ OneInputStreamOperatorFactory<IN, OUT> operator,
KeySelector<IN, K> keySelector,
TypeInformation<K> keyType,
IN element)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/ContinuousProcessingTimeTriggerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/ContinuousProcessingTimeTriggerTest.java
index 661e3641387..ae4c18ee386 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/ContinuousProcessingTimeTriggerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/ContinuousProcessingTimeTriggerTest.java
@@ -108,17 +108,20 @@ class ContinuousProcessingTimeTriggerTest {
"window-contents",
BasicTypeInfo.INT_TYPE_INFO.createSerializer(new SerializerConfigImpl()));
- WindowOperator<Byte, Integer, Iterable<Integer>, WindowedInteger, TimeWindow> operator =
- new WindowOperator<>(
- TumblingProcessingTimeWindows.of(Duration.ofMillis(10)),
- new TimeWindow.Serializer(),
- new NullByteKeySelector<>(),
- BasicTypeInfo.BYTE_TYPE_INFO.createSerializer(new SerializerConfigImpl()),
- stateDesc,
- new InternalIterableWindowFunction<>(new IntegerSumWindowFunction()),
- trigger,
- 0,
- null);
+ WindowOperatorFactory<Byte, Integer, Iterable<Integer>, WindowedInteger, TimeWindow>
+ operator =
+ new WindowOperatorFactory<>(
+ TumblingProcessingTimeWindows.of(Duration.ofMillis(10)),
+ new TimeWindow.Serializer(),
+ new NullByteKeySelector<>(),
+ BasicTypeInfo.BYTE_TYPE_INFO.createSerializer(
+ new SerializerConfigImpl()),
+ stateDesc,
+ new InternalIterableWindowFunction<>(
+ new IntegerSumWindowFunction()),
+ trigger,
+ 0,
+ null);
KeyedOneInputStreamOperatorTestHarness<Byte, Integer, WindowedInteger> testHarness =
new KeyedOneInputStreamOperatorTestHarness<>(
@@ -183,17 +186,20 @@ class ContinuousProcessingTimeTriggerTest {
"window-contents",
BasicTypeInfo.INT_TYPE_INFO.createSerializer(new SerializerConfigImpl()));
- WindowOperator<Byte, Integer, Iterable<Integer>, WindowedInteger, TimeWindow> operator =
- new WindowOperator<>(
- TumblingEventTimeWindows.of(Duration.ofMillis(10)),
- new TimeWindow.Serializer(),
- new NullByteKeySelector<>(),
- BasicTypeInfo.BYTE_TYPE_INFO.createSerializer(new SerializerConfigImpl()),
- stateDesc,
- new InternalIterableWindowFunction<>(new IntegerSumWindowFunction()),
- trigger,
- 0,
- null);
+ WindowOperatorFactory<Byte, Integer, Iterable<Integer>, WindowedInteger, TimeWindow>
+ operator =
+ new WindowOperatorFactory<>(
+ TumblingEventTimeWindows.of(Duration.ofMillis(10)),
+ new TimeWindow.Serializer(),
+ new NullByteKeySelector<>(),
+ BasicTypeInfo.BYTE_TYPE_INFO.createSerializer(
+ new SerializerConfigImpl()),
+ stateDesc,
+ new InternalIterableWindowFunction<>(
+ new IntegerSumWindowFunction()),
+ trigger,
+ 0,
+ null);
KeyedOneInputStreamOperatorTestHarness<Byte, Integer, WindowedInteger> testHarness =
new KeyedOneInputStreamOperatorTestHarness<>(
@@ -243,17 +249,20 @@ class ContinuousProcessingTimeTriggerTest {
"window-contents",
BasicTypeInfo.INT_TYPE_INFO.createSerializer(new SerializerConfigImpl()));
- WindowOperator<Byte, Integer, Iterable<Integer>, WindowedInteger, TimeWindow> operator =
- new WindowOperator<>(
- ProcessingTimeSessionWindows.withGap(Duration.ofMillis(10)),
- new TimeWindow.Serializer(),
- new NullByteKeySelector<>(),
- BasicTypeInfo.BYTE_TYPE_INFO.createSerializer(new SerializerConfigImpl()),
- stateDesc,
- new InternalIterableWindowFunction<>(new IntegerSumWindowFunction()),
- trigger,
- 0,
- null);
+ WindowOperatorFactory<Byte, Integer, Iterable<Integer>, WindowedInteger, TimeWindow>
+ operator =
+ new WindowOperatorFactory<>(
+ ProcessingTimeSessionWindows.withGap(Duration.ofMillis(10)),
+ new TimeWindow.Serializer(),
+ new NullByteKeySelector<>(),
+ BasicTypeInfo.BYTE_TYPE_INFO.createSerializer(
+ new SerializerConfigImpl()),
+ stateDesc,
+ new InternalIterableWindowFunction<>(
+ new IntegerSumWindowFunction()),
+ trigger,
+ 0,
+ null);
KeyedOneInputStreamOperatorTestHarness<Byte, Integer, WindowedInteger> testHarness =
new KeyedOneInputStreamOperatorTestHarness<>(
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperatorTest.java
index ebf388f4fb7..cd43d6bdf35 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperatorTest.java
@@ -83,10 +83,10 @@ class EvictingWindowOperatorTest {
ListStateDescriptor<StreamRecord<Tuple2<String, Integer>>> stateDesc =
new ListStateDescriptor<>("window-contents", streamRecordSerializer);
- EvictingWindowOperator<
+ EvictingWindowOperatorFactory<
String, Tuple2<String, Integer>, Tuple2<String, Integer>, GlobalWindow>
operator =
- new EvictingWindowOperator<>(
+ new EvictingWindowOperatorFactory<>(
GlobalWindows.create(),
new GlobalWindow.Serializer(),
new TupleKeySelector(),
@@ -175,10 +175,10 @@ class EvictingWindowOperatorTest {
ListStateDescriptor<StreamRecord<Tuple2<String, Integer>>> stateDesc =
new ListStateDescriptor<>("window-contents", streamRecordSerializer);
- EvictingWindowOperator<
+ EvictingWindowOperatorFactory<
String, Tuple2<String, Integer>, Tuple2<String, Integer>, GlobalWindow>
operator =
- new EvictingWindowOperator<>(
+ new EvictingWindowOperatorFactory<>(
GlobalWindows.create(),
new GlobalWindow.Serializer(),
new TupleKeySelector(),
@@ -257,9 +257,10 @@ class EvictingWindowOperatorTest {
ListStateDescriptor<StreamRecord<Tuple2<String, Integer>>> stateDesc =
new ListStateDescriptor<>("window-contents", streamRecordSerializer);
- EvictingWindowOperator<String, Tuple2<String, Integer>, Tuple2<String, Integer>, TimeWindow>
+ EvictingWindowOperatorFactory<
+ String, Tuple2<String, Integer>, Tuple2<String, Integer>, TimeWindow>
operator =
- new EvictingWindowOperator<>(
+ new EvictingWindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -341,10 +342,10 @@ class EvictingWindowOperatorTest {
ListStateDescriptor<StreamRecord<Tuple2<String, Integer>>> stateDesc =
new ListStateDescriptor<>("window-contents", streamRecordSerializer);
- EvictingWindowOperator<
+ EvictingWindowOperatorFactory<
String, Tuple2<String, Integer>, Tuple2<String, Integer>, GlobalWindow>
operator =
- new EvictingWindowOperator<>(
+ new EvictingWindowOperatorFactory<>(
GlobalWindows.create(),
new GlobalWindow.Serializer(),
new TupleKeySelector(),
@@ -422,10 +423,10 @@ class EvictingWindowOperatorTest {
ListStateDescriptor<StreamRecord<Tuple2<String, Integer>>> stateDesc =
new ListStateDescriptor<>("window-contents", streamRecordSerializer);
- EvictingWindowOperator<
+ EvictingWindowOperatorFactory<
String, Tuple2<String, Integer>, Tuple2<String, Integer>, GlobalWindow>
operator =
- new EvictingWindowOperator<>(
+ new EvictingWindowOperatorFactory<>(
GlobalWindows.create(),
new GlobalWindow.Serializer(),
new TupleKeySelector(),
@@ -516,10 +517,10 @@ class EvictingWindowOperatorTest {
ListStateDescriptor<StreamRecord<Tuple2<String, Integer>>> stateDesc =
new ListStateDescriptor<>("window-contents", streamRecordSerializer);
- EvictingWindowOperator<
+ EvictingWindowOperatorFactory<
String, Tuple2<String, Integer>, Tuple2<String, Integer>, GlobalWindow>
operator =
- new EvictingWindowOperator<>(
+ new EvictingWindowOperatorFactory<>(
GlobalWindows.create(),
new GlobalWindow.Serializer(),
new TupleKeySelector(),
@@ -609,10 +610,10 @@ class EvictingWindowOperatorTest {
ListStateDescriptor<StreamRecord<Tuple2<String, Integer>>> stateDesc =
new ListStateDescriptor<>("window-contents", streamRecordSerializer);
- EvictingWindowOperator<
+ EvictingWindowOperatorFactory<
String, Tuple2<String, Integer>, Tuple2<String, Integer>, GlobalWindow>
operator =
- new EvictingWindowOperator<>(
+ new EvictingWindowOperatorFactory<>(
GlobalWindows.create(),
new GlobalWindow.Serializer(),
new TupleKeySelector(),
@@ -700,10 +701,10 @@ class EvictingWindowOperatorTest {
ListStateDescriptor<StreamRecord<Tuple2<String, Integer>>> stateDesc =
new ListStateDescriptor<>("window-contents", streamRecordSerializer);
- EvictingWindowOperator<
+ EvictingWindowOperatorFactory<
String, Tuple2<String, Integer>, Tuple2<String, Integer>, GlobalWindow>
operator =
- new EvictingWindowOperator<>(
+ new EvictingWindowOperatorFactory<>(
GlobalWindows.create(),
new GlobalWindow.Serializer(),
new TupleKeySelector(),
@@ -785,9 +786,10 @@ class EvictingWindowOperatorTest {
ListStateDescriptor<StreamRecord<Tuple2<String, Integer>>> stateDesc =
new ListStateDescriptor<>("window-contents", streamRecordSerializer);
- EvictingWindowOperator<String, Tuple2<String, Integer>, Tuple2<String, Integer>, TimeWindow>
+ EvictingWindowOperatorFactory<
+ String, Tuple2<String, Integer>, Tuple2<String, Integer>, TimeWindow>
operator =
- new EvictingWindowOperator<>(
+ new EvictingWindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorMigrationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorMigrationTest.java
index 55053a16ea6..6ad8e9ee34c 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorMigrationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorMigrationTest.java
@@ -103,14 +103,14 @@ public class WindowOperatorMigrationTest implements MigrationTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(sessionSize)),
new TimeWindow.Serializer(),
new TupleKeySelector<String>(),
@@ -161,14 +161,14 @@ public class WindowOperatorMigrationTest implements MigrationTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(sessionSize)),
new TimeWindow.Serializer(),
new TupleKeySelector<String>(),
@@ -235,14 +235,14 @@ public class WindowOperatorMigrationTest implements MigrationTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(sessionSize)),
new TimeWindow.Serializer(),
new TupleKeySelector<String>(),
@@ -287,14 +287,14 @@ public class WindowOperatorMigrationTest implements MigrationTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(sessionSize)),
new TimeWindow.Serializer(),
new TupleKeySelector<String>(),
@@ -372,14 +372,14 @@ public class WindowOperatorMigrationTest implements MigrationTest {
new SumReducer<>(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector<>(),
@@ -452,14 +452,14 @@ public class WindowOperatorMigrationTest implements MigrationTest {
new SumReducer<>(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector<>(),
@@ -523,14 +523,14 @@ public class WindowOperatorMigrationTest implements MigrationTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector<>(),
@@ -601,14 +601,14 @@ public class WindowOperatorMigrationTest implements MigrationTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector<>(),
@@ -672,14 +672,14 @@ public class WindowOperatorMigrationTest implements MigrationTest {
new SumReducer<>(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingProcessingTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector<>(),
@@ -741,14 +741,14 @@ public class WindowOperatorMigrationTest implements MigrationTest {
new SumReducer<>(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingProcessingTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector<>(),
@@ -807,14 +807,14 @@ public class WindowOperatorMigrationTest implements MigrationTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingProcessingTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector<>(),
@@ -874,14 +874,14 @@ public class WindowOperatorMigrationTest implements MigrationTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingProcessingTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector<>(),
@@ -947,14 +947,14 @@ public class WindowOperatorMigrationTest implements MigrationTest {
TypeInformation.of(NonPojoType.class).createSerializer(new SerializerConfigImpl());
assertThat(keySerializer).isInstanceOf(KryoSerializer.class);
- WindowOperator<
+ WindowOperatorFactory<
NonPojoType,
Tuple2<NonPojoType, Integer>,
Tuple2<NonPojoType, Integer>,
Tuple2<NonPojoType, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector<>(),
@@ -1042,14 +1042,14 @@ public class WindowOperatorMigrationTest implements MigrationTest {
TypeInformation.of(NonPojoType.class).createSerializer(new SerializerConfigImpl());
assertThat(keySerializer).isInstanceOf(KryoSerializer.class);
- WindowOperator<
+ WindowOperatorFactory<
NonPojoType,
Tuple2<NonPojoType, Integer>,
Tuple2<NonPojoType, Integer>,
Tuple2<NonPojoType, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector<>(),
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorTest.java
index dbff5c61781..a842e467314 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorTest.java
@@ -35,7 +35,7 @@ import org.apache.flink.streaming.api.functions.windowing.PassThroughWindowFunct
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.DynamicEventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.DynamicProcessingTimeSessionWindows;
@@ -102,7 +102,8 @@ class WindowOperatorTest {
new OutputTag<Tuple2<String, Integer>>("late-output") {};
private void testSlidingEventTimeWindows(
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator)
+ OneInputStreamOperatorFactory<Tuple2<String, Integer>, Tuple2<String, Integer>>
+ operator)
throws Exception {
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>>
@@ -220,14 +221,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
SlidingEventTimeWindows.of(
Duration.ofSeconds(windowSize),
Duration.ofSeconds(windowSlide)),
@@ -258,14 +259,14 @@ class WindowOperatorTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
SlidingEventTimeWindows.of(
Duration.ofSeconds(windowSize),
Duration.ofSeconds(windowSlide)),
@@ -287,7 +288,8 @@ class WindowOperatorTest {
}
private void testTumblingEventTimeWindows(
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator)
+ OneInputStreamOperatorFactory<Tuple2<String, Integer>, Tuple2<String, Integer>>
+ operator)
throws Exception {
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>>
testHarness = createTestHarness(operator);
@@ -402,14 +404,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -438,14 +440,14 @@ class WindowOperatorTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -476,14 +478,14 @@ class WindowOperatorTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(sessionSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -569,14 +571,14 @@ class WindowOperatorTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(sessionSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -664,14 +666,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(sessionSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -751,14 +753,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(sessionSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -837,14 +839,14 @@ class WindowOperatorTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(sessionSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -928,14 +930,14 @@ class WindowOperatorTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(sessionSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -1022,14 +1024,14 @@ class WindowOperatorTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
new PointSessionWindows(3000),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -1084,7 +1086,7 @@ class WindowOperatorTest {
private static <OUT>
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, OUT> createTestHarness(
- OneInputStreamOperator<Tuple2<String, Integer>, OUT> operator)
+ OneInputStreamOperatorFactory<Tuple2<String, Integer>, OUT> operator)
throws Exception {
return new KeyedOneInputStreamOperatorTestHarness<>(
operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
@@ -1103,14 +1105,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
GlobalWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
GlobalWindows.create(),
new GlobalWindow.Serializer(),
new TupleKeySelector(),
@@ -1227,14 +1229,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
GlobalWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
GlobalWindows.create(),
new GlobalWindow.Serializer(),
new TupleKeySelector(),
@@ -1282,7 +1284,7 @@ class WindowOperatorTest {
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
GlobalWindows.create(),
new GlobalWindow.Serializer(),
new TupleKeySelector(),
@@ -1338,14 +1340,14 @@ class WindowOperatorTest {
STRING_INT_TUPLE.createSerializer(
new ExecutionConfig().getSerializerConfig()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
GlobalWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
GlobalWindows.createWithEndOfStreamTrigger(),
new GlobalWindow.Serializer(),
new TupleKeySelector(),
@@ -1406,14 +1408,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingProcessingTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -1483,14 +1485,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
SlidingProcessingTimeWindows.of(
Duration.ofSeconds(windowSize),
Duration.ofSeconds(windowSlide)),
@@ -1583,14 +1585,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
ProcessingTimeSessionWindows.withGap(Duration.ofSeconds(windowGap)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -1690,14 +1692,14 @@ class WindowOperatorTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
DynamicEventTimeSessionWindows.withDynamicGap(extractor),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -1790,14 +1792,14 @@ class WindowOperatorTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
DynamicProcessingTimeSessionWindows.withDynamicGap(extractor),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -1872,14 +1874,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -1958,14 +1960,14 @@ class WindowOperatorTest {
TumblingEventTimeWindows windowAssigner =
TumblingEventTimeWindows.of(Duration.ofMillis(windowSize));
- final WindowOperator<
+ final WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
windowAssigner,
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -1994,7 +1996,8 @@ class WindowOperatorTest {
new WindowAssigner.WindowAssignerContext() {
@Override
public long getCurrentProcessingTime() {
- return operator.windowAssignerContext.getCurrentProcessingTime();
+ return ((WindowOperator<?, ?, ?, ?, ?>) testHarness.getOperator())
+ .windowAssignerContext.getCurrentProcessingTime();
}
});
TimeWindow window = Iterables.getOnlyElement(windows);
@@ -2041,14 +2044,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -2123,14 +2126,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
SlidingEventTimeWindows.of(
Duration.ofSeconds(windowSize),
Duration.ofSeconds(windowSlide)),
@@ -2221,14 +2224,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(gapSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -2327,14 +2330,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(gapSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -2430,14 +2433,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(gapSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -2527,14 +2530,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(gapSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -2638,14 +2641,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(gapSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -2740,14 +2743,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(gapSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -2843,14 +2846,14 @@ class WindowOperatorTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
String,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -2915,14 +2918,14 @@ class WindowOperatorTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -2973,14 +2976,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -3032,14 +3035,14 @@ class WindowOperatorTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(gapSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -3085,14 +3088,14 @@ class WindowOperatorTest {
new SumReducer(),
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple3<String, Long, Long>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
EventTimeSessionWindows.withGap(Duration.ofSeconds(gapSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -3137,14 +3140,14 @@ class WindowOperatorTest {
"window-contents",
STRING_INT_TUPLE.createSerializer(new SerializerConfigImpl()));
- WindowOperator<
+ WindowOperatorFactory<
String,
Tuple2<String, Integer>,
Iterable<Tuple2<String, Integer>>,
Tuple2<String, Integer>,
TimeWindow>
operator =
- new WindowOperator<>(
+ new WindowOperatorFactory<>(
TumblingEventTimeWindows.of(Duration.ofSeconds(windowSize)),
new TimeWindow.Serializer(),
new TupleKeySelector(),
@@ -3165,7 +3168,8 @@ class WindowOperatorTest {
// normal element
testHarness.processElement(new StreamRecord<>(new Tuple2<>("test_key", 1), 1000));
assertThat(
- operator.processContext
+ ((WindowOperator<?, ?, ?, ?, ?>) testHarness.getOperator())
+ .processContext
.windowState()
.getListState(windowStateDesc)
.get()
@@ -3173,7 +3177,8 @@ class WindowOperatorTest {
.isEqualTo("[(test_key,1)]");
testHarness.processWatermark(new Watermark(1599));
assertThat(
- operator.processContext
+ ((WindowOperator<?, ?, ?, ?, ?>) testHarness.getOperator())
+ .processContext
.windowState()
.getListState(windowStateDesc)
.get()
@@ -3181,7 +3186,8 @@ class WindowOperatorTest {
.isEqualTo("[(test_key,1)]");
testHarness.processWatermark(new Watermark(1699));
assertThat(
- operator.processContext
+ ((WindowOperator<?, ?, ?, ?, ?>) testHarness.getOperator())
+ .processContext
.windowState()
.getListState(windowStateDesc)
.get()
@@ -3189,7 +3195,8 @@ class WindowOperatorTest {
.isEqualTo("[(test_key,1)]");
testHarness.processWatermark(new Watermark(1799));
assertThat(
- operator.processContext
+ ((WindowOperator<?, ?, ?, ?, ?>) testHarness.getOperator())
+ .processContext
.windowState()
.getListState(windowStateDesc)
.get()
@@ -3197,7 +3204,8 @@ class WindowOperatorTest {
.isEqualTo("[(test_key,1)]");
testHarness.processWatermark(new Watermark(1999));
assertThat(
- operator.processContext
+ ((WindowOperator<?, ?, ?, ?, ?>) testHarness.getOperator())
+ .processContext
.windowState()
.getListState(windowStateDesc)
.get()
@@ -3205,7 +3213,8 @@ class WindowOperatorTest {
.isEqualTo("[(test_key,1)]");
testHarness.processWatermark(new Watermark(2000));
assertThat(
- operator.processContext
+ ((WindowOperator<?, ?, ?, ?, ?>) testHarness.getOperator())
+ .processContext
.windowState()
.getListState(windowStateDesc)
.get()
@@ -3213,7 +3222,8 @@ class WindowOperatorTest {
.isEqualTo("[]");
testHarness.processWatermark(new Watermark(5000));
assertThat(
- operator.processContext
+ ((WindowOperator<?, ?, ?, ?, ?>) testHarness.getOperator())
+ .processContext
.windowState()
.getListState(windowStateDesc)
.get()
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/CoordinatorEventsToStreamOperatorRecipientExactlyOnceITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/CoordinatorEventsToStreamOperatorRecipientExactlyOnceITCase.java
index 21821a1c25b..847729f3332 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/CoordinatorEventsToStreamOperatorRecipientExactlyOnceITCase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/CoordinatorEventsToStreamOperatorRecipientExactlyOnceITCase.java
@@ -236,11 +236,7 @@ class CoordinatorEventsToStreamOperatorRecipientExactlyOnceITCase
@Override
public <T extends StreamOperator<OUT>> T createStreamOperator(
StreamOperatorParameters<OUT> parameters) {
- EventReceivingOperator<OUT> operator = new EventReceivingOperator<>();
- operator.setup(
- parameters.getContainingTask(),
- parameters.getStreamConfig(),
- parameters.getOutput());
+ EventReceivingOperator<OUT> operator = new EventReceivingOperator<>(parameters);
parameters
.getOperatorEventDispatcher()
.registerEventHandler(parameters.getStreamConfig().getOperatorID(), operator);
@@ -380,6 +376,18 @@ class CoordinatorEventsToStreamOperatorRecipientExactlyOnceITCase
protected ListState<Integer> state;
+ private EventReceivingOperator(StreamOperatorParameters<T> parameters) {
+ super(parameters);
+ }
+
+ @Override
+ public void setup(
+ StreamTask<?, ?> containingTask,
+ StreamConfig config,
+ Output<StreamRecord<T>> output) {
+ super.setup(containingTask, config, output);
+ }
+
@Override
public void open() throws Exception {
super.open();
@@ -465,7 +473,7 @@ class CoordinatorEventsToStreamOperatorRecipientExactlyOnceITCase
public <T extends StreamOperator<OUT>> T createStreamOperator(
StreamOperatorParameters<OUT> parameters) {
EventReceivingOperator<OUT> operator =
- new EventReceivingOperatorWithFailure<>(name, numEvents);
+ new EventReceivingOperatorWithFailure<>(parameters, name, numEvents);
operator.setup(
parameters.getContainingTask(),
parameters.getStreamConfig(),
@@ -490,7 +498,9 @@ class CoordinatorEventsToStreamOperatorRecipientExactlyOnceITCase
private TestScript testScript;
- private EventReceivingOperatorWithFailure(String name, int numEvents) {
+ private EventReceivingOperatorWithFailure(
+ StreamOperatorParameters<T> parameters, String name, int numEvents) {
+ super(parameters);
this.name = name;
this.maxNumberBeforeFailure = numEvents / 3 + new Random().nextInt(numEvents / 6);
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
index f69fffe48c1..6ee5efddd29 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
@@ -87,6 +87,7 @@ import java.util.function.Supplier;
import static org.apache.flink.runtime.checkpoint.CheckpointType.CHECKPOINT;
import static org.apache.flink.shaded.guava32.com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
+import static org.apache.flink.streaming.api.operators.StreamOperatorUtils.setProcessingTimeService;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.fail;
@@ -282,7 +283,7 @@ class SubtaskCheckpointCoordinatorTest {
StreamMap<String, String> streamMap =
new StreamMap<>((MapFunction<String, String>) value -> value);
- streamMap.setProcessingTimeService(new TestProcessingTimeService());
+ setProcessingTimeService(streamMap, new TestProcessingTimeService());
final OperatorChain<String, AbstractStreamOperator<String>> operatorChain =
operatorChain(streamMap);
StreamTaskStateInitializerImpl stateInitializer =
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecTableSourceScan.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecTableSourceScan.java
index 929bdc43ae1..d5a9127bd45 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecTableSourceScan.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecTableSourceScan.java
@@ -32,6 +32,7 @@ import org.apache.flink.streaming.api.functions.source.legacy.ParallelSourceFunc
import org.apache.flink.streaming.api.functions.source.legacy.SourceFunction;
import org.apache.flink.streaming.api.lineage.LineageDataset;
import org.apache.flink.streaming.api.lineage.LineageVertex;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.StreamSource;
import org.apache.flink.streaming.api.transformations.LegacySourceTransformation;
import org.apache.flink.streaming.api.transformations.PartitionTransformation;
@@ -354,13 +355,16 @@ public abstract class CommonExecTableSourceScan extends ExecNodeBase<RowData>
}
final StreamSource<RowData, ?> sourceOperator = new StreamSource<>(function, !isBounded);
- return new LegacySourceTransformation<>(
- operatorName,
- sourceOperator,
- outputTypeInfo,
- parallelism,
- boundedness,
- sourceParallelismConfigured);
+ LegacySourceTransformation<RowData> transformation =
+ new LegacySourceTransformation<>(
+ operatorName,
+ sourceOperator,
+ outputTypeInfo,
+ parallelism,
+ boundedness,
+ sourceParallelismConfigured);
+ transformation.setChainingStrategy(ChainingStrategy.HEAD);
+ return transformation;
}
/**
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/TableStreamOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/TableStreamOperator.java
index cbfc62aee08..d3b1d910c86 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/TableStreamOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/TableStreamOperator.java
@@ -22,8 +22,8 @@ import org.apache.flink.core.memory.ManagedMemoryUseCase;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
@@ -43,7 +43,11 @@ public abstract class TableStreamOperator<OUT> extends AbstractStreamOperator<OU
protected transient ContextImpl ctx;
public TableStreamOperator() {
- setChainingStrategy(ChainingStrategy.ALWAYS);
+ this(null);
+ }
+
+ public TableStreamOperator(StreamOperatorParameters<OUT> parameters) {
+ super(parameters);
}
@Override
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/LocalSlicingWindowAggOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/LocalSlicingWindowAggOperator.java
index 9e7f6f4e654..5086ad1451a 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/LocalSlicingWindowAggOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/LocalSlicingWindowAggOperator.java
@@ -21,7 +21,6 @@ package org.apache.flink.table.runtime.operators.aggregate.window;
import org.apache.flink.core.memory.ManagedMemoryUseCase;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.TimestampedCollector;
import org.apache.flink.streaming.api.watermark.Watermark;
@@ -81,7 +80,6 @@ public class LocalSlicingWindowAggOperator extends AbstractStreamOperator<RowDat
SliceAssigner sliceAssigner,
WindowBuffer.LocalFactory windowBufferFactory,
ZoneId shiftTimezone) {
- chainingStrategy = ChainingStrategy.ALWAYS;
this.keySelector = keySelector;
this.sliceAssigner = sliceAssigner;
this.windowInterval = sliceAssigner.getSliceEndInterval();
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/bundle/AbstractMapBundleOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/bundle/AbstractMapBundleOperator.java
index 523594da292..7f7bfaf86d6 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/bundle/AbstractMapBundleOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/bundle/AbstractMapBundleOperator.java
@@ -21,7 +21,6 @@ package org.apache.flink.table.runtime.operators.bundle;
import org.apache.flink.api.common.functions.util.FunctionUtils;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -71,7 +70,6 @@ public abstract class AbstractMapBundleOperator<K, V, IN, OUT> extends AbstractS
AbstractMapBundleOperator(
MapBundleFunction<K, V, IN, OUT> function, BundleTrigger<IN> bundleTrigger) {
- chainingStrategy = ChainingStrategy.ALWAYS;
this.function = checkNotNull(function, "function is null");
this.bundleTrigger = checkNotNull(bundleTrigger, "bundleTrigger is null");
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/dynamicfiltering/DynamicFilteringDataCollectorOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/dynamicfiltering/DynamicFilteringDataCollectorOperator.java
index 55c6b140ca7..52288e11b86 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/dynamicfiltering/DynamicFilteringDataCollectorOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/dynamicfiltering/DynamicFilteringDataCollectorOperator.java
@@ -27,6 +27,7 @@ import org.apache.flink.runtime.operators.coordination.OperatorEventGateway;
import org.apache.flink.runtime.source.event.SourceEventWrapper;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.connector.source.DynamicFilteringData;
import org.apache.flink.table.connector.source.DynamicFilteringEvent;
@@ -72,10 +73,12 @@ public class DynamicFilteringDataCollectorOperator extends AbstractStreamOperato
private transient FieldGetter[] fieldGetters;
public DynamicFilteringDataCollectorOperator(
+ StreamOperatorParameters<Object> parameters,
RowType dynamicFilteringFieldType,
List<Integer> dynamicFilteringFieldIndices,
long threshold,
OperatorEventGateway operatorEventGateway) {
+ super(parameters);
this.dynamicFilteringFieldType = checkNotNull(dynamicFilteringFieldType);
this.dynamicFilteringFieldIndices = checkNotNull(dynamicFilteringFieldIndices);
this.threshold = threshold;
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/dynamicfiltering/DynamicFilteringDataCollectorOperatorFactory.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/dynamicfiltering/DynamicFilteringDataCollectorOperatorFactory.java
index e9ad35e9c9b..852cc38b966 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/dynamicfiltering/DynamicFilteringDataCollectorOperatorFactory.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/dynamicfiltering/DynamicFilteringDataCollectorOperatorFactory.java
@@ -64,16 +64,12 @@ public class DynamicFilteringDataCollectorOperatorFactory
DynamicFilteringDataCollectorOperator operator =
new DynamicFilteringDataCollectorOperator(
+ parameters,
dynamicFilteringFieldType,
dynamicFilteringFieldIndices,
threshold,
operatorEventGateway);
- operator.setup(
- parameters.getContainingTask(),
- parameters.getStreamConfig(),
- parameters.getOutput());
-
// today's lunch is generics spaghetti
@SuppressWarnings("unchecked")
final T castedOperator = (T) operator;
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/BufferDataOverWindowOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/BufferDataOverWindowOperator.java
index 50301646260..d72f286119a 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/BufferDataOverWindowOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/BufferDataOverWindowOperator.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.runtime.operators.over;
import org.apache.flink.runtime.memory.MemoryManager;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.binary.BinaryRowData;
@@ -61,6 +62,19 @@ public class BufferDataOverWindowOperator extends TableStreamOperator<RowData>
OverWindowFrame[] overWindowFrames,
GeneratedRecordComparator genComparator,
boolean isRowAllInFixedPart) {
+ this(null, overWindowFrames, genComparator, isRowAllInFixedPart);
+ }
+
+ /**
+ * @param overWindowFrames the window frames belong to this operator.
+ * @param genComparator the generated sort which is used for generating the comparator among
+ */
+ public BufferDataOverWindowOperator(
+ StreamOperatorParameters<RowData> parameters,
+ OverWindowFrame[] overWindowFrames,
+ GeneratedRecordComparator genComparator,
+ boolean isRowAllInFixedPart) {
+ super(parameters);
this.overWindowFrames = overWindowFrames;
this.genComparator = genComparator;
this.isRowAllInFixedPart = isRowAllInFixedPart;
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/NonBufferOverWindowOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/NonBufferOverWindowOperator.java
index 592c98d6c7f..d1bc7c58fb2 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/NonBufferOverWindowOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/NonBufferOverWindowOperator.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.runtime.operators.over;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.utils.JoinedRowData;
@@ -59,6 +60,15 @@ public class NonBufferOverWindowOperator extends TableStreamOperator<RowData>
GeneratedAggsHandleFunction[] aggsHandlers,
GeneratedRecordComparator genComparator,
boolean[] resetAccumulators) {
+ this(null, aggsHandlers, genComparator, resetAccumulators);
+ }
+
+ public NonBufferOverWindowOperator(
+ StreamOperatorParameters<RowData> parameters,
+ GeneratedAggsHandleFunction[] aggsHandlers,
+ GeneratedRecordComparator genComparator,
+ boolean[] resetAccumulators) {
+ super(parameters);
this.aggsHandlers = aggsHandlers;
this.genComparator = genComparator;
this.resetAccumulators = resetAccumulators;
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/SinkOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/SinkOperator.java
index 3c9d0b7f51e..d78489cbd94 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/SinkOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/SinkOperator.java
@@ -20,7 +20,6 @@ package org.apache.flink.table.runtime.operators.sink;
import org.apache.flink.streaming.api.functions.sink.legacy.SinkFunction;
import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.InternalTimerService;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.StreamOperator;
@@ -50,7 +49,6 @@ public class SinkOperator extends AbstractUdfStreamOperator<Object, SinkFunction
public SinkOperator(SinkFunction<RowData> sinkFunction, int rowtimeFieldIndex) {
super(sinkFunction);
this.rowtimeFieldIndex = rowtimeFieldIndex;
- chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sort/BaseTemporalSortOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sort/BaseTemporalSortOperator.java
index 185441dac5b..355a246e9f4 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sort/BaseTemporalSortOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sort/BaseTemporalSortOperator.java
@@ -23,7 +23,6 @@ import org.apache.flink.runtime.state.VoidNamespaceSerializer;
import org.apache.flink.streaming.api.SimpleTimerService;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.InternalTimerService;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.TimestampedCollector;
@@ -37,9 +36,7 @@ abstract class BaseTemporalSortOperator extends AbstractStreamOperator<RowData>
protected transient TimerService timerService;
protected transient TimestampedCollector<RowData> collector;
- BaseTemporalSortOperator() {
- setChainingStrategy(ChainingStrategy.ALWAYS);
- }
+ BaseTemporalSortOperator() {}
@Override
public void open() throws Exception {
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/groupwindow/operator/WindowOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/groupwindow/operator/WindowOperator.java
index 8e7c26c67d4..8e910375374 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/groupwindow/operator/WindowOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/groupwindow/operator/WindowOperator.java
@@ -33,7 +33,6 @@ import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.state.internal.InternalMergingState;
import org.apache.flink.runtime.state.internal.InternalValueState;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.InternalTimer;
import org.apache.flink.streaming.api.operators.InternalTimerService;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
@@ -207,8 +206,6 @@ public abstract class WindowOperator<K, W extends Window> extends AbstractStream
this.rowtimeIndex = rowtimeIndex;
this.shiftTimeZone = shiftTimeZone;
this.recordCounter = RecordCounter.of(inputCountIndex);
-
- setChainingStrategy(ChainingStrategy.ALWAYS);
}
WindowOperator(
@@ -240,8 +237,6 @@ public abstract class WindowOperator<K, W extends Window> extends AbstractStream
this.rowtimeIndex = rowtimeIndex;
this.shiftTimeZone = shiftTimeZone;
this.recordCounter = RecordCounter.of(inputCountIndex);
-
- setChainingStrategy(ChainingStrategy.ALWAYS);
}
protected abstract void compileGeneratedCode();
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/tvf/common/WindowAggOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/tvf/common/WindowAggOperator.java
index b94c1b0f806..26c941cf8b3 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/tvf/common/WindowAggOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/tvf/common/WindowAggOperator.java
@@ -32,7 +32,6 @@ import org.apache.flink.runtime.memory.MemoryManager;
import org.apache.flink.runtime.state.KeyedStateBackend;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContext;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.InternalTimer;
import org.apache.flink.streaming.api.operators.InternalTimerService;
import org.apache.flink.streaming.api.operators.KeyContext;
@@ -132,7 +131,6 @@ public final class WindowAggOperator<K, W> extends TableStreamOperator<RowData>
public WindowAggOperator(WindowProcessor<W> windowProcessor, boolean isEventTime) {
this.windowProcessor = windowProcessor;
this.isEventTime = isEventTime;
- setChainingStrategy(ChainingStrategy.ALWAYS);
}
@Override
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/tvf/operator/WindowTableFunctionOperatorBase.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/tvf/operator/WindowTableFunctionOperatorBase.java
index 8f8fafbe356..0e78ad72bb3 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/tvf/operator/WindowTableFunctionOperatorBase.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/tvf/operator/WindowTableFunctionOperatorBase.java
@@ -20,7 +20,6 @@ package org.apache.flink.table.runtime.operators.window.tvf.operator;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.metrics.Counter;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.TimestampedCollector;
import org.apache.flink.table.data.GenericRowData;
@@ -79,8 +78,6 @@ public abstract class WindowTableFunctionOperatorBase extends TableStreamOperato
this.rowtimeIndex = rowtimeIndex;
this.windowAssigner = windowAssigner;
checkArgument(!windowAssigner.isEventTime() || rowtimeIndex >= 0);
-
- setChainingStrategy(ChainingStrategy.ALWAYS);
}
@Override
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/ProcTimeMiniBatchAssignerOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/ProcTimeMiniBatchAssignerOperator.java
index 5a81bea6660..1fcb490c526 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/ProcTimeMiniBatchAssignerOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/ProcTimeMiniBatchAssignerOperator.java
@@ -22,7 +22,6 @@ import org.apache.flink.api.common.operators.ProcessingTimeService;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -52,7 +51,6 @@ public class ProcTimeMiniBatchAssignerOperator extends AbstractStreamOperator<Ro
public ProcTimeMiniBatchAssignerOperator(long intervalMs) {
this.intervalMs = intervalMs;
- this.chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/RowTimeMiniBatchAssginerOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/RowTimeMiniBatchAssginerOperator.java
index d3e54980158..079eec0b21e 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/RowTimeMiniBatchAssginerOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/RowTimeMiniBatchAssginerOperator.java
@@ -19,7 +19,6 @@
package org.apache.flink.table.runtime.operators.wmassigners;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -58,7 +57,6 @@ public class RowTimeMiniBatchAssginerOperator extends AbstractStreamOperator<Row
public RowTimeMiniBatchAssginerOperator(long minibatchInterval) {
this.minibatchInterval = minibatchInterval;
- this.chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperator.java
index 486ee5274e9..6548608e3e4 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperator.java
@@ -22,8 +22,8 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.functions.DefaultOpenContext;
import org.apache.flink.api.common.functions.util.FunctionUtils;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.util.PausableRelativeClock;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -82,16 +82,16 @@ public class WatermarkAssignerOperator extends AbstractStreamOperator<RowData>
* @param idleTimeout (idleness checking timeout)
*/
public WatermarkAssignerOperator(
+ StreamOperatorParameters<RowData> parameters,
int rowtimeFieldIndex,
WatermarkGenerator watermarkGenerator,
long idleTimeout,
ProcessingTimeService processingTimeService) {
-
+ super(parameters);
this.rowtimeFieldIndex = rowtimeFieldIndex;
this.watermarkGenerator = watermarkGenerator;
this.idleTimeout = idleTimeout;
- this.chainingStrategy = ChainingStrategy.ALWAYS;
this.processingTimeService = checkNotNull(processingTimeService);
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorFactory.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorFactory.java
index 45287177794..2a7d6c13e7a 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorFactory.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorFactory.java
@@ -53,14 +53,12 @@ public class WatermarkAssignerOperatorFactory extends AbstractStreamOperatorFact
WatermarkGenerator watermarkGenerator =
generatedWatermarkGenerator.newInstance(
initializer.getContainingTask().getUserCodeClassLoader());
- WatermarkAssignerOperator operator =
- new WatermarkAssignerOperator(
- rowtimeFieldIndex, watermarkGenerator, idleTimeout, processingTimeService);
- operator.setup(
- initializer.getContainingTask(),
- initializer.getStreamConfig(),
- initializer.getOutput());
- return operator;
+ return new WatermarkAssignerOperator(
+ initializer,
+ rowtimeFieldIndex,
+ watermarkGenerator,
+ idleTimeout,
+ processingTimeService);
}
@Override
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/dynamicfiltering/DynamicFilteringDataCollectorOperatorTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/dynamicfiltering/DynamicFilteringDataCollectorOperatorTest.java
index 0fc6a9b8b97..c813d400fc9 100644
--- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/dynamicfiltering/DynamicFilteringDataCollectorOperatorTest.java
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/dynamicfiltering/DynamicFilteringDataCollectorOperatorTest.java
@@ -51,7 +51,7 @@ class DynamicFilteringDataCollectorOperatorTest {
MockOperatorEventGateway gateway = new MockOperatorEventGateway();
DynamicFilteringDataCollectorOperator operator =
- new DynamicFilteringDataCollectorOperator(rowType, indexes, -1, gateway);
+ new DynamicFilteringDataCollectorOperator(null, rowType, indexes, -1, gateway);
ConcurrentLinkedQueue<Object> output;
try (OneInputStreamOperatorTestHarness<RowData, Object> harness =
new OneInputStreamOperatorTestHarness<>(operator)) {
@@ -91,7 +91,8 @@ class DynamicFilteringDataCollectorOperatorTest {
int thresholds = 100;
DynamicFilteringDataCollectorOperator operator =
- new DynamicFilteringDataCollectorOperator(rowType, indexes, thresholds, gateway);
+ new DynamicFilteringDataCollectorOperator(
+ null, rowType, indexes, thresholds, gateway);
try (OneInputStreamOperatorTestHarness<RowData, Object> harness =
new OneInputStreamOperatorTestHarness<>(operator)) {
harness.setup();
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/MultipleInputTestBase.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/MultipleInputTestBase.java
index 90f5d324e37..e904ab0065c 100644
--- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/MultipleInputTestBase.java
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/MultipleInputTestBase.java
@@ -64,15 +64,25 @@ public class MultipleInputTestBase {
}
protected TestingOneInputStreamOperator createOneInputStreamOperator() throws Exception {
- TestingOneInputStreamOperator op = new TestingOneInputStreamOperator();
- op.setup(createStreamTask(), createStreamConfig(), new BlackHoleOutput());
- return op;
+ return new TestingOneInputStreamOperator(
+ new StreamOperatorParameters<>(
+ createStreamTask(),
+ createStreamConfig(),
+ new BlackHoleOutput(),
+ TestProcessingTimeService::new,
+ null,
+ null));
}
protected TestingTwoInputStreamOperator createTwoInputStreamOperator() throws Exception {
- TestingTwoInputStreamOperator op = new TestingTwoInputStreamOperator();
- op.setup(createStreamTask(), createStreamConfig(), new BlackHoleOutput());
- return op;
+ return new TestingTwoInputStreamOperator(
+ new StreamOperatorParameters<>(
+ createStreamTask(),
+ createStreamConfig(),
+ new BlackHoleOutput(),
+ TestProcessingTimeService::new,
+ null,
+ null));
}
protected OneInputTransformation<RowData, RowData> createOneInputTransform(
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/TestingOneInputStreamOperator.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/TestingOneInputStreamOperator.java
index c570db9ccbf..72ff8646442 100644
--- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/TestingOneInputStreamOperator.java
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/TestingOneInputStreamOperator.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.runtime.operators.multipleinput;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -45,10 +46,20 @@ public class TestingOneInputStreamOperator extends AbstractStreamOperator<RowDat
private final List<StreamRecord<RowData>> receivedElements = new ArrayList<>();
public TestingOneInputStreamOperator() {
- this(false);
+ this(null, false);
+ }
+
+ public TestingOneInputStreamOperator(StreamOperatorParameters<RowData> parameters) {
+ this(parameters, false);
}
public TestingOneInputStreamOperator(boolean emitDataInEndInput) {
+ this(null, emitDataInEndInput);
+ }
+
+ public TestingOneInputStreamOperator(
+ StreamOperatorParameters<RowData> parameters, boolean emitDataInEndInput) {
+ super(parameters);
this.emitDataInEndInput = emitDataInEndInput;
}
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/TestingTwoInputStreamOperator.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/TestingTwoInputStreamOperator.java
index 252fa9abdab..85fb0d58682 100644
--- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/TestingTwoInputStreamOperator.java
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/TestingTwoInputStreamOperator.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.runtime.operators.multipleinput;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedMultiInput;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
@@ -51,7 +52,17 @@ public class TestingTwoInputStreamOperator extends AbstractStreamOperator<RowDat
this(false);
}
+ public TestingTwoInputStreamOperator(StreamOperatorParameters<RowData> parameters) {
+ this(parameters, false);
+ }
+
public TestingTwoInputStreamOperator(boolean emitDataInEndInput) {
+ this(null, emitDataInEndInput);
+ }
+
+ public TestingTwoInputStreamOperator(
+ StreamOperatorParameters<RowData> parameters, boolean emitDataInEndInput) {
+ super(parameters);
this.emitDataInEndInput = emitDataInEndInput;
}
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/BufferDataOverWindowOperatorTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/BufferDataOverWindowOperatorTest.java
index 42c32620b54..280755d6362 100644
--- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/BufferDataOverWindowOperatorTest.java
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/BufferDataOverWindowOperatorTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.runtime.operators.testutils.MockEnvironment;
import org.apache.flink.runtime.operators.testutils.MockEnvironmentBuilder;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.StreamTask;
@@ -204,25 +205,30 @@ class BufferDataOverWindowOperatorTest {
.thenReturn(0.99);
when(streamConfig.getOperatorID()).thenReturn(new OperatorID());
operator =
- new BufferDataOverWindowOperator(frames, comparator, true) {
+ new BufferDataOverWindowOperator(
+ new StreamOperatorParameters<>(
+ task,
+ streamConfig,
+ new NonBufferOverWindowOperatorTest.ConsumerOutput(
+ r ->
+ collect.add(
+ GenericRowData.of(
+ r.getInt(0),
+ r.getLong(1),
+ r.getLong(2),
+ r.getLong(3),
+ r.getLong(4)))),
+ TestProcessingTimeService::new,
+ null,
+ null),
+ frames,
+ comparator,
+ true) {
@Override
public StreamingRuntimeContext getRuntimeContext() {
return mock(StreamingRuntimeContext.class);
}
};
- operator.setProcessingTimeService(new TestProcessingTimeService());
- operator.setup(
- task,
- streamConfig,
- new NonBufferOverWindowOperatorTest.ConsumerOutput(
- r ->
- collect.add(
- GenericRowData.of(
- r.getInt(0),
- r.getLong(1),
- r.getLong(2),
- r.getLong(3),
- r.getLong(4)))));
operator.open();
addRow(0, 1L, 4L); /* 1 **/
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonBufferOverWindowOperatorTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonBufferOverWindowOperatorTest.java
index 301ae10668a..8c1ecce8fb3 100644
--- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonBufferOverWindowOperatorTest.java
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonBufferOverWindowOperatorTest.java
@@ -26,6 +26,7 @@ import org.apache.flink.runtime.operators.testutils.MockEnvironmentBuilder;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.Output;
import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
@@ -146,24 +147,29 @@ class NonBufferOverWindowOperatorTest {
.thenReturn(0.99);
when(streamConfig.getOperatorID()).thenReturn(new OperatorID());
operator =
- new NonBufferOverWindowOperator(functions, comparator, resetAccumulators) {
+ new NonBufferOverWindowOperator(
+ new StreamOperatorParameters<>(
+ task,
+ streamConfig,
+ new ConsumerOutput(
+ r ->
+ collect.add(
+ GenericRowData.of(
+ r.getInt(0),
+ r.getLong(1),
+ r.getLong(2),
+ r.getLong(3),
+ r.getLong(4)))),
+ TestProcessingTimeService::new,
+ null,
+ null),
+ functions,
+ comparator,
+ resetAccumulators) {
public StreamingRuntimeContext getRuntimeContext() {
return mock(StreamingRuntimeContext.class);
}
};
- operator.setProcessingTimeService(new TestProcessingTimeService());
- operator.setup(
- task,
- streamConfig,
- new ConsumerOutput(
- r ->
- collect.add(
- GenericRowData.of(
- r.getInt(0),
- r.getLong(1),
- r.getLong(2),
- r.getLong(3),
- r.getLong(4)))));
operator.open();
addRow(0, 1L, 4L);
addRow(0, 1L, 1L);
diff --git a/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/graph/OneInputTestStreamOperator.java b/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/graph/OneInputTestStreamOperator.java
index a4d15a25c44..6cbbc803aeb 100644
--- a/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/graph/OneInputTestStreamOperator.java
+++ b/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/graph/OneInputTestStreamOperator.java
@@ -36,6 +36,7 @@ import org.apache.flink.runtime.state.StateSnapshotContext;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -67,7 +68,11 @@ class OneInputTestStreamOperator extends AbstractStreamOperator<TestDataElement>
private transient ListState<String> state;
OneInputTestStreamOperator(
- String operatorID, TestEventQueue eventQueue, TestCommandDispatcher dispatcher) {
+ StreamOperatorParameters<TestDataElement> parameters,
+ String operatorID,
+ TestEventQueue eventQueue,
+ TestCommandDispatcher dispatcher) {
+ super(parameters);
this.operatorID = operatorID;
this.eventQueue = eventQueue;
this.receivedCommands = new HashSet<>();
diff --git a/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/graph/OneInputTestStreamOperatorFactory.java b/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/graph/OneInputTestStreamOperatorFactory.java
index b0e36594bfb..f71c8d4011e 100644
--- a/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/graph/OneInputTestStreamOperatorFactory.java
+++ b/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/graph/OneInputTestStreamOperatorFactory.java
@@ -48,12 +48,8 @@ public class OneInputTestStreamOperatorFactory
public <T extends StreamOperator<TestDataElement>> T createStreamOperator(
StreamOperatorParameters<TestDataElement> parameters) {
OneInputTestStreamOperator operator =
- new OneInputTestStreamOperator(operatorID, eventQueue, commandDispatcher);
- operator.setup(
- parameters.getContainingTask(),
- parameters.getStreamConfig(),
- parameters.getOutput());
- operator.setProcessingTimeService(processingTimeService);
+ new OneInputTestStreamOperator(
+ parameters, operatorID, eventQueue, commandDispatcher);
return (T) operator;
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
index 9b35af274be..07c1a0022c5 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
@@ -204,8 +204,7 @@ public class SavepointITCase extends TestLogger {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
- BoundedPassThroughOperator<Long> operator =
- new BoundedPassThroughOperator<>(ChainingStrategy.ALWAYS);
+ BoundedPassThroughOperator<Long> operator = new BoundedPassThroughOperator<>();
DataStream<Long> stream =
env.fromSequence(0, Long.MAX_VALUE)
.transform("pass-through", BasicTypeInfo.LONG_TYPE_INFO, operator);
@@ -851,9 +850,7 @@ public class SavepointITCase extends TestLogger {
private transient boolean processed;
- BoundedPassThroughOperator(ChainingStrategy chainingStrategy) {
- this.chainingStrategy = chainingStrategy;
- }
+ BoundedPassThroughOperator() {}
private static void allowSnapshots() {
snapshotAllowedLatch.countDown();
@@ -913,8 +910,7 @@ public class SavepointITCase extends TestLogger {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
- BoundedPassThroughOperator<Integer> operator =
- new BoundedPassThroughOperator<>(chainingStrategy);
+ BoundedPassThroughOperator<Integer> operator = new BoundedPassThroughOperator<>();
DataStream<Integer> stream =
env.addSource(new InfiniteTestSource())
.transform("pass-through", BasicTypeInfo.INT_TYPE_INFO, operator);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/scheduling/JMFailoverITCase.java b/flink-tests/src/test/java/org/apache/flink/test/scheduling/JMFailoverITCase.java
index 6905a8d4a8f..add680b5ca4 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/scheduling/JMFailoverITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/scheduling/JMFailoverITCase.java
@@ -59,7 +59,6 @@ import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.Output;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -528,12 +527,10 @@ class JMFailoverITCase {
public SourceTail() {
super();
- // chain with source.
- setChainingStrategy(ChainingStrategy.ALWAYS);
}
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<Long>> output) {
@@ -595,7 +592,7 @@ class JMFailoverITCase {
public static Map<Integer, Integer> attemptIds = new ConcurrentHashMap<>();
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<Tuple2<Integer, Integer>>> output) {
@@ -648,7 +645,7 @@ class JMFailoverITCase {
public static Map<Integer, Integer> attemptIds = new ConcurrentHashMap<>();
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<Tuple2<Integer, Integer>>> output) {
@@ -704,7 +701,7 @@ class JMFailoverITCase {
public static Map<Integer, Integer> countResults = new ConcurrentHashMap<>();
@Override
- public void setup(
+ protected void setup(
StreamTask<?, ?> containingTask,
StreamConfig config,
Output<StreamRecord<Void>> output) {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskSelectiveReadingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskSelectiveReadingITCase.java
index 0096a9290a9..8cc67fb76ae 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskSelectiveReadingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskSelectiveReadingITCase.java
@@ -24,7 +24,6 @@ import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.legacy.RichParallelSourceFunction;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.util.TestSequentialReadingStreamOperator;
import org.apache.flink.test.streaming.runtime.util.TestListResultSink;
@@ -58,7 +57,6 @@ public class StreamTaskSelectiveReadingITCase {
TestSequentialReadingStreamOperator twoInputStreamOperator =
new TestSequentialReadingStreamOperator("Operator0");
- twoInputStreamOperator.setChainingStrategy(ChainingStrategy.NEVER);
source0.connect(source1)
.transform(
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskTimerITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskTimerITCase.java
index 4d7f5bca549..1f20ea83b0a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskTimerITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskTimerITCase.java
@@ -26,7 +26,6 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.legacy.SourceFunction;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
@@ -75,10 +74,7 @@ public class StreamTaskTimerITCase extends AbstractTestBaseJUnit4 {
DataStream<String> source = env.addSource(new InfiniteTestSource());
- source.transform(
- "Custom Operator",
- BasicTypeInfo.STRING_TYPE_INFO,
- new TimerOperator(ChainingStrategy.ALWAYS));
+ source.transform("Custom Operator", BasicTypeInfo.STRING_TYPE_INFO, new TimerOperator());
try {
env.execute("Timer test");
@@ -115,10 +111,7 @@ public class StreamTaskTimerITCase extends AbstractTestBaseJUnit4 {
DataStream<String> source = env.addSource(new InfiniteTestSource());
- source.transform(
- "Custom Operator",
- BasicTypeInfo.STRING_TYPE_INFO,
- new TimerOperator(ChainingStrategy.NEVER));
+ source.transform("Custom Operator", BasicTypeInfo.STRING_TYPE_INFO, new TimerOperator());
try {
env.execute("Timer test");
@@ -139,7 +132,7 @@ public class StreamTaskTimerITCase extends AbstractTestBaseJUnit4 {
.transform(
"Custom Operator",
BasicTypeInfo.STRING_TYPE_INFO,
- new TwoInputTimerOperator(ChainingStrategy.NEVER));
+ new TwoInputTimerOperator());
try {
env.execute("Timer test");
@@ -159,9 +152,7 @@ public class StreamTaskTimerITCase extends AbstractTestBaseJUnit4 {
private Semaphore semaphore = new Semaphore(1);
- public TimerOperator(ChainingStrategy chainingStrategy) {
- setChainingStrategy(chainingStrategy);
- }
+ public TimerOperator() {}
@Override
public void processElement(StreamRecord<String> element) throws Exception {
@@ -219,9 +210,7 @@ public class StreamTaskTimerITCase extends AbstractTestBaseJUnit4 {
private Semaphore semaphore = new Semaphore(1);
- public TwoInputTimerOperator(ChainingStrategy chainingStrategy) {
- setChainingStrategy(chainingStrategy);
- }
+ public TwoInputTimerOperator() {}
@Override
public void processElement1(StreamRecord<String> element) throws Exception {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/TimestampITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/TimestampITCase.java
index 5dd4eed738c..cb4799343f9 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/TimestampITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/TimestampITCase.java
@@ -49,7 +49,6 @@ import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
import org.apache.flink.streaming.api.functions.source.legacy.SourceFunction;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
@@ -767,7 +766,6 @@ public class TimestampITCase extends TestLogger {
private final boolean timestampsEnabled;
public CustomOperator(boolean timestampsEnabled) {
- setChainingStrategy(ChainingStrategy.ALWAYS);
this.timestampsEnabled = timestampsEnabled;
}
@@ -809,9 +807,7 @@ public class TimestampITCase extends TestLogger {
private static class TimestampCheckingOperator extends AbstractStreamOperator<Integer>
implements OneInputStreamOperator<Integer, Integer> {
- public TimestampCheckingOperator() {
- setChainingStrategy(ChainingStrategy.ALWAYS);
- }
+ public TimestampCheckingOperator() {}
@Override
public void processElement(StreamRecord<Integer> element) throws Exception {