Skip to content

[FLINK-38930][checkpoint] Filtering record before processing without spilling strategy #27783

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
1996fanrui wants to merge 6 commits into
base: master
Choose a base branch
from
+1,598 −36
Open

[FLINK-38930][checkpoint] Filtering record before processing without spilling strategy #27783

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
90da75c
[hotfix] Rename isUnalignedDuringRecoveryEnabled to isCheckpointingDu…
1996fanrui Mar 26, 2026
e185b20
[FLINK-38930][checkpoint] Filtering record before processing without …
1996fanrui Feb 18, 2026
d24b3f3
Addressing comments: Reorder methods in ChannelStateFilteringHandler
1996fanrui Mar 27, 2026
684afcc
Address review comments
1996fanrui Mar 27, 2026
3e7418a
Address review comments: Stream filtering records instead of batch ac…
1996fanrui Mar 27, 2026
26602df
[FLINK-38930][checkpoint] Use heap buffer fallback to avoid deadlock …
1996fanrui Mar 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -800,11 +800,11 @@ public static boolean isUnalignedCheckpointInterruptibleTimersEnabled(Configurat
* execution.
*
* @param config the configuration to check
* @return {@code true} if unaligned checkpoint during recovery is enabled, {@code false}
* @return {@code true} if unaligned checkpointing during recovery is enabled, {@code false}
* otherwise
*/
@Internal
public static boolean isUnalignedDuringRecoveryEnabled(Configuration config) {
public static boolean isCheckpointingDuringRecoveryEnabled(Configuration config) {
if (!config.get(UNALIGNED_RECOVER_OUTPUT_ON_DOWNSTREAM)) {
return false;
}
Expand Down
14 changes: 9 additions & 5 deletions flink-core/src/test/java/org/apache/flink/configuration/CheckpointingOptionsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -333,14 +333,16 @@ void testIsUnalignedCheckpointInterruptibleTimersEnabled() {
void testIsUnalignedDuringRecoveryEnabled() {
// Test when both options are disabled (default) - should return false
Configuration defaultConfig = new Configuration();
assertThat(CheckpointingOptions.isUnalignedDuringRecoveryEnabled(defaultConfig))
assertThat(CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(defaultConfig))
.as("During-recovery should be disabled by default")
.isFalse();

// Test when during-recovery is enabled but recover-output-on-downstream is disabled
Configuration onlyDuringRecoveryConfig = new Configuration();
onlyDuringRecoveryConfig.set(CheckpointingOptions.UNALIGNED_DURING_RECOVERY_ENABLED, true);
assertThat(CheckpointingOptions.isUnalignedDuringRecoveryEnabled(onlyDuringRecoveryConfig))
assertThat(
CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(
onlyDuringRecoveryConfig))
.as(
"During-recovery should be disabled when recover-output-on-downstream is not enabled")
.isFalse();
Expand All @@ -350,7 +352,7 @@ void testIsUnalignedDuringRecoveryEnabled() {
onlyRecoverOnDownstreamConfig.set(
CheckpointingOptions.UNALIGNED_RECOVER_OUTPUT_ON_DOWNSTREAM, true);
assertThat(
CheckpointingOptions.isUnalignedDuringRecoveryEnabled(
CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(
onlyRecoverOnDownstreamConfig))
.as("During-recovery should be disabled when during-recovery option is not enabled")
.isFalse();
Expand All @@ -359,7 +361,7 @@ void testIsUnalignedDuringRecoveryEnabled() {
Configuration bothEnabledConfig = new Configuration();
bothEnabledConfig.set(CheckpointingOptions.UNALIGNED_RECOVER_OUTPUT_ON_DOWNSTREAM, true);
bothEnabledConfig.set(CheckpointingOptions.UNALIGNED_DURING_RECOVERY_ENABLED, true);
assertThat(CheckpointingOptions.isUnalignedDuringRecoveryEnabled(bothEnabledConfig))
assertThat(CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(bothEnabledConfig))
.as(
"During-recovery should be enabled when both recover-output-on-downstream and during-recovery are enabled")
.isTrue();
Expand All @@ -369,7 +371,9 @@ void testIsUnalignedDuringRecoveryEnabled() {
explicitlyDisabledConfig.set(
CheckpointingOptions.UNALIGNED_RECOVER_OUTPUT_ON_DOWNSTREAM, false);
explicitlyDisabledConfig.set(CheckpointingOptions.UNALIGNED_DURING_RECOVERY_ENABLED, true);
assertThat(CheckpointingOptions.isUnalignedDuringRecoveryEnabled(explicitlyDisabledConfig))
assertThat(
CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(
explicitlyDisabledConfig))
.as(
"During-recovery should be disabled when recover-output-on-downstream is explicitly false")
.isFalse();
Expand Down
Loading