Myasuka commented on a change in pull request #12470:
URL: https://github.com/apache/flink/pull/12470#discussion_r435692788
##########
File path:
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
##########
@@ -969,6 +970,34 @@ public void testNotifyCheckpointOnClosedOperator() throws
Throwable {
assertEquals(true, operator.closed.get());
}
+ @Test
+ public void testFailToConfirmCheckpointCompleted() throws Exception {
+ testFailToConfirmCheckpointMessage(streamTask ->
streamTask.notifyCheckpointCompleteAsync(1L));
+ }
+
+ @Test
+ public void testFailToConfirmCheckpointAborted() throws Exception {
+ testFailToConfirmCheckpointMessage(streamTask ->
streamTask.notifyCheckpointAbortAsync(1L));
+ }
+
+ private void testFailToConfirmCheckpointMessage(Consumer<StreamTask<?,
?>> consumer) throws Exception {
+ FailOnNotifyCheckpointOperator<Integer> operator = new
FailOnNotifyCheckpointOperator<>();
+ MultipleInputStreamTaskTestHarnessBuilder<Integer> builder =
+ new
MultipleInputStreamTaskTestHarnessBuilder<>(OneInputStreamTask::new,
BasicTypeInfo.INT_TYPE_INFO)
+ .addInput(BasicTypeInfo.INT_TYPE_INFO);
+ StreamTaskMailboxTestHarness<Integer> harness = builder
+ .setupOutputForSingletonOperatorChain(operator)
+ .build();
+
+ try {
+ consumer.accept(harness.streamTask);
+ harness.streamTask.runMailboxStep();
+ fail();
+ } catch (ExpectedTestException expected) {
+ // expected exception
Review comment:
Sure, I run the test locally without my fix on `StreamTask` and the test
failed as expected.
----------------------------------------------------------------
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]
