reswqa commented on code in PR #22432:
URL: https://github.com/apache/flink/pull/22432#discussion_r1188891342
##########
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)
+ .build()
+
.chain(BasicTypeInfo.INT_TYPE_INFO.createSerializer(new ExecutionConfig()))
+ .setOperatorFactory(SimpleOperatorFactory.of(new
DuplicatingOperator()))
+ .addNonChainedOutputsCount(1)
+ .build()
+ .finish()
+ .setTaskMetricGroup(taskMetricGroup)
+ .build()) {
+ Counter numRecordsInCounter =
+
taskMetricGroup.getIOMetricGroup().getNumRecordsInCounter();
+ Counter numRecordsOutCounter =
+
taskMetricGroup.getIOMetricGroup().getNumRecordsOutCounter();
+
+ final int numEvenRecords = 5;
+ final int numOddRecords = 3;
+
+ for (int x = 0; x < numEvenRecords; x++) {
+ testHarness.processElement(new StreamRecord<>(2 * x));
+ }
+
+ for (int x = 0; x < numOddRecords; x++) {
+ testHarness.processElement(new StreamRecord<>(2 * x + 1));
+ }
+ assertEquals(numOddRecords + numEvenRecords,
numRecordsInCounter.getCount());
+ assertEquals(
+ numOddRecords
+ + (numOddRecords + numEvenRecords)
+ + (numOddRecords + numEvenRecords) * 2,
+ numRecordsOutCounter.getCount());
Review Comment:
The topology graph of this test case is as follows:
- The first operator does not have `NonChainedOutput`.
- The second operator have two `RecordWriter With Odd Tag` as this
code:`.addNonChainedOutputsCount(new OutputTag<>("odd",
BasicTypeInfo.INT_TYPE_INFO), 2)`
The assertion value(`numOddRecords + (numOddRecords + numEvenRecords) +
(numOddRecords + numEvenRecords) * 2`) from the following three parts:
- `numOddRecords` from the second `OddEvenOperator`'s two
`RecordWriterOutput With Odd Tag`. It is only calculated once here, which is
guaranteed by the logic of `BroadcastingOutputCollector`.
- `numOddRecords + numEvenRecords` from the second `OddEvenOperator`'s
`RecordWriterOutput`.
- `(numOddRecords + numEvenRecords) * 2` from the `DuplicatingOperator`
It should be noted here that:
- `OddEvenOperator` will send duplicated data to the side output and normal
output. So we have the first two parts above.
- `ChainedOutput` will ignore the data with output tag as we don't set
`outputTag` for chaining `StreamEdge`. So subsequent operators will not receive
duplicate data.
```
public void processElement(StreamRecord<Integer> element) {
if (element.getValue() % 2 == 0) {
output.collect(evenOutputTag, element);
} else {
output.collect(oddOutputTag, element);
}
output.collect(element);
}
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
