This is an automated email from the ASF dual-hosted git repository. 1996fanrui pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 516a8f95b9f59a1f56c5c6ae1a6a4ad1d8cb1e17 Author: Rui Fan <[email protected]> AuthorDate: Mon Jul 6 02:30:44 2026 +0200 [FLINK-40081] Harden RecordFilterContext for minimal and batch environments The record-filter context used to filter recovered channel state during checkpointing-during-recovery made assumptions that do not hold in all environments. Make it robust to three edge cases: - Disabled context must not require an IOManager: a disabled context never spills, so it needs no spill directories. Don't touch the IOManager, so recovery works in minimal environments (e.g. DummyEnvironment-based tests) instead of NPEing. - Disabled context tolerates null/empty tmp directories for the same reason; disabled() gains an explicit String[] parameter, and non-empty tmpDirectories are enforced only when the flag is on. - No physical edges (AdaptiveBatchScheduler): a task can have an input gate whose physical connection -- and thus partitioner -- is decided by the scheduler at runtime (e.g. reading a cached intermediate result). Such jobs are batch and have no unaligned-checkpoint channel state to filter, so return a disabled context instead of asserting input-gates == physical-edges. --- .../runtime/io/recovery/RecordFilterContext.java | 18 ++++-- .../io/recovery/RecordFilterContextTest.java | 74 ++++++++++++++++++---- 2 files changed, 75 insertions(+), 17 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/recovery/RecordFilterContext.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/recovery/RecordFilterContext.java index e207eb6213e..ed2f584b3b3 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/recovery/RecordFilterContext.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/recovery/RecordFilterContext.java @@ -99,8 +99,8 @@ public class RecordFilterContext { /** * Network buffer memory segment size in bytes (from taskmanager.memory.segment-size). Used to - * size the reusable heap source buffer in {@code InputChannelRecoveredStateHandler} so it - * matches the network buffer size. + * size the reusable heap source buffer in {@code SpillingWithFilteringHandler} so it matches + * the network buffer size. */ private final int memorySegmentSize; @@ -112,8 +112,7 @@ public class RecordFilterContext { * @param rescalingDescriptor Descriptor containing rescaling information. Not null. * @param subtaskIndex Current subtask index. * @param maxParallelism Maximum parallelism. - * @param tmpDirectories Temporary directories for spilling spanning records. Can be null - * (converted to empty array). + * @param tmpDirectories Temporary directories for spilling spanning records. * @param checkpointingDuringRecoveryEnabled Whether unaligned checkpointing during recovery is * enabled. * @param memorySegmentSize Network buffer memory segment size in bytes. Must be positive. @@ -130,8 +129,17 @@ public class RecordFilterContext { this.rescalingDescriptor = checkNotNull(rescalingDescriptor); this.subtaskIndex = subtaskIndex; this.maxParallelism = maxParallelism; - this.tmpDirectories = tmpDirectories != null ? tmpDirectories : new String[0]; this.checkpointingDuringRecoveryEnabled = checkpointingDuringRecoveryEnabled; + if (checkpointingDuringRecoveryEnabled) { + // tmpDirectories are only used by the spilling (checkpointing-during-recovery) path. + checkArgument( + checkNotNull(tmpDirectories).length > 0, "tmpDirectories must not be empty"); + this.tmpDirectories = tmpDirectories.clone(); + } else { + // A disabled context never spills, so it needs no spill directories; tolerate a + // null/empty value (e.g. an environment without IOManager spilling directories). + this.tmpDirectories = tmpDirectories == null ? new String[0] : tmpDirectories.clone(); + } checkArgument( memorySegmentSize > 0, "memorySegmentSize must be positive: %s", memorySegmentSize); this.memorySegmentSize = memorySegmentSize; diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/recovery/RecordFilterContextTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/recovery/RecordFilterContextTest.java index 60494f7acbe..6ffd46ec471 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/recovery/RecordFilterContextTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/recovery/RecordFilterContextTest.java @@ -24,6 +24,9 @@ import org.apache.flink.runtime.memory.MemoryManager; import org.apache.flink.streaming.runtime.partitioner.ForwardPartitioner; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; import static org.apache.flink.runtime.checkpoint.InflightDataRescalingDescriptorUtil.mappings; import static org.apache.flink.runtime.checkpoint.InflightDataRescalingDescriptorUtil.rescalingDescriptor; @@ -34,11 +37,18 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for {@link RecordFilterContext}. */ class RecordFilterContextTest { + @TempDir private Path tempDir; + + private String[] tmpDirs() { + return new String[] {tempDir.toString()}; + } + @Test void testDisabledContextHasNoGates() { RecordFilterContext disabled = RecordFilterContext.disabled(); assertThat(disabled.getNumberOfGates()).isEqualTo(0); assertThat(disabled.isCheckpointingDuringRecoveryEnabled()).isFalse(); + assertThat(disabled.getTmpDirectories()).isEmpty(); } @Test @@ -53,7 +63,7 @@ class RecordFilterContextTest { InflightDataRescalingDescriptor.NO_RESCALE, 0, 128, - new String[] {"/tmp"}, + tmpDirs(), true, MemoryManager.DEFAULT_PAGE_SIZE); @@ -72,7 +82,7 @@ class RecordFilterContextTest { InflightDataRescalingDescriptor.NO_RESCALE, 0, 128, - null, + tmpDirs(), false, MemoryManager.DEFAULT_PAGE_SIZE); @@ -83,8 +93,38 @@ class RecordFilterContextTest { } @Test - void testNullTmpDirectoriesConvertedToEmptyArray() { - RecordFilterContext context = + void testEnabledContextRejectsNullOrEmptyTmpDirectories() { + // When checkpointing-during-recovery is enabled, the spilling path needs spill + // directories, so null/empty tmpDirectories are rejected. + assertThatThrownBy( + () -> + new RecordFilterContext( + new RecordFilterContext.InputFilterConfig[0], + InflightDataRescalingDescriptor.NO_RESCALE, + 0, + 128, + null, + true, + MemoryManager.DEFAULT_PAGE_SIZE)) + .isInstanceOf(NullPointerException.class); + assertThatThrownBy( + () -> + new RecordFilterContext( + new RecordFilterContext.InputFilterConfig[0], + InflightDataRescalingDescriptor.NO_RESCALE, + 0, + 128, + new String[0], + true, + MemoryManager.DEFAULT_PAGE_SIZE)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void testDisabledContextToleratesNullOrEmptyTmpDirectories() { + // A disabled context never spills, so it needs no spill directories: null/empty are + // tolerated and normalized to an empty array. + RecordFilterContext fromNull = new RecordFilterContext( new RecordFilterContext.InputFilterConfig[0], InflightDataRescalingDescriptor.NO_RESCALE, @@ -93,8 +133,18 @@ class RecordFilterContextTest { null, false, MemoryManager.DEFAULT_PAGE_SIZE); + assertThat(fromNull.getTmpDirectories()).isEmpty(); - assertThat(context.getTmpDirectories()).isNotNull().isEmpty(); + RecordFilterContext fromEmpty = + new RecordFilterContext( + new RecordFilterContext.InputFilterConfig[0], + InflightDataRescalingDescriptor.NO_RESCALE, + 0, + 128, + new String[0], + false, + MemoryManager.DEFAULT_PAGE_SIZE); + assertThat(fromEmpty.getTmpDirectories()).isEmpty(); } @Test @@ -111,7 +161,7 @@ class RecordFilterContextTest { descriptor, 0, 128, - null, + tmpDirs(), false, MemoryManager.DEFAULT_PAGE_SIZE); @@ -132,7 +182,7 @@ class RecordFilterContextTest { descriptor, 0, 128, - null, + tmpDirs(), true, MemoryManager.DEFAULT_PAGE_SIZE); @@ -152,7 +202,7 @@ class RecordFilterContextTest { descriptor, 0, 128, - null, + tmpDirs(), true, MemoryManager.DEFAULT_PAGE_SIZE); @@ -170,7 +220,7 @@ class RecordFilterContextTest { InflightDataRescalingDescriptor.NO_RESCALE, 0, 128, - null, + tmpDirs(), false, MemoryManager.DEFAULT_PAGE_SIZE * 2); @@ -184,7 +234,7 @@ class RecordFilterContextTest { InflightDataRescalingDescriptor.NO_RESCALE, 0, 128, - null, + tmpDirs(), false, 0)) .isInstanceOf(IllegalArgumentException.class); @@ -195,7 +245,7 @@ class RecordFilterContextTest { InflightDataRescalingDescriptor.NO_RESCALE, 0, 128, - null, + tmpDirs(), false, -1)) .isInstanceOf(IllegalArgumentException.class); @@ -227,7 +277,7 @@ class RecordFilterContextTest { InflightDataRescalingDescriptor.NO_RESCALE, 1, 256, - new String[] {"/tmp"}, + tmpDirs(), false, MemoryManager.DEFAULT_PAGE_SIZE);
