1996fanrui opened a new pull request, #28652:
URL: https://github.com/apache/flink/pull/28652
## What is the purpose of the change
Part 2 of a 7-PR series implementing FLINK-38544 (checkpointing during
recovery of unaligned-checkpoint state).
**Depends on #28651 ([FLINK-39520]).** This branch contains the commits of
#28651; please review only the 8 commits prefixed `[FLINK-39521]`.
This PR teaches the physical Local/RemoteInputChannel to be created directly
in a recovery state and to receive recovered buffers via a new
`RecoverableInputChannel` interface; recovered-channel conversion hands buffers
over through that interface instead of migrating them via constructors. All new
paths are gated on `needsRecovery`/`inRecovery` and nothing in this PR passes
`needsRecovery=true`, so default behavior is unchanged. The v1
conversion-migration path is retired here; the (experimental, default-false)
`execution.checkpointing.during-recovery.enabled` flag is therefore temporarily
removed from ITCase randomization while the flag-on path is being rewired
(restored later in the series).
Note: `NettyMessage.PartitionRequest` gains a `needsRecovery` boolean — a
wire-format change (+1 byte per message), safe under the single-version-cluster
assumption of the TaskManager network protocol.
## Brief change log
- Temporarily remove `execution.checkpointing.during-recovery.enabled` from
ITCase randomization; `@Disabled` `RecoveredStateFilteringLargeRecordITCase`
for the same window
- `BufferManager`: credit notification gated behind a new
`notifyInitiallyEnabled` ctor param + `enableNotify()` (all existing callers
pass `true` — behavior unchanged)
- New `RecoverableInputChannel` contract; new recovery sentinels
`RecoveryCheckpointBarrier` / `EndOfFetchedChannelStateEvent` with dedicated
`EventSerializer` type tags; `getStateConsumedFuture()` unified on the
`InputChannel` base class
- `LocalInputChannel`: in-recovery state machine — recovered-buffer queue,
live-data hiding, upstream-readiness gating, recovery buffer lending,
`checkpointStarted` split into in-recovery / normal branches with
`collectPreRecoveryBarrier`
- `RemoteInputChannel`: same, with recovered buffers appended into
`receivedBuffers` (consume path identical to the normal case), ordinary
upstream events stashed under suppressed credit and replayed after recovery,
credit reopened on `onRecoveredStateConsumed`
- Recovered-channel conversion via the push interface
(`toInputChannel(boolean)`); `requestPartitions(needsRecovery)` threaded
through the gates; unified state-consumed-future aggregation
- `PartitionRequest` carries `needsRecovery`; the producer-side view reader
starts with zero credit while the consumer's exclusive buffers are lent to
recovery
- `CheckpointedInputGate` consumes `EndOfFetchedChannelStateEvent` (never
delivered to operators)
## Verifying this change
This change added tests and can be verified as follows:
- New/extended unit tests: `LocalInputChannelTest` /
`RemoteInputChannelTest` (recovery push, sentinels, barrier collection,
missing-barrier decline, persist window), `RecoveredInputChannelTest`,
`LocalRecoveredInputChannelTest`, `RemoteRecoveredInputChannelTest`,
`RecoveryCheckpointBarrierTest`, `InputChannelTest`, and the netty
serialization/handler test suites
- Default configuration is unchanged by construction: all new code paths are
gated on `needsRecovery`, which nothing sets to `true` in this PR (e.g.
`RemoteInputChannel#checkpointStarted`'s non-recovery branch is unchanged from
master)
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: no
- The serializers: no (two new internal runtime-event type tags in
`EventSerializer`; no type serializers touched)
- The runtime per-record code paths (performance sensitive): yes
(`Local/RemoteInputChannel#getNextBuffer` are restructured; the fast path is
unchanged when `needsRecovery=false`, which is always the case in this PR)
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes (channel-state
recovery; also a TaskManager network wire-format change: `PartitionRequest` +1
byte)
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Claude Fable 5)
--
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]
