pnowojski commented on a change in pull request #14831:
URL: https://github.com/apache/flink/pull/14831#discussion_r570945552
##########
File path:
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java
##########
@@ -961,6 +965,52 @@ public void testCheckpointBarrierMetrics() throws
Exception {
testHarness.waitForTaskCompletion();
}
+ @Test
+ public void testTriggerCheckpointViaRpcMessage() throws Exception {
+ // A global barrier handler to record the history of checkpoint
trigger.
+ TestCheckpointBarrierHandler barrierHandler =
+ new TestCheckpointBarrierHandler(new DummyInvokable());
+ OneInputStreamTaskTestHarness<String, String> testHarness =
+ new OneInputStreamTaskTestHarness<>(
+ (Environment env) ->
+ new OneInputStreamTask<String, String>(env) {
+ @Override
+ protected CheckpointBarrierHandler
+ getCheckpointBarrierHandler() {
+ return barrierHandler;
+ }
+ },
Review comment:
Am I missing something? But isn't it dangerous what you are doing here?
Basically in your test there will be existing two very different
`CheckpointBarrierHandler` instances. One real instance created in
`StreamTask#init` and passed to `CheckpointedInputGate` and then a completelly
different one `TestCheckpointBarrierHandler ` returned from
`getCheckpointBarrierHandler`?
Maybe you could hook in the testing code in for example `TestOperator()` and
track the triggered checkpoints from there?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
