pnowojski commented on code in PR #22432:
URL: https://github.com/apache/flink/pull/22432#discussion_r1189479094
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java:
##########
@@ -996,6 +1002,88 @@ public void testCanEmitBatchOfRecords() throws Exception {
}
}
+ @Test
+ public void testTaskSideOutputStatistics() throws Exception {
+ TaskMetricGroup taskMetricGroup =
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup();
+
+ ResultPartitionWriter[] partitionWriters = new
ResultPartitionWriter[3];
+ for (int i = 0; i < partitionWriters.length; ++i) {
+ partitionWriters[i] =
+ new RecordOrEventCollectingResultPartitionWriter<>(
+ new ArrayDeque<>(),
+ new StreamElementSerializer<>(
+
BasicTypeInfo.INT_TYPE_INFO.createSerializer(
+ new ExecutionConfig())));
+ partitionWriters[i].setup();
+ }
+
+ try (StreamTaskMailboxTestHarness<Integer> testHarness =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ OneInputStreamTask::new,
BasicTypeInfo.INT_TYPE_INFO)
+ .addInput(BasicTypeInfo.INT_TYPE_INFO)
+ .addAdditionalOutput(partitionWriters)
+ .setupOperatorChain(new OperatorID(), new
OddEvenOperator())
+
.chain(BasicTypeInfo.INT_TYPE_INFO.createSerializer(new ExecutionConfig()))
+ .setOperatorFactory(SimpleOperatorFactory.of(new
OddEvenOperator()))
+ .addNonChainedOutputsCount(
+ new OutputTag<>("odd",
BasicTypeInfo.INT_TYPE_INFO), 2)
+ .addNonChainedOutputsCount(1)
Review Comment:
ok, I get it know. It's fine as it is :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
