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/06/22 10:53:44 UTC
(flink) branch master updated: [FLINK-39909][state] Fix heap state backend savepoint NPE with null map state values
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
The following commit(s) were added to refs/heads/master by this push:
new 9bc156b7faf [FLINK-39909][state] Fix heap state backend savepoint NPE with null map state values
9bc156b7faf is described below
commit 9bc156b7faf811ca6940a05e02df900ab9bf6627
Author: Mukul Gupta <mu...@ericsson.com>
AuthorDate: Thu Jun 11 06:49:36 2026 +0000
[FLINK-39909][state] Fix heap state backend savepoint NPE with null map state values
---
.../state/heap/HeapKeyValueStateIterator.java | 4 +-
.../MapStateNullValueCheckpointingITCase.java | 139 +++++++++++++++++++++
2 files changed, 142 insertions(+), 1 deletion(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyValueStateIterator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyValueStateIterator.java
index 80338a66dd3..b525e5f1e3e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyValueStateIterator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyValueStateIterator.java
@@ -362,7 +362,9 @@ public final class HeapKeyValueStateIterator implements KeyValueStateIterator {
compositeKeyBuilder.buildCompositeKeyUserKey(entry.getKey(), userKeySerializer);
Object userValue = entry.getValue();
valueOut.writeBoolean(userValue == null);
- userValueSerializer.serialize(userValue, valueOut);
+ if (userValue != null) {
+ userValueSerializer.serialize(userValue, valueOut);
+ }
currentValue = valueOut.getCopyOfBuffer();
if (!mapEntries.hasNext()) {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/MapStateNullValueCheckpointingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/MapStateNullValueCheckpointingITCase.java
index f5237534c86..a9ba3070806 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/MapStateNullValueCheckpointingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/MapStateNullValueCheckpointingITCase.java
@@ -114,6 +114,8 @@ class MapStateNullValueCheckpointingITCase {
StatefulMapper.firstRunFuture = new CompletableFuture<>();
StatefulMapper.secondRunFuture = new CompletableFuture<>();
+ NullUnsafeStatefulMapper.firstRunFuture = new CompletableFuture<>();
+ NullUnsafeStatefulMapper.secondRunFuture = new CompletableFuture<>();
}
@AfterEach
@@ -218,6 +220,143 @@ class MapStateNullValueCheckpointingITCase {
assertThat(restoredState).containsKey("null-key");
}
+ /**
+ * Tests that MapState with null values works correctly with null-unsafe serializers (e.g.,
+ * IntSerializer) during checkpoint/savepoint and restore. This verifies the fix in {@link
+ * org.apache.flink.runtime.state.heap.HeapKeyValueStateIterator} which previously would NPE
+ * when serializing null values during savepoint.
+ */
+ @TestTemplate
+ void testMapStateWithNullUnsafeSerializerCheckpointingAndRestore() throws Exception {
+ final String savepointPath = runJobWithNullUnsafeSerializer();
+ assertThat(savepointPath).isNotEmpty();
+ restoreAndVerifyNullUnsafeSerializer(savepointPath);
+ }
+
+ private String runJobWithNullUnsafeSerializer() throws Exception {
+ Configuration conf = new Configuration();
+ conf.set(
+ CheckpointingOptions.CHECKPOINTS_DIRECTORY,
+ TempDirUtils.newFolder(tmpFolder).toURI().toString());
+ conf.set(CheckpointingOptions.EXTERNALIZED_CHECKPOINT_RETENTION, RETAIN_ON_CANCELLATION);
+ conf.set(
+ CheckpointingOptions.SAVEPOINT_DIRECTORY,
+ TempDirUtils.newFolder(tmpFolder).toURI().toString());
+ conf.set(StateBackendOptions.STATE_BACKEND, stateBackend);
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
+ env.setParallelism(1);
+
+ env.fromSource(createSource(), WatermarkStrategy.noWatermarks(), "Data Generator Source")
+ .keyBy(v -> 0)
+ .map(new NullUnsafeStatefulMapper(true))
+ .sinkTo(new DiscardingSink<>());
+
+ JobGraph jobGraph = env.getStreamGraph().getJobGraph();
+ MiniCluster miniCluster = cluster.getMiniCluster();
+ miniCluster.submitJob(jobGraph).get();
+
+ JobID jobID = jobGraph.getJobID();
+ NullUnsafeStatefulMapper.firstRunFuture.get(2, TimeUnit.MINUTES);
+
+ if (snapshotType.isLeft()) {
+ cluster.getClusterClient()
+ .triggerCheckpoint(jobID, snapshotType.left())
+ .get(2, TimeUnit.MINUTES);
+ String checkpointPath =
+ CommonTestUtils.getLatestCompletedCheckpointPath(jobID, miniCluster)
+ .<NoSuchElementException>orElseThrow(
+ () ->
+ new NoSuchElementException(
+ "No checkpoint was created yet"));
+ cluster.getClusterClient().cancel(jobID);
+ return checkpointPath;
+ } else {
+ return cluster.getClusterClient()
+ .stopWithSavepoint(jobID, false, null, snapshotType.right())
+ .get(2, TimeUnit.MINUTES);
+ }
+ }
+
+ private void restoreAndVerifyNullUnsafeSerializer(String savepointPath) throws Exception {
+ Configuration conf = new Configuration();
+ conf.set(
+ CheckpointingOptions.CHECKPOINTS_DIRECTORY,
+ TempDirUtils.newFolder(tmpFolder).toURI().toString());
+ conf.set(
+ CheckpointingOptions.SAVEPOINT_DIRECTORY,
+ TempDirUtils.newFolder(tmpFolder).toURI().toString());
+ conf.set(StateBackendOptions.STATE_BACKEND, stateBackend);
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
+ env.setParallelism(1);
+
+ env.fromSource(createSource(), WatermarkStrategy.noWatermarks(), "Data Generator Source")
+ .keyBy(v -> 0)
+ .map(new NullUnsafeStatefulMapper(false))
+ .sinkTo(new DiscardingSink<>());
+
+ JobGraph jobGraph = env.getStreamGraph().getJobGraph();
+ jobGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(savepointPath));
+
+ MiniCluster miniCluster = cluster.getMiniCluster();
+ miniCluster.submitJob(jobGraph).get();
+
+ Map<String, Integer> restoredState =
+ NullUnsafeStatefulMapper.secondRunFuture.get(2, TimeUnit.MINUTES);
+
+ assertThat(restoredState.get("key")).isEqualTo(42);
+ assertThat(restoredState.get("null-key")).isNull();
+ assertThat(restoredState).containsKey("null-key");
+ }
+
+ /**
+ * A stateful mapper using IntSerializer (null-unsafe) for the map state value type. This
+ * exercises the code path where serializers that cannot handle null will fail during
+ * savepoint/checkpoint if the null-handling logic is incorrect.
+ */
+ private static class NullUnsafeStatefulMapper extends RichMapFunction<Long, Long> {
+
+ static CompletableFuture<Void> firstRunFuture;
+ static CompletableFuture<Map<String, Integer>> secondRunFuture;
+
+ private final boolean isFirstRun;
+ private boolean hasPopulated;
+ private transient MapState<String, Integer> mapState;
+
+ NullUnsafeStatefulMapper(boolean isFirstRun) {
+ this.isFirstRun = isFirstRun;
+ }
+
+ @Override
+ public void open(OpenContext context) {
+ MapStateDescriptor<String, Integer> mapStateDescriptor =
+ new MapStateDescriptor<>(
+ "map-state-int",
+ BasicTypeInfo.STRING_TYPE_INFO,
+ BasicTypeInfo.INT_TYPE_INFO);
+ mapState = getRuntimeContext().getMapState(mapStateDescriptor);
+ hasPopulated = false;
+ }
+
+ @Override
+ public Long map(Long value) throws Exception {
+ if (hasPopulated) {
+ return value;
+ }
+ if (isFirstRun) {
+ mapState.put("key", 42);
+ mapState.put("null-key", null);
+ firstRunFuture.complete(null);
+ } else {
+ Map<String, Integer> restoredState = new HashMap<>();
+ restoredState.put("key", mapState.get("key"));
+ restoredState.put("null-key", mapState.get("null-key"));
+ secondRunFuture.complete(restoredState);
+ }
+ hasPopulated = true;
+ return value;
+ }
+ }
+
private static class StatefulMapper extends RichMapFunction<Long, Long> {
static CompletableFuture<Void> firstRunFuture;