pnowojski commented on a change in pull request #14057:
URL: https://github.com/apache/flink/pull/14057#discussion_r540858619
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AlternatingController.java
##########
@@ -114,6 +114,7 @@ public void barrierAnnouncement(
lastSeenBarrier = barrier.getId();
firstBarrierArrivalTime = getArrivalTime(barrier);
}
+ activeController = chooseController(barrier);
Review comment:
But doesn't it mean we should support time outing on every UC barrier?
And it looks like we are doing that:
```
@Override
public Optional<CheckpointBarrier> barrierReceived(InputChannelInfo
channelInfo, CheckpointBarrier barrier) throws IOException, CheckpointException
{
if (barrier.getCheckpointOptions().isUnalignedCheckpoint() &&
activeController == alignedController) {
barrier = barrier.asUnaligned();
switchToUnaligned(channelInfo, barrier);
activeController.barrierReceived(channelInfo, barrier);
return Optional.of(barrier);
}
```
##########
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:
> Besides, why timeout alignment if it's the last barrier? This
essentially means that alignment is done.
The alignment is done, but that's just input. As we do not have code to
timeout outputs, it's better to timeout to UC even if the alignment was
completed (although too late). Think especially about a case with just a single
input channel.
----------------------------------------------------------------
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]
