rkhachatryan commented on a change in pull request #14057:
URL: https://github.com/apache/flink/pull/14057#discussion_r543283061
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AlternatingController.java
##########
@@ -146,28 +146,7 @@ private void switchToUnaligned(
@Override
public Optional<CheckpointBarrier>
postProcessLastBarrier(InputChannelInfo channelInfo, CheckpointBarrier barrier)
throws IOException, CheckpointException {
- Optional<CheckpointBarrier> maybeTimeOut = asTimedOut(barrier);
- if (maybeTimeOut.isPresent() && activeController ==
alignedController) {
- switchToUnaligned(channelInfo, maybeTimeOut.get());
- checkState(activeController == unalignedController);
-
checkState(!activeController.postProcessLastBarrier(channelInfo,
maybeTimeOut.orElse(barrier)).isPresent());
- return maybeTimeOut;
- }
-
- barrier = maybeTimeOut.orElse(barrier);
- if (barrier.getCheckpointOptions().isUnalignedCheckpoint()) {
- checkState(activeController == unalignedController);
-
checkState(!activeController.postProcessLastBarrier(channelInfo,
maybeTimeOut.orElse(barrier)).isPresent());
- return Optional.empty();
- }
- else {
- checkState(activeController == alignedController);
- Optional<CheckpointBarrier> triggerResult =
activeController.postProcessLastBarrier(
- channelInfo,
- barrier);
- checkState(triggerResult.isPresent());
- return triggerResult;
- }
Review comment:
Isn't it enough to just convert the barrier upon triggering checkpoint
in subtask?
This is what I've done in `[FLINK-19681][checkpointing] Use converted
barrier after disabling alignment`. It is needed anyways for correct handling
in `SubtaskCheckpointCoordinatorImpl`.
----------------------------------------------------------------
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]
