pltbkd commented on PR #28523:
URL: https://github.com/apache/flink/pull/28523#issuecomment-4936027761
Here’s the review guide. I hope it helps with the code review process.
Please feel free to let me know here or on Slack if any further explanation is
necessary.
----------------
This design adds POINTWISE edge support to Unaligned Checkpoint while
preserving the existing ALL_TO_ALL recovery flow.
POINTWISE edges lack global connectivity, and local channel index does not
equal peer subtask index, so A2A recovery strategy cannot be reused.
The entire recovery process answers the following questions:
## 1. How to determine which old channel state a new task should inherit:
- Downstream uses standard ROUND_ROBIN assignment
> `TaskStateAssignment.java` — `computePointwiseOldSubtaskInstances`
(isInput=true) (:770)
> `TaskStateAssignment.java` —
`SubtaskStateMapper.ROUND_ROBIN.getOldSubtasks`
- Upstream needs to find connected downstreams, then read their upstream
states
> `PointwiseChannelMappingUtils.java` — `computeInheritedOldUpstreams`
(:112) /
> `TaskStateAssignment.java` — `computePointwiseOldSubtaskInstances`
(isInput=false) (:770)
- To find the corresponding upstreams, the original job's edge connectivity
must be recorded
> `EdgeDistributionPatternHook.java`
> `EdgeDistributionPatternSnapshot.java`
## 2. How to route recovered buffers to the correct position in the new
topology
- Downstream follows standard recovery path, using historical local channel
index to build virtual channels
> `RecoveredChannelStateHandler.java` —
`InputChannelRecoveredStateHandler.recover` standard branch (:198)
> `DemultiplexingRecordDeserializer.java:55` — VirtualChannel construction
- Upstream reads a channel's state, computes the target downstream using
old/new topology info, and constructs a channel descriptor
> `RecoveredChannelStateHandler.java` —
`ResultSubpartitionRecoveredStateHandler.recoverPointwise` (:434)
> `TaskStateAssignment.java` —
`distributePointwiseOutputBufferToDownstream` (:879)
- For n:1 connections, multiple upstreams read the same state; the "primary
upstream" (smallest-ID upstream) is responsible, others skip
> `PointwiseChannelMappingUtils.java` — `computeNewLocalSubpartitionIndex`
(:249)
> `PointwiseChannelMappingUtils.java` — `producersOf` (:288)
## 3. Handling of the downstream recovery mode
- Consistent with the upstream recovery approach, state assignment phase is
identical
- Buffer distribution algorithm is identical, with direct dedup; result must
guarantee each handle has exactly one downstream receiver
> `TaskStateAssignment.java` —
`distributePointwiseOutputBufferToDownstream` (:879)
> `RecoveredChannelStateHandler.java` —
`ResultSubpartitionRecoveredStateHandler.recoverPointwise` (:434)
## 4. Handling of unchanged parallelism
- If pattern is also unchanged, in-place recovery is used; primary upstream
dedup and other logic are skipped
> `TaskStateAssignment.java` — `getOutputMapping` identity check (:557)
> `TaskStateAssignment.java` — `createPointwiseRescalingDescriptor`
isIdentity (:751)
- If pattern has changed, full redistribution is required
> `TaskStateAssignment.java` — `hasPointwiseEdge` (:653)
> `StateAssignmentOperation.java` — `reDistributeResultSubpartitionStates`
hasPointwiseEdge skip early-return (:435)
## Other Modifications introduced
- Introduced EdgeDistributionPatternHook to record job edge patterns;
absence of this hook is treated as A2A, since Flink versions at that time only
supported A2A edges
- Introduced PointwiseChannelMappingUtils to share computation logic across
all chains, and to unify some tricky implementation issues (e.g., different
local index calculation for input vs output sides)
- Extended InflightDataRescalingDescriptor with PointwiseRescaleParams to
pass necessary information for TM-side computation
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
