You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fa...@apache.org on 2026/08/14 00:55:40 UTC
(flink) 01/02: [FLINK-40081] Harden RecordFilterContext for minimal and batch environments
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 <19...@gmail.com>
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);