You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2025/08/06 21:52:31 UTC
(flink) branch master updated: [FLINK-38199] Allow to specify output in AbstractStreamOperatorTestHarness
This is an automated email from the ASF dual-hosted git repository.
roman 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 6a31842a439 [FLINK-38199] Allow to specify output in AbstractStreamOperatorTestHarness
6a31842a439 is described below
commit 6a31842a4392683349ba465170b8c01e8ec5c555
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Wed Aug 6 18:10:40 2025 +0200
[FLINK-38199] Allow to specify output in AbstractStreamOperatorTestHarness
---
.../streaming/util/AbstractStreamOperatorTestHarness.java | 13 +++++++++++--
1 file changed, 11 insertions(+), 2 deletions(-)
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 ec032a61653..31a6019a50a 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
@@ -99,6 +99,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.function.BiConsumer;
+import java.util.function.Function;
import java.util.stream.Collectors;
import static org.apache.flink.streaming.api.operators.StreamOperatorUtils.setProcessingTimeService;
@@ -187,6 +188,9 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
private long restoredCheckpointId = 0;
+ private Function<TypeSerializer<OUT>, Output<StreamRecord<OUT>>> outputCreator =
+ MockOutput::new;
+
public AbstractStreamOperatorTestHarness(
StreamOperator<OUT> operator, int maxParallelism, int parallelism, int subtaskIndex)
throws Exception {
@@ -364,6 +368,11 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
}
}
+ public void setOutputCreator(
+ Function<TypeSerializer<OUT>, Output<StreamRecord<OUT>>> outputCreator) {
+ this.outputCreator = outputCreator;
+ }
+
public void setCheckpointStorage(CheckpointStorage storage) {
if (stateBackend instanceof CheckpointStorage) {
return;
@@ -469,7 +478,7 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
factory,
mockTask,
config,
- new MockOutput(outputSerializer),
+ outputCreator.apply(outputSerializer),
new OperatorEventDispatcherImpl(
this.getClass().getClassLoader(),
new NoOpTaskOperatorEventGateway()))
@@ -482,7 +491,7 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
(AbstractStreamOperator) operator,
mockTask,
config,
- new MockOutput(outputSerializer));
+ outputCreator.apply(outputSerializer));
}
}
setupCalled = true;