rkhachatryan commented on a change in pull request #12575:
URL: https://github.com/apache/flink/pull/12575#discussion_r439064498
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
##########
@@ -91,31 +91,25 @@
private final ThreadSafeUnaligner threadSafeUnaligner;
CheckpointBarrierUnaligner(
- int[] numberOfInputChannelsPerGate,
SubtaskCheckpointCoordinator checkpointCoordinator,
String taskName,
- AbstractInvokable toNotifyOnCheckpoint) {
+ AbstractInvokable toNotifyOnCheckpoint,
+ IndexedInputGate... inputGates) {
super(toNotifyOnCheckpoint);
this.taskName = taskName;
-
- final int numGates = numberOfInputChannelsPerGate.length;
-
- gateChannelOffsets = new int[numGates];
- for (int index = 1; index < numGates; index++) {
- gateChannelOffsets[index] = gateChannelOffsets[index -
1] + numberOfInputChannelsPerGate[index - 1];
- }
-
- final int totalNumChannels = gateChannelOffsets[numGates - 1] +
numberOfInputChannelsPerGate[numGates - 1];
- hasInflightBuffers = new boolean[totalNumChannels];
-
- channelInfos = IntStream.range(0, numGates)
- .mapToObj(gateIndex -> IntStream.range(0,
numberOfInputChannelsPerGate[gateIndex])
- .mapToObj(channelIndex -> new
InputChannelInfo(gateIndex, channelIndex)))
- .flatMap(Function.identity())
+ this.channelInfos = Arrays.stream(inputGates)
+ .flatMap(gate ->
gate.getChannels().stream().map(InputChannel::getChannelInfo))
.toArray(InputChannelInfo[]::new);
-
- threadSafeUnaligner = new ThreadSafeUnaligner(totalNumChannels,
checkNotNull(checkpointCoordinator), this);
+ hasInflightBuffers = new boolean[channelInfos.length];
+ threadSafeUnaligner = new
ThreadSafeUnaligner(channelInfos.length, checkNotNull(checkpointCoordinator),
this);
+
+ gateChannelOffsets = new int[inputGates.length];
+ int offset = 0;
+ for (final IndexedInputGate gate: inputGates) {
+ gateChannelOffsets[gate.getGateIndex()] = offset;
+ offset += gate.getNumberOfInputChannels();
+ }
Review comment:
I look at it from a bit different perspective.
`InputProcessorUtil.createCheckpointedInputGates` passes offsets to
`CheckpointedInputGate` which uses them to talk with `barrierHandler`:
```
// CheckpointedInputGate with offsetChannelIndex() inlined for clarity
barrierHandler.isBlocked(bufferOrEvent.getChannelIndex() +
channelIndexOffset)
barrierHandler.processBarrier(checkpointBarrier,
bufferOrEvent.getChannelIndex() + channelIndexOffset);
```
`CheckpointBarrierAligner.processBarrier` internally calls
`resumeConsumption` which unflattens index back:
```
private void resumeConsumption(int channelIndex) {
InputGate inputGate = channelIndexToInputGate[channelIndex];
inputGate.resumeConsumption(channelIndex -
inputGateToChannelIndexOffset.get(inputGate));
}
```
So, they **are** related.
Now, are they equal?
The fact that `CheckpointedInputGate` can use `UnionInputGate` is accounted
by
`UnionInputGate.getNumberOfInputChannels` which returns the total number of
input channels. So these offsets indeed are the same.
I'm more inclined now to use `gateId+channelId` pairs or `InputChannelInfo`
s :)
----------------------------------------------------------------
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]
