From 5c20e34c6f315e03775a6dd496dcf8e784bc37a2 Mon Sep 17 00:00:00 2001 From: wonyongChoi05 Date: Mon, 3 Aug 2026 20:45:28 +0900 Subject: [PATCH 1/3] [FLINK-40302][state] Fix materialization after aborted native savepoint --- .../changelog/ChangelogKeyedStateBackend.java | 11 +- ...gKeyedStateBackendMaterializationTest.java | 105 ++++++++++++++++++ 2 files changed, 113 insertions(+), 3 deletions(-) create mode 100644 flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendMaterializationTest.java diff --git a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackend.java b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackend.java index 4182023900b37a..3e4b5db5beb739 100644 --- a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackend.java +++ b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackend.java @@ -199,6 +199,9 @@ public class ChangelogKeyedStateBackend /** last failed or cancelled materialization. */ private long lastFailedMaterializationId = -1L; + /** ID of the last materialization actually triggered by {@link #initMaterialization()}. */ + private long lastTriggeredMaterializationId = -1L; + private final ChangelogTruncateHelper changelogTruncateHelper; /** @@ -820,6 +823,7 @@ private ChangelogSnapshotState completeRestore( } } this.lastConfirmedMaterializationId = materializationId; + this.lastTriggeredMaterializationId = materializationId; this.materializedId = materializationId + 1; if (!isRescaling @@ -851,15 +855,15 @@ private ChangelogSnapshotState completeRestore( */ @Override public Optional initMaterialization() throws Exception { - if (lastConfirmedMaterializationId < materializedId - 1 - && lastFailedMaterializationId < materializedId - 1) { + if (lastConfirmedMaterializationId < lastTriggeredMaterializationId + && lastFailedMaterializationId < lastTriggeredMaterializationId) { // SharedStateRegistry potentially requires that the checkpoint's dependency on the // shared file be continuous, it will be broken if we trigger a new materialization // before the previous one has either confirmed or failed. See discussion in // https://github.com/apache/flink/pull/22669#issuecomment-1593370772 . LOG.info( "materialization:{} not confirmed or failed or cancelled, skip trigger new one.", - materializedId - 1); + lastTriggeredMaterializationId); return Optional.empty(); } @@ -878,6 +882,7 @@ public Optional initMaterialization() throws Exception // streamFactory that is designed for state backend snapshot, which requires unique // checkpoint ID. A faked materialized Id is provided here. long materializationID = materializedId++; + lastTriggeredMaterializationId = materializationID; MaterializationRunnable materializationRunnable = new MaterializationRunnable( diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendMaterializationTest.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendMaterializationTest.java new file mode 100644 index 00000000000000..f9bbb9b889b394 --- /dev/null +++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendMaterializationTest.java @@ -0,0 +1,105 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.changelog; + +import org.apache.flink.api.common.ExecutionConfig; +import org.apache.flink.api.common.JobID; +import org.apache.flink.api.common.typeutils.base.IntSerializer; +import org.apache.flink.core.execution.SavepointFormatType; +import org.apache.flink.core.fs.CloseableRegistry; +import org.apache.flink.runtime.checkpoint.CheckpointOptions; +import org.apache.flink.runtime.checkpoint.SavepointType; +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups; +import org.apache.flink.runtime.query.KvStateRegistry; +import org.apache.flink.runtime.state.CheckpointStorageLocationReference; +import org.apache.flink.runtime.state.KeyGroupRange; +import org.apache.flink.runtime.state.UncompressedStreamCompressionDecorator; +import org.apache.flink.runtime.state.changelog.inmemory.InMemoryStateChangelogStorage; +import org.apache.flink.runtime.state.memory.MemCheckpointStreamFactory; +import org.apache.flink.runtime.state.metrics.LatencyTrackingStateConfig; +import org.apache.flink.runtime.state.metrics.SizeTrackingStateConfig; +import org.apache.flink.runtime.state.ttl.TtlTimeProvider; +import org.apache.flink.runtime.state.ttl.mock.MockKeyedStateBackend; +import org.apache.flink.runtime.state.ttl.mock.MockKeyedStateBackend.MockSnapshotSupplier; +import org.apache.flink.runtime.state.ttl.mock.MockKeyedStateBackendBuilder; +import org.apache.flink.state.changelog.ChangelogStateBackendTestUtils.DummyCheckpointingStorageAccess; + +import org.junit.jupiter.api.Test; + +import static java.util.Collections.emptyList; +import static org.assertj.core.api.Assertions.assertThat; + +class ChangelogKeyedStateBackendMaterializationTest { + + @Test + void testInitMaterializationAfterAbortedNativeSavepoint() throws Exception { + final ChangelogKeyedStateBackend backend = createChangelogBackend(); + + try { + final long savepointId = 1L; + backend.snapshot( + savepointId, + 0L, + new MemCheckpointStreamFactory(1000), + new CheckpointOptions( + SavepointType.savepoint(SavepointFormatType.NATIVE), + CheckpointStorageLocationReference.getDefault())); + backend.notifyCheckpointAborted(savepointId); + + backend.getChangelogWriter().append(0, new byte[] {'s'}); + + assertThat(backend.initMaterialization()).isPresent(); + } finally { + backend.close(); + backend.dispose(); + } + } + + private static ChangelogKeyedStateBackend createChangelogBackend() { + final MockKeyedStateBackend delegatedBackend = + new MockKeyedStateBackendBuilder<>( + new KvStateRegistry() + .createTaskRegistry(new JobID(), new JobVertexID()), + IntSerializer.INSTANCE, + ChangelogKeyedStateBackendMaterializationTest.class.getClassLoader(), + 1, + KeyGroupRange.of(0, 0), + new ExecutionConfig(), + TtlTimeProvider.DEFAULT, + LatencyTrackingStateConfig.disabled(), + SizeTrackingStateConfig.disabled(), + emptyList(), + UncompressedStreamCompressionDecorator.INSTANCE, + new CloseableRegistry(), + MockSnapshotSupplier.EMPTY) + .build(); + + return new ChangelogKeyedStateBackend<>( + delegatedBackend, + "test", + new ExecutionConfig(), + TtlTimeProvider.DEFAULT, + UnregisteredMetricGroups.createUnregisteredOperatorMetricGroup(), + new InMemoryStateChangelogStorage() + .createWriter("test", KeyGroupRange.EMPTY_KEY_GROUP_RANGE, null), + emptyList(), + new DummyCheckpointingStorageAccess()); + } +} From e2d99c506156818f9f9fea87da3d2c9458c0c9ca Mon Sep 17 00:00:00 2001 From: wonyongChoi05 Date: Mon, 3 Aug 2026 21:02:37 +0900 Subject: [PATCH 2/3] [FLINK-40302][state] Fix materialization after aborted native savepoint --- ...gKeyedStateBackendMaterializationTest.java | 105 ------------------ .../ChangelogKeyedStateBackendTest.java | 27 +++++ 2 files changed, 27 insertions(+), 105 deletions(-) delete mode 100644 flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendMaterializationTest.java diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendMaterializationTest.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendMaterializationTest.java deleted file mode 100644 index f9bbb9b889b394..00000000000000 --- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendMaterializationTest.java +++ /dev/null @@ -1,105 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.flink.state.changelog; - -import org.apache.flink.api.common.ExecutionConfig; -import org.apache.flink.api.common.JobID; -import org.apache.flink.api.common.typeutils.base.IntSerializer; -import org.apache.flink.core.execution.SavepointFormatType; -import org.apache.flink.core.fs.CloseableRegistry; -import org.apache.flink.runtime.checkpoint.CheckpointOptions; -import org.apache.flink.runtime.checkpoint.SavepointType; -import org.apache.flink.runtime.jobgraph.JobVertexID; -import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups; -import org.apache.flink.runtime.query.KvStateRegistry; -import org.apache.flink.runtime.state.CheckpointStorageLocationReference; -import org.apache.flink.runtime.state.KeyGroupRange; -import org.apache.flink.runtime.state.UncompressedStreamCompressionDecorator; -import org.apache.flink.runtime.state.changelog.inmemory.InMemoryStateChangelogStorage; -import org.apache.flink.runtime.state.memory.MemCheckpointStreamFactory; -import org.apache.flink.runtime.state.metrics.LatencyTrackingStateConfig; -import org.apache.flink.runtime.state.metrics.SizeTrackingStateConfig; -import org.apache.flink.runtime.state.ttl.TtlTimeProvider; -import org.apache.flink.runtime.state.ttl.mock.MockKeyedStateBackend; -import org.apache.flink.runtime.state.ttl.mock.MockKeyedStateBackend.MockSnapshotSupplier; -import org.apache.flink.runtime.state.ttl.mock.MockKeyedStateBackendBuilder; -import org.apache.flink.state.changelog.ChangelogStateBackendTestUtils.DummyCheckpointingStorageAccess; - -import org.junit.jupiter.api.Test; - -import static java.util.Collections.emptyList; -import static org.assertj.core.api.Assertions.assertThat; - -class ChangelogKeyedStateBackendMaterializationTest { - - @Test - void testInitMaterializationAfterAbortedNativeSavepoint() throws Exception { - final ChangelogKeyedStateBackend backend = createChangelogBackend(); - - try { - final long savepointId = 1L; - backend.snapshot( - savepointId, - 0L, - new MemCheckpointStreamFactory(1000), - new CheckpointOptions( - SavepointType.savepoint(SavepointFormatType.NATIVE), - CheckpointStorageLocationReference.getDefault())); - backend.notifyCheckpointAborted(savepointId); - - backend.getChangelogWriter().append(0, new byte[] {'s'}); - - assertThat(backend.initMaterialization()).isPresent(); - } finally { - backend.close(); - backend.dispose(); - } - } - - private static ChangelogKeyedStateBackend createChangelogBackend() { - final MockKeyedStateBackend delegatedBackend = - new MockKeyedStateBackendBuilder<>( - new KvStateRegistry() - .createTaskRegistry(new JobID(), new JobVertexID()), - IntSerializer.INSTANCE, - ChangelogKeyedStateBackendMaterializationTest.class.getClassLoader(), - 1, - KeyGroupRange.of(0, 0), - new ExecutionConfig(), - TtlTimeProvider.DEFAULT, - LatencyTrackingStateConfig.disabled(), - SizeTrackingStateConfig.disabled(), - emptyList(), - UncompressedStreamCompressionDecorator.INSTANCE, - new CloseableRegistry(), - MockSnapshotSupplier.EMPTY) - .build(); - - return new ChangelogKeyedStateBackend<>( - delegatedBackend, - "test", - new ExecutionConfig(), - TtlTimeProvider.DEFAULT, - UnregisteredMetricGroups.createUnregisteredOperatorMetricGroup(), - new InMemoryStateChangelogStorage() - .createWriter("test", KeyGroupRange.EMPTY_KEY_GROUP_RANGE, null), - emptyList(), - new DummyCheckpointingStorageAccess()); - } -} diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendTest.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendTest.java index 10a13a9b3d4762..83297da8e04d3a 100644 --- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendTest.java +++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendTest.java @@ -20,11 +20,14 @@ import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.JobID; import org.apache.flink.api.common.typeutils.base.IntSerializer; +import org.apache.flink.core.execution.SavepointFormatType; import org.apache.flink.core.fs.CloseableRegistry; import org.apache.flink.runtime.checkpoint.CheckpointOptions; +import org.apache.flink.runtime.checkpoint.SavepointType; import org.apache.flink.runtime.jobgraph.JobVertexID; import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups; import org.apache.flink.runtime.query.KvStateRegistry; +import org.apache.flink.runtime.state.CheckpointStorageLocationReference; import org.apache.flink.runtime.state.KeyGroupRange; import org.apache.flink.runtime.state.KeyedStateHandle; import org.apache.flink.runtime.state.SnapshotResult; @@ -182,4 +185,28 @@ private RunnableFuture> checkpoint( new MemCheckpointStreamFactory(1000), CheckpointOptions.forCheckpointWithDefaultLocation()); } + + @Test + public void testInitMaterializationAfterAbortedNativeSavepoint() throws Exception { + ChangelogKeyedStateBackend backend = createChangelog(createMock()); + try { + long savepointId = 1L; + backend.snapshot( + savepointId, + 0L, + new MemCheckpointStreamFactory(1000), + new CheckpointOptions( + SavepointType.savepoint(SavepointFormatType.NATIVE), + CheckpointStorageLocationReference.getDefault() + ) + ); + backend.notifyCheckpointAborted(savepointId); + appendMockStateChange(backend); + + assertTrue("materialization should not be blocked by an unconfirmed native savepoint", backend.initMaterialization().isPresent()); + } finally { + backend.close(); + backend.dispose(); + } + } } From 79ae8d063fc6a2e17b2db3c1093e48f4f007cd0c Mon Sep 17 00:00:00 2001 From: wonyongChoi05 Date: Mon, 3 Aug 2026 21:28:15 +0900 Subject: [PATCH 3/3] [FLINK-40302][state] Fix materialization after aborted native savepoint --- .../state/changelog/ChangelogKeyedStateBackendTest.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendTest.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendTest.java index 83297da8e04d3a..c0f01b5d0ea86f 100644 --- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendTest.java +++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendTest.java @@ -197,13 +197,13 @@ public void testInitMaterializationAfterAbortedNativeSavepoint() throws Exceptio new MemCheckpointStreamFactory(1000), new CheckpointOptions( SavepointType.savepoint(SavepointFormatType.NATIVE), - CheckpointStorageLocationReference.getDefault() - ) - ); + CheckpointStorageLocationReference.getDefault())); backend.notifyCheckpointAborted(savepointId); appendMockStateChange(backend); - assertTrue("materialization should not be blocked by an unconfirmed native savepoint", backend.initMaterialization().isPresent()); + assertTrue( + "materialization should not be blocked by an unconfirmed native savepoint", + backend.initMaterialization().isPresent()); } finally { backend.close(); backend.dispose();