You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "1996fanrui (via GitHub)" <gi...@apache.org> on 2026/04/08 16:11:48 UTC
[PR] [FLINK-38544] Add spill-to-disk for channel state recovery with filtering [flink]
1996fanrui opened a new pull request, #27905:
URL: https://github.com/apache/flink/pull/27905
## What is the purpose of the change
Add spill-to-disk infrastructure for channel state recovery during unaligned checkpoint recovery with filtering (FLINK-38544).
When unaligned checkpoint recovery runs with state filtering (rescaling), the current implementation holds all recovered buffers in memory (Network Buffer Pool), which can cause memory pressure and deadlocks. This PR introduces a spill-to-disk mechanism that:
1. **Decouples Source Buffers from Network Buffer Pool**: Uses heap-allocated buffers for reading S3 data (Source Buffers), avoiding competition with the Network Buffer Pool
2. **Three-path routing for filtered buffers**:
- **P1 (Memory)**: When no disk data exists, deliver filtered buffers directly
- **P2 (Spill)**: When Network Buffer Pool is exhausted, spill to local disk
- **P3 (Replay)**: When disk has data, spill current buffer and replay from disk (FIFO ordering)
3. **Phase 2 drain**: After all S3 data is read, drain remaining disk data using blocking buffer requests (safe because all heap Source Buffers are released)
### Core Components
- `SpillFileWriter`: Writes raw buffer bytes to spill files
- `SpillFileReader`: Reads raw buffer bytes back from spill files
- `SpillingBufferManager`: Manages spill/replay lifecycle with FIFO ordering, 64MB file rotation, reference counting, and old attempt cleanup
- `ChannelStateSerializer.wrapWithoutRecycle()`: Supports heap buffer lifecycle where caller manages recycling
### Integration Changes
- `InputChannelRecoveredStateHandler`: Updated with heap buffer allocation, three-path routing, and Phase 2 drain
- `SequentialChannelStateReaderImpl`: Passes spill directories and calls `drainDiskData()` after read phase
- `RecoveredInputChannel`: Changed `requestBufferBlocking` to non-blocking `requestBuffer`
## Brief change log
- Add `SpillFileWriter`, `SpillFileReader`, and `SpillingBufferManager` for spill-to-disk infrastructure
- Add `ChannelStateSerializer.wrapWithoutRecycle()` for filtering buffer lifecycle
- Update `InputChannelRecoveredStateHandler` with heap buffer allocation, three-path routing, and Phase 2 disk drain
- Update `SequentialChannelStateReaderImpl` to pass spill dirs and trigger Phase 2 drain
- Change `RecoveredInputChannel.requestBufferBlocking` to non-blocking `requestBuffer`
## Verifying this change
This change is verified by new unit tests:
- `SpillFileWriterReaderTest`: Covers spill file I/O, file rotation, large buffers
- `SpillingBufferManagerTest`: Covers replay ordering, checkpoint iterator, concurrent ref counting, FIFO guarantee
- `InputChannelRecoveredStateHandlerTest`: Covers heap buffer allocation, source buffer isolation, sequential channel processing, no-filtering compatibility
## 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
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: **yes** (recovery with state filtering)
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no (internal improvement to existing recovery mechanism)
- If yes, how is the feature documented? not applicable
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38544] Add spill-to-disk for channel state recovery with filtering [flink]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27905:
URL: https://github.com/apache/flink/pull/27905#issuecomment-4912260281
This PR is being marked as stale since it has not had any activity in the last 90 days.
If you would like to keep this PR alive, please leave a comment asking for a review.
If the PR has merge conflicts, update it with the latest from the base branch.
If you are having difficulty finding a reviewer, please reach out to the
community, contact details can be found here: https://flink.apache.org/what-is-flink/community/
If this PR is no longer valid or desired, please feel free to close it.
If no activity occurs in the next 30 days, it will be automatically closed.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38544] Add spill-to-disk for channel state recovery with filtering [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui closed pull request #27905: [FLINK-38544] Add spill-to-disk for channel state recovery with filtering
URL: https://github.com/apache/flink/pull/27905
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38544] Add spill-to-disk for channel state recovery with filtering [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #27905:
URL: https://github.com/apache/flink/pull/27905#issuecomment-4207765473
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1760f92ffc7b221e3f52bb4b6f2e36e9d0308479",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1760f92ffc7b221e3f52bb4b6f2e36e9d0308479",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1760f92ffc7b221e3f52bb4b6f2e36e9d0308479 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org