pnowojski commented on a change in pull request #14797:
URL: https://github.com/apache/flink/pull/14797#discussion_r566976700
##########
File path:
flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java
##########
@@ -221,15 +228,17 @@ public InputStatus pollNext(ReaderOutput<Long> output)
throws InterruptedExcepti
output.collect(withHeader(split.nextNumber), split.nextNumber);
split.nextNumber += split.increment;
- if (throttle) {
- // throttle source as long as sink is not backpressuring
(which it does only
- // after full recovery)
- Thread.sleep(1);
+ switch (pollingState) {
+ case FINISHING:
+ return InputStatus.END_OF_INPUT;
+ case THROTTLING:
+ // throttle source as long as sink is not
backpressuring (which it does only
+ // after full recovery)
+ Thread.sleep(1);
+ return InputStatus.MORE_AVAILABLE;
+ default:
+ return InputStatus.MORE_AVAILABLE;
Review comment:
nitty nit: add `case PUMPING` and make `default: throw new
UnssuportedOperationException(pollingState);`?
----------------------------------------------------------------
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]
