You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ta...@apache.org on 2022/04/25 06:29:07 UTC
[flink] branch master updated (cb68ccf1b2c -> ece745c39ea)
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from cb68ccf1b2c [FLINK-27187][state/changelog] Add changelog storage metric totalAttemptsPerUpload
new 7b48eae6b44 [FLINK-23252][changelog] Support recovery from checkpoint after disabling changelog backend
new ece745c39ea [FLINK-23252][changelog] Introduce CheckpointBoundKeyedStateHandle#rebound
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../changelog/fs/FsStateChangelogStorage.java | 9 +-
.../fs/FsStateChangelogStorageFactory.java | 6 +
.../fs/FsStateChangelogStorageForRecovery.java | 40 +++
.../flink/changelog/fs/FsStateChangelogWriter.java | 7 +-
.../metadata/MetadataV2V3SerializerBase.java | 34 ++-
.../runtime/state/AbstractKeyedStateBackend.java | 65 ++++-
.../state/CheckpointBoundKeyedStateHandle.java | 10 +-
.../state/IncrementalLocalKeyedStateHandle.java | 11 +
.../state/IncrementalRemoteKeyedStateHandle.java | 13 +
.../flink/runtime/state/StateBackendLoader.java | 51 +++-
.../changelog/ChangelogStateBackendHandle.java | 44 ++-
.../state/changelog/ChangelogStateHandle.java | 6 +-
.../changelog/ChangelogStateHandleStreamImpl.java | 24 +-
.../state/changelog/StateChangelogStorage.java | 8 +-
.../changelog/StateChangelogStorageFactory.java | 3 +
.../changelog/StateChangelogStorageLoader.java | 22 ++
...Factory.java => StateChangelogStorageView.java} | 20 +-
.../inmemory/InMemoryChangelogStateHandle.java | 5 +
.../InMemoryStateChangelogStorageFactory.java | 10 +-
.../flink/runtime/state/ChangelogTestUtils.java | 7 +
.../runtime/state/SharedStateRegistryTest.java | 3 +
...kExecutorStateChangelogStoragesManagerTest.java | 6 +
.../changelog/ChangelogStateBackendHandleTest.java | 18 +-
.../inmemory/StateChangelogStorageLoaderTest.java | 8 +-
...end.java => AbstractChangelogStateBackend.java} | 99 +------
.../changelog/ChangelogKeyedStateBackend.java | 172 ++++++------
.../state/changelog/ChangelogStateBackend.java | 222 ++-------------
.../state/changelog/ChangelogStateFactory.java | 158 +++++++++++
.../DeactivatedChangelogStateBackend.java | 84 ++++++
.../flink/state/changelog/KvStateChangeLogger.java | 5 +-
.../flink/state/changelog/StateChangeLogger.java | 4 +-
.../restore/ChangelogBackendLogApplier.java | 35 +--
.../restore/ChangelogBackendRestoreOperation.java | 30 ++-
.../restore/ChangelogMigrationRestoreTarget.java | 298 +++++++++++++++++++++
.../changelog/restore/ChangelogRestoreTarget.java | 72 +++++
.../ChangelogStateBackendLoadingTest.java | 14 +
.../operators/StreamTaskStateInitializerImpl.java | 33 ++-
.../ChangelogPeriodicMaterializationITCase.java | 29 +-
...ngelogPeriodicMaterializationRescaleITCase.java | 97 +------
...ogPeriodicMaterializationSwitchEnvTestBase.java | 133 +++++++++
...dicMaterializationSwitchStateBackendITCase.java | 66 +++++
.../ChangelogPeriodicMaterializationTestBase.java | 105 ++++++--
.../test/state/ChangelogCompatibilityITCase.java | 7 +-
43 files changed, 1487 insertions(+), 606 deletions(-)
create mode 100644 flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogStorageForRecovery.java
copy flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/{StateChangelogStorageFactory.java => StateChangelogStorageView.java} (58%)
copy flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/{ChangelogStateBackend.java => AbstractChangelogStateBackend.java} (66%)
create mode 100644 flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateFactory.java
create mode 100644 flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/DeactivatedChangelogStateBackend.java
create mode 100644 flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogMigrationRestoreTarget.java
create mode 100644 flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogRestoreTarget.java
create mode 100644 flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationSwitchEnvTestBase.java
create mode 100644 flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationSwitchStateBackendITCase.java
[flink] 01/02: [FLINK-23252][changelog] Support recovery from checkpoint after disabling changelog backend
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7b48eae6b448963e0aed2b33d06e105311a48e05
Author: Hangxiang Yu <ma...@gmail.com>
AuthorDate: Wed Apr 20 14:24:02 2022 +0800
[FLINK-23252][changelog] Support recovery from checkpoint after disabling changelog backend
---
.../changelog/fs/FsStateChangelogStorage.java | 9 +-
.../fs/FsStateChangelogStorageFactory.java | 6 +
.../fs/FsStateChangelogStorageForRecovery.java | 40 +++
.../flink/changelog/fs/FsStateChangelogWriter.java | 7 +-
.../metadata/MetadataV2V3SerializerBase.java | 18 +-
.../runtime/state/AbstractKeyedStateBackend.java | 65 ++++-
.../flink/runtime/state/StateBackendLoader.java | 51 +++-
.../state/changelog/ChangelogStateHandle.java | 6 +-
.../changelog/ChangelogStateHandleStreamImpl.java | 24 +-
.../state/changelog/StateChangelogStorage.java | 8 +-
.../changelog/StateChangelogStorageFactory.java | 3 +
.../changelog/StateChangelogStorageLoader.java | 22 ++
...Factory.java => StateChangelogStorageView.java} | 20 +-
.../inmemory/InMemoryChangelogStateHandle.java | 5 +
.../InMemoryStateChangelogStorageFactory.java | 10 +-
.../flink/runtime/state/ChangelogTestUtils.java | 6 +
...kExecutorStateChangelogStoragesManagerTest.java | 6 +
.../inmemory/StateChangelogStorageLoaderTest.java | 8 +-
...end.java => AbstractChangelogStateBackend.java} | 98 +------
.../changelog/ChangelogKeyedStateBackend.java | 163 +++++------
.../state/changelog/ChangelogStateBackend.java | 222 ++-------------
.../state/changelog/ChangelogStateFactory.java | 158 +++++++++++
.../DeactivatedChangelogStateBackend.java | 64 +++++
.../flink/state/changelog/KvStateChangeLogger.java | 5 +-
.../flink/state/changelog/StateChangeLogger.java | 4 +-
.../restore/ChangelogBackendLogApplier.java | 35 +--
.../restore/ChangelogBackendRestoreOperation.java | 30 ++-
.../restore/ChangelogMigrationRestoreTarget.java | 298 +++++++++++++++++++++
.../changelog/restore/ChangelogRestoreTarget.java | 72 +++++
.../ChangelogStateBackendLoadingTest.java | 14 +
.../operators/StreamTaskStateInitializerImpl.java | 33 ++-
.../ChangelogPeriodicMaterializationITCase.java | 29 +-
...ngelogPeriodicMaterializationRescaleITCase.java | 97 +------
...ogPeriodicMaterializationSwitchEnvTestBase.java | 133 +++++++++
...dicMaterializationSwitchStateBackendITCase.java | 66 +++++
.../ChangelogPeriodicMaterializationTestBase.java | 105 ++++++--
.../test/state/ChangelogCompatibilityITCase.java | 7 +-
37 files changed, 1350 insertions(+), 597 deletions(-)
diff --git a/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogStorage.java b/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogStorage.java
index adeabed918a..bfe9abc8f33 100644
--- a/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogStorage.java
+++ b/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogStorage.java
@@ -26,8 +26,6 @@ import org.apache.flink.runtime.io.AvailabilityProvider;
import org.apache.flink.runtime.metrics.groups.TaskManagerJobMetricGroup;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.changelog.ChangelogStateHandleStreamImpl;
-import org.apache.flink.runtime.state.changelog.StateChangelogHandleReader;
-import org.apache.flink.runtime.state.changelog.StateChangelogHandleStreamHandleReader;
import org.apache.flink.runtime.state.changelog.StateChangelogStorage;
import org.slf4j.Logger;
@@ -46,7 +44,7 @@ import static org.apache.flink.changelog.fs.StateChangeUploadScheduler.fromConfi
/** Filesystem-based implementation of {@link StateChangelogStorage}. */
@Experimental
@ThreadSafe
-public class FsStateChangelogStorage
+public class FsStateChangelogStorage extends FsStateChangelogStorageForRecovery
implements StateChangelogStorage<ChangelogStateHandleStreamImpl> {
private static final Logger LOG = LoggerFactory.getLogger(FsStateChangelogStorage.class);
@@ -100,11 +98,6 @@ public class FsStateChangelogStorage
logId, keyGroupRange, uploader, preEmptivePersistThresholdInBytes, mailboxExecutor);
}
- @Override
- public StateChangelogHandleReader<ChangelogStateHandleStreamImpl> createReader() {
- return new StateChangelogHandleStreamHandleReader(new StateChangeFormat());
- }
-
@Override
public void close() throws Exception {
uploader.close();
diff --git a/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogStorageFactory.java b/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogStorageFactory.java
index 8927fc8023d..c5489ea86b2 100644
--- a/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogStorageFactory.java
+++ b/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogStorageFactory.java
@@ -22,6 +22,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.metrics.groups.TaskManagerJobMetricGroup;
import org.apache.flink.runtime.state.changelog.StateChangelogStorage;
import org.apache.flink.runtime.state.changelog.StateChangelogStorageFactory;
+import org.apache.flink.runtime.state.changelog.StateChangelogStorageView;
import java.io.File;
import java.io.IOException;
@@ -46,6 +47,11 @@ public class FsStateChangelogStorageFactory implements StateChangelogStorageFact
return new FsStateChangelogStorage(configuration, metricGroup);
}
+ @Override
+ public StateChangelogStorageView<?> createStorageView() {
+ return new FsStateChangelogStorageForRecovery();
+ }
+
public static void configure(Configuration configuration, File newFolder) {
configuration.setString(STATE_CHANGE_LOG_STORAGE, IDENTIFIER);
configuration.setString(BASE_PATH, newFolder.getAbsolutePath());
diff --git a/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogStorageForRecovery.java b/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogStorageForRecovery.java
new file mode 100644
index 00000000000..377ac81e053
--- /dev/null
+++ b/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogStorageForRecovery.java
@@ -0,0 +1,40 @@
+/*
+ * 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.changelog.fs;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.runtime.state.changelog.ChangelogStateHandleStreamImpl;
+import org.apache.flink.runtime.state.changelog.StateChangelogHandleReader;
+import org.apache.flink.runtime.state.changelog.StateChangelogHandleStreamHandleReader;
+import org.apache.flink.runtime.state.changelog.StateChangelogStorage;
+import org.apache.flink.runtime.state.changelog.StateChangelogStorageView;
+
+import javax.annotation.concurrent.ThreadSafe;
+
+/** Filesystem-based implementation of {@link StateChangelogStorage} just for recovery. */
+@Experimental
+@ThreadSafe
+public class FsStateChangelogStorageForRecovery
+ implements StateChangelogStorageView<ChangelogStateHandleStreamImpl> {
+
+ @Override
+ public StateChangelogHandleReader<ChangelogStateHandleStreamImpl> createReader() {
+ return new StateChangelogHandleStreamHandleReader(new StateChangeFormat());
+ }
+}
diff --git a/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogWriter.java b/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogWriter.java
index 87e6240fd23..2c255f2ab6c 100644
--- a/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogWriter.java
+++ b/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogWriter.java
@@ -306,7 +306,12 @@ class FsStateChangelogWriter implements StateChangelogWriter<ChangelogStateHandl
tuples.add(Tuple2.of(uploadResult.getStreamStateHandle(), uploadResult.getOffset()));
size += uploadResult.getSize();
}
- return new ChangelogStateHandleStreamImpl(tuples, keyGroupRange, size, incrementalSize);
+ return new ChangelogStateHandleStreamImpl(
+ tuples,
+ keyGroupRange,
+ size,
+ incrementalSize,
+ FsStateChangelogStorageFactory.IDENTIFIER);
}
@VisibleForTesting
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java
index d9d10c08af3..8bb9db305d1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java
@@ -117,6 +117,8 @@ public abstract class MetadataV2V3SerializerBase {
private static final byte INCREMENTAL_KEY_GROUPS_HANDLE_V2 = 11;
// KEY_GROUPS_HANDLE_V2 is introduced to add new field of stateHandleId.
private static final byte KEY_GROUPS_HANDLE_V2 = 12;
+ // CHANGELOG_FILE_INCREMENT_HANDLE_V2 is introduced to add new field of storageIdentifier.
+ private static final byte CHANGELOG_FILE_INCREMENT_HANDLE_V2 = 13;
// ------------------------------------------------------------------------
// (De)serialization entry points
@@ -379,7 +381,7 @@ public abstract class MetadataV2V3SerializerBase {
writeStateHandleId(handle, dos);
} else if (stateHandle instanceof ChangelogStateHandleStreamImpl) {
ChangelogStateHandleStreamImpl handle = (ChangelogStateHandleStreamImpl) stateHandle;
- dos.writeByte(CHANGELOG_FILE_INCREMENT_HANDLE);
+ dos.writeByte(CHANGELOG_FILE_INCREMENT_HANDLE_V2);
dos.writeInt(handle.getKeyGroupRange().getStartKeyGroup());
dos.writeInt(handle.getKeyGroupRange().getNumberOfKeyGroups());
dos.writeInt(handle.getHandlesAndOffsets().size());
@@ -391,7 +393,7 @@ public abstract class MetadataV2V3SerializerBase {
dos.writeLong(handle.getStateSize());
dos.writeLong(handle.getCheckpointedSize());
writeStateHandleId(handle, dos);
-
+ dos.writeUTF(handle.getStorageIdentifier());
} else {
throw new IllegalStateException(
"Unknown KeyedStateHandle type: " + stateHandle.getClass());
@@ -492,7 +494,8 @@ public abstract class MetadataV2V3SerializerBase {
keyGroupRange,
stateHandleId);
- } else if (CHANGELOG_FILE_INCREMENT_HANDLE == type) {
+ } else if (CHANGELOG_FILE_INCREMENT_HANDLE == type
+ || CHANGELOG_FILE_INCREMENT_HANDLE_V2 == type) {
int start = dis.readInt();
int numKeyGroups = dis.readInt();
KeyGroupRange keyGroupRange = KeyGroupRange.of(start, start + numKeyGroups - 1);
@@ -507,8 +510,15 @@ public abstract class MetadataV2V3SerializerBase {
long size = dis.readLong();
long checkpointedSize = dis.readLong();
StateHandleID stateHandleId = new StateHandleID(dis.readUTF());
+ String storageIdentifier =
+ CHANGELOG_FILE_INCREMENT_HANDLE_V2 == type ? dis.readUTF() : "filesystem";
return ChangelogStateHandleStreamImpl.restore(
- streamHandleAndOffset, keyGroupRange, size, checkpointedSize, stateHandleId);
+ streamHandleAndOffset,
+ keyGroupRange,
+ size,
+ checkpointedSize,
+ storageIdentifier,
+ stateHandleId);
} else {
throw new IllegalStateException("Reading invalid KeyedStateHandle, type: " + type);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java
index 29769824eb9..f9c26725093 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java
@@ -131,9 +131,64 @@ public abstract class AbstractKeyedStateBackend<K>
CloseableRegistry cancelStreamRegistry,
StreamCompressionDecorator keyGroupCompressionDecorator,
InternalKeyContext<K> keyContext) {
+ this(
+ kvStateRegistry,
+ keySerializer,
+ userCodeClassLoader,
+ executionConfig,
+ ttlTimeProvider,
+ latencyTrackingStateConfig,
+ cancelStreamRegistry,
+ keyGroupCompressionDecorator,
+ Preconditions.checkNotNull(keyContext),
+ keyContext.getNumberOfKeyGroups(),
+ keyContext.getKeyGroupRange(),
+ new HashMap<>(),
+ new ArrayList<>(1),
+ null,
+ null);
+ }
+
+ // Copy constructor
+ protected AbstractKeyedStateBackend(AbstractKeyedStateBackend<K> abstractKeyedStateBackend) {
+ this(
+ abstractKeyedStateBackend.kvStateRegistry,
+ abstractKeyedStateBackend.keySerializer,
+ abstractKeyedStateBackend.userCodeClassLoader,
+ abstractKeyedStateBackend.executionConfig,
+ abstractKeyedStateBackend.ttlTimeProvider,
+ abstractKeyedStateBackend.latencyTrackingStateConfig,
+ abstractKeyedStateBackend.cancelStreamRegistry,
+ abstractKeyedStateBackend.keyGroupCompressionDecorator,
+ abstractKeyedStateBackend.keyContext,
+ abstractKeyedStateBackend.numberOfKeyGroups,
+ abstractKeyedStateBackend.keyGroupRange,
+ abstractKeyedStateBackend.keyValueStatesByName,
+ abstractKeyedStateBackend.keySelectionListeners,
+ abstractKeyedStateBackend.lastState,
+ abstractKeyedStateBackend.lastName);
+ }
+
+ @SuppressWarnings("rawtypes")
+ private AbstractKeyedStateBackend(
+ TaskKvStateRegistry kvStateRegistry,
+ TypeSerializer<K> keySerializer,
+ ClassLoader userCodeClassLoader,
+ ExecutionConfig executionConfig,
+ TtlTimeProvider ttlTimeProvider,
+ LatencyTrackingStateConfig latencyTrackingStateConfig,
+ CloseableRegistry cancelStreamRegistry,
+ StreamCompressionDecorator keyGroupCompressionDecorator,
+ InternalKeyContext<K> keyContext,
+ int numberOfKeyGroups,
+ KeyGroupRange keyGroupRange,
+ HashMap<String, InternalKvState<K, ?, ?>> keyValueStatesByName,
+ ArrayList<KeySelectionListener<K>> keySelectionListeners,
+ InternalKvState lastState,
+ String lastName) {
this.keyContext = Preconditions.checkNotNull(keyContext);
- this.numberOfKeyGroups = keyContext.getNumberOfKeyGroups();
- this.keyGroupRange = Preconditions.checkNotNull(keyContext.getKeyGroupRange());
+ this.numberOfKeyGroups = numberOfKeyGroups;
+ this.keyGroupRange = Preconditions.checkNotNull(keyGroupRange);
Preconditions.checkArgument(
numberOfKeyGroups >= 1, "NumberOfKeyGroups must be a positive number");
Preconditions.checkArgument(
@@ -147,12 +202,14 @@ public abstract class AbstractKeyedStateBackend<K>
this.keySerializer = keySerializer;
this.userCodeClassLoader = Preconditions.checkNotNull(userCodeClassLoader);
this.cancelStreamRegistry = cancelStreamRegistry;
- this.keyValueStatesByName = new HashMap<>();
+ this.keyValueStatesByName = keyValueStatesByName;
this.executionConfig = executionConfig;
this.keyGroupCompressionDecorator = keyGroupCompressionDecorator;
this.ttlTimeProvider = Preconditions.checkNotNull(ttlTimeProvider);
this.latencyTrackingStateConfig = Preconditions.checkNotNull(latencyTrackingStateConfig);
- this.keySelectionListeners = new ArrayList<>(1);
+ this.keySelectionListeners = keySelectionListeners;
+ this.lastState = lastState;
+ this.lastName = lastName;
}
private static StreamCompressionDecorator determineStreamCompression(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java
index bb60d0b4520..929728bdd04 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java
@@ -24,6 +24,7 @@ import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.configuration.StateBackendOptions;
import org.apache.flink.configuration.StateChangelogOptions;
+import org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle;
import org.apache.flink.runtime.state.delegate.DelegatingStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackendFactory;
@@ -40,6 +41,7 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
+import java.util.Collection;
import java.util.Optional;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -53,6 +55,10 @@ public class StateBackendLoader {
private static final String CHANGELOG_STATE_BACKEND =
"org.apache.flink.state.changelog.ChangelogStateBackend";
+ /** Used for Loading TempChangelogStateBackend. */
+ private static final String DEACTIVATED_CHANGELOG_STATE_BACKEND =
+ "org.apache.flink.state.changelog.DeactivatedChangelogStateBackend";
+
/** Used for loading RocksDBStateBackend. */
private static final String ROCKSDB_STATE_BACKEND_FACTORY =
"org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackendFactory";
@@ -305,7 +311,7 @@ public class StateBackendLoader {
StateBackend backend;
if (enableChangeLog) {
- backend = loadChangelogStateBackend(rootBackend, classLoader);
+ backend = wrapStateBackend(rootBackend, classLoader, CHANGELOG_STATE_BACKEND);
LOG.info(
"State backend loader loads {} to delegate {}",
backend.getClass().getSimpleName(),
@@ -358,26 +364,57 @@ public class StateBackendLoader {
return false;
}
- private static StateBackend loadChangelogStateBackend(
- StateBackend backend, ClassLoader classLoader) throws DynamicCodeLoadingException {
+ /**
+ * Load state backend which may wrap the original state backend for recovery.
+ *
+ * @param originalStateBackend StateBackend loaded from application or config.
+ * @param classLoader User code classloader.
+ * @param keyedStateHandles The state handles for restore.
+ * @return Wrapped state backend for recovery.
+ * @throws DynamicCodeLoadingException Thrown if keyed state handles of wrapped state backend
+ * are found and the class was not found or could not be instantiated.
+ */
+ public static StateBackend loadStateBackendFromKeyedStateHandles(
+ StateBackend originalStateBackend,
+ ClassLoader classLoader,
+ Collection<KeyedStateHandle> keyedStateHandles)
+ throws DynamicCodeLoadingException {
+ // Wrapping ChangelogStateBackend or ChangelogStateBackendHandle is not supported currently.
+ if (!isChangelogStateBackend(originalStateBackend)
+ && keyedStateHandles.stream()
+ .anyMatch(
+ stateHandle ->
+ stateHandle instanceof ChangelogStateBackendHandle)) {
+ return wrapStateBackend(
+ originalStateBackend, classLoader, DEACTIVATED_CHANGELOG_STATE_BACKEND);
+ }
+ return originalStateBackend;
+ }
+
+ private static boolean isChangelogStateBackend(StateBackend backend) {
+ return CHANGELOG_STATE_BACKEND.equals(backend.getClass().getName());
+ }
+
+ private static StateBackend wrapStateBackend(
+ StateBackend backend, ClassLoader classLoader, String className)
+ throws DynamicCodeLoadingException {
// ChangelogStateBackend resides in a separate module, load it using reflection
try {
Constructor<? extends DelegatingStateBackend> constructor =
- Class.forName(CHANGELOG_STATE_BACKEND, false, classLoader)
+ Class.forName(className, false, classLoader)
.asSubclass(DelegatingStateBackend.class)
.getDeclaredConstructor(StateBackend.class);
constructor.setAccessible(true);
return constructor.newInstance(backend);
} catch (ClassNotFoundException e) {
throw new DynamicCodeLoadingException(
- "Cannot find DelegateStateBackend class: " + CHANGELOG_STATE_BACKEND, e);
+ "Cannot find DelegateStateBackend class: " + className, e);
} catch (InstantiationException
| IllegalAccessException
| NoSuchMethodException
| InvocationTargetException e) {
- throw new DynamicCodeLoadingException(
- "Fail to initialize: " + CHANGELOG_STATE_BACKEND, e);
+ throw new DynamicCodeLoadingException("Fail to initialize: " + className, e);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/ChangelogStateHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/ChangelogStateHandle.java
index 070afeb37da..ddb18687d1c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/ChangelogStateHandle.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/ChangelogStateHandle.java
@@ -22,4 +22,8 @@ import org.apache.flink.runtime.state.KeyedStateHandle;
/** A handle to saved {@link StateChange state changes}. */
@Internal
-public interface ChangelogStateHandle extends KeyedStateHandle {}
+public interface ChangelogStateHandle extends KeyedStateHandle {
+
+ /** Get the storage identifier related to a specific {@link StateChangelogStorage}. */
+ String getStorageIdentifier();
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/ChangelogStateHandleStreamImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/ChangelogStateHandleStreamImpl.java
index 5a996b239bf..24f9df9705d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/ChangelogStateHandleStreamImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/ChangelogStateHandleStreamImpl.java
@@ -48,16 +48,20 @@ public final class ChangelogStateHandleStreamImpl implements ChangelogStateHandl
private final long incrementalSize;
private final StateHandleID stateHandleID;
+ private final String storageIdentifier;
+
public ChangelogStateHandleStreamImpl(
List<Tuple2<StreamStateHandle, Long>> handlesAndOffsets,
KeyGroupRange keyGroupRange,
long size,
- long incrementalSize) {
+ long incrementalSize,
+ String storageIdentifier) {
this(
handlesAndOffsets,
keyGroupRange,
size,
incrementalSize,
+ storageIdentifier,
new StateHandleID(UUID.randomUUID().toString()));
}
@@ -66,11 +70,13 @@ public final class ChangelogStateHandleStreamImpl implements ChangelogStateHandl
KeyGroupRange keyGroupRange,
long size,
long incrementalSize,
+ String storageIdentifier,
StateHandleID stateHandleId) {
this.handlesAndOffsets = handlesAndOffsets;
this.keyGroupRange = keyGroupRange;
this.size = size;
this.incrementalSize = incrementalSize;
+ this.storageIdentifier = storageIdentifier;
this.stateHandleID = stateHandleId;
}
@@ -79,9 +85,15 @@ public final class ChangelogStateHandleStreamImpl implements ChangelogStateHandl
KeyGroupRange keyGroupRange,
long size,
long incrementalSize,
+ String storageIdentifier,
StateHandleID stateHandleID) {
return new ChangelogStateHandleStreamImpl(
- handlesAndOffsets, keyGroupRange, size, incrementalSize, stateHandleID);
+ handlesAndOffsets,
+ keyGroupRange,
+ size,
+ incrementalSize,
+ storageIdentifier,
+ stateHandleID);
}
@Override
@@ -104,7 +116,8 @@ public final class ChangelogStateHandleStreamImpl implements ChangelogStateHandl
if (offsets.getNumberOfKeyGroups() == 0) {
return null;
}
- return new ChangelogStateHandleStreamImpl(handlesAndOffsets, offsets, 0L, 0L /* unknown */);
+ return new ChangelogStateHandleStreamImpl(
+ handlesAndOffsets, offsets, 0L, 0L /* unknown */, storageIdentifier);
}
@Override
@@ -133,6 +146,11 @@ public final class ChangelogStateHandleStreamImpl implements ChangelogStateHandl
return incrementalSize;
}
+ @Override
+ public String getStorageIdentifier() {
+ return storageIdentifier;
+ }
+
private static SharedStateRegistryKey getKey(StreamStateHandle stateHandle) {
// StateHandle key used in SharedStateRegistry should only be based on the file name
// and not on backend UUID or keygroup (multiple handles can refer to the same file and
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorage.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorage.java
index 14794b40efa..4ee07c4cb65 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorage.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorage.java
@@ -29,16 +29,12 @@ import org.apache.flink.runtime.state.KeyGroupRange;
* obtain an instance.
*/
@Internal
-public interface StateChangelogStorage<Handle extends ChangelogStateHandle> extends AutoCloseable {
+public interface StateChangelogStorage<Handle extends ChangelogStateHandle>
+ extends StateChangelogStorageView<Handle> {
StateChangelogWriter<Handle> createWriter(
String operatorID, KeyGroupRange keyGroupRange, MailboxExecutor mailboxExecutor);
- StateChangelogHandleReader<Handle> createReader();
-
- @Override
- default void close() throws Exception {}
-
default AvailabilityProvider getAvailabilityProvider() {
return () -> AvailabilityProvider.AVAILABLE;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorageFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorageFactory.java
index 715bfc2ad13..2af108181ca 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorageFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorageFactory.java
@@ -36,4 +36,7 @@ public interface StateChangelogStorageFactory {
/** Create the storage based on a configuration. */
StateChangelogStorage<?> createStorage(
Configuration configuration, TaskManagerJobMetricGroup metricGroup) throws IOException;
+
+ /** Create the storage for recovery. */
+ StateChangelogStorageView<?> createStorageView() throws IOException;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorageLoader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorageLoader.java
index 9a59a3535b6..28834aef2d1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorageLoader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorageLoader.java
@@ -22,10 +22,12 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.StateChangelogOptions;
import org.apache.flink.core.plugin.PluginManager;
import org.apache.flink.runtime.metrics.groups.TaskManagerJobMetricGroup;
+import org.apache.flink.util.FlinkRuntimeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.io.IOException;
@@ -99,4 +101,24 @@ public class StateChangelogStorageLoader {
return factory.createStorage(configuration, metricGroup);
}
}
+
+ @Nonnull
+ public static StateChangelogStorageView<?> loadFromStateHandle(
+ ChangelogStateHandle changelogStateHandle) throws IOException {
+ StateChangelogStorageFactory factory =
+ STATE_CHANGELOG_STORAGE_FACTORIES.get(changelogStateHandle.getStorageIdentifier());
+ if (factory == null) {
+ throw new FlinkRuntimeException(
+ String.format(
+ "Cannot find a factory for changelog storage with name '%s' to restore from '%s'.",
+ changelogStateHandle.getStorageIdentifier(),
+ changelogStateHandle.getClass().getSimpleName()));
+ } else {
+ LOG.info(
+ "Creating a changelog storage with name '{}' to restore from '{}'.",
+ changelogStateHandle.getStorageIdentifier(),
+ changelogStateHandle.getClass().getSimpleName());
+ return factory.createStorageView();
+ }
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorageFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorageView.java
similarity index 58%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorageFactory.java
copy to flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorageView.java
index 715bfc2ad13..2a9050887a5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorageFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChangelogStorageView.java
@@ -19,21 +19,17 @@
package org.apache.flink.runtime.state.changelog;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.metrics.groups.TaskManagerJobMetricGroup;
-
-import java.io.IOException;
/**
- * A factory for {@link StateChangelogStorage}. Please use {@link StateChangelogStorageLoader} to
- * create {@link StateChangelogStorage}.
+ * A storage view for changelog. Could produce {@link StateChangelogHandleReader} for read. Please
+ * use {@link StateChangelogStorageLoader} to obtain an instance.
*/
@Internal
-public interface StateChangelogStorageFactory {
- /** Get the identifier for user to use this changelog storage factory. */
- String getIdentifier();
+public interface StateChangelogStorageView<Handle extends ChangelogStateHandle>
+ extends AutoCloseable {
+
+ StateChangelogHandleReader<Handle> createReader();
- /** Create the storage based on a configuration. */
- StateChangelogStorage<?> createStorage(
- Configuration configuration, TaskManagerJobMetricGroup metricGroup) throws IOException;
+ @Override
+ default void close() throws Exception {}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/inmemory/InMemoryChangelogStateHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/inmemory/InMemoryChangelogStateHandle.java
index 953b2b9930c..bf72b073654 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/inmemory/InMemoryChangelogStateHandle.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/inmemory/InMemoryChangelogStateHandle.java
@@ -127,4 +127,9 @@ public class InMemoryChangelogStateHandle implements ChangelogStateHandle {
public long getTo() {
return ((SequenceNumber.GenericSequenceNumber) to).number;
}
+
+ @Override
+ public String getStorageIdentifier() {
+ return InMemoryStateChangelogStorageFactory.IDENTIFIER;
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/inmemory/InMemoryStateChangelogStorageFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/inmemory/InMemoryStateChangelogStorageFactory.java
index 203388d5407..c55ec12cc07 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/inmemory/InMemoryStateChangelogStorageFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/inmemory/InMemoryStateChangelogStorageFactory.java
@@ -21,15 +21,16 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.metrics.groups.TaskManagerJobMetricGroup;
import org.apache.flink.runtime.state.changelog.StateChangelogStorage;
import org.apache.flink.runtime.state.changelog.StateChangelogStorageFactory;
+import org.apache.flink.runtime.state.changelog.StateChangelogStorageView;
/** An {@link StateChangelogStorageFactory} for creating {@link InMemoryStateChangelogStorage}. */
public class InMemoryStateChangelogStorageFactory implements StateChangelogStorageFactory {
- public static String identifier = "memory";
+ public static final String IDENTIFIER = "memory";
@Override
public String getIdentifier() {
- return identifier;
+ return IDENTIFIER;
}
@Override
@@ -37,4 +38,9 @@ public class InMemoryStateChangelogStorageFactory implements StateChangelogStora
Configuration configuration, TaskManagerJobMetricGroup metricGroup) {
return new InMemoryStateChangelogStorage();
}
+
+ @Override
+ public StateChangelogStorageView<?> createStorageView() {
+ return new InMemoryStateChangelogStorage();
+ }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChangelogTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChangelogTestUtils.java
index cf5a657d0df..1897cf92d18 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChangelogTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChangelogTestUtils.java
@@ -21,6 +21,7 @@
package org.apache.flink.runtime.state;
import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.runtime.checkpoint.StateHandleDummyUtil;
import org.apache.flink.runtime.checkpoint.metadata.CheckpointTestUtils;
import org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle;
import org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle.ChangelogStateBackendHandleImpl;
@@ -35,6 +36,11 @@ import java.util.concurrent.ThreadLocalRandom;
/** Test utils for changelog * */
public class ChangelogTestUtils {
+ public static ChangelogStateBackendHandle createChangelogStateBackendHandle() {
+ return createChangelogStateBackendHandle(
+ StateHandleDummyUtil.createNewKeyedStateHandle(new KeyGroupRange(0, 1)));
+ }
+
public static ChangelogStateBackendHandle createChangelogStateBackendHandle(
KeyedStateHandle keyedStateHandle) {
return new ChangelogStateBackendHandleImpl(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorStateChangelogStoragesManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorStateChangelogStoragesManagerTest.java
index db3502aed49..77e9a09ed4c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorStateChangelogStoragesManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorStateChangelogStoragesManagerTest.java
@@ -29,6 +29,7 @@ import org.apache.flink.runtime.state.changelog.StateChangelogHandleReader;
import org.apache.flink.runtime.state.changelog.StateChangelogStorage;
import org.apache.flink.runtime.state.changelog.StateChangelogStorageFactory;
import org.apache.flink.runtime.state.changelog.StateChangelogStorageLoader;
+import org.apache.flink.runtime.state.changelog.StateChangelogStorageView;
import org.apache.flink.runtime.state.changelog.StateChangelogWriter;
import org.junit.Assert;
@@ -209,5 +210,10 @@ public class TaskExecutorStateChangelogStoragesManagerTest {
Configuration configuration, TaskManagerJobMetricGroup metricGroup) {
return new TestStateChangelogStorage();
}
+
+ @Override
+ public StateChangelogStorageView<?> createStorageView() throws IOException {
+ return new TestStateChangelogStorage();
+ }
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/changelog/inmemory/StateChangelogStorageLoaderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/changelog/inmemory/StateChangelogStorageLoaderTest.java
index 8fdfc26493b..64d00c38d89 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/changelog/inmemory/StateChangelogStorageLoaderTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/changelog/inmemory/StateChangelogStorageLoaderTest.java
@@ -28,6 +28,7 @@ import org.apache.flink.runtime.state.changelog.StateChangelogHandleReader;
import org.apache.flink.runtime.state.changelog.StateChangelogStorage;
import org.apache.flink.runtime.state.changelog.StateChangelogStorageFactory;
import org.apache.flink.runtime.state.changelog.StateChangelogStorageLoader;
+import org.apache.flink.runtime.state.changelog.StateChangelogStorageView;
import org.apache.flink.runtime.state.changelog.StateChangelogWriter;
import org.junit.Test;
@@ -108,7 +109,7 @@ public class StateChangelogStorageLoaderTest {
@Override
public String getIdentifier() {
// same identifier for overlapping test.
- return InMemoryStateChangelogStorageFactory.identifier;
+ return InMemoryStateChangelogStorageFactory.IDENTIFIER;
}
@Override
@@ -116,5 +117,10 @@ public class StateChangelogStorageLoaderTest {
Configuration configuration, TaskManagerJobMetricGroup metricGroup) {
return new TestStateChangelogStorage();
}
+
+ @Override
+ public StateChangelogStorageView<?> createStorageView() throws IOException {
+ return new TestStateChangelogStorage();
+ }
}
}
diff --git a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateBackend.java b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/AbstractChangelogStateBackend.java
similarity index 66%
copy from flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateBackend.java
copy to flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/AbstractChangelogStateBackend.java
index afcd17aa46c..826ad4828a3 100644
--- a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateBackend.java
+++ b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/AbstractChangelogStateBackend.java
@@ -19,11 +19,8 @@
package org.apache.flink.state.changelog;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.configuration.IllegalConfigurationException;
-import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.metrics.MetricGroup;
@@ -32,7 +29,6 @@ import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.CheckpointBoundKeyedStateHandle;
import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
-import org.apache.flink.runtime.state.ConfigurableStateBackend;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateBackend;
@@ -41,11 +37,8 @@ import org.apache.flink.runtime.state.SavepointKeyedStateHandle;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle;
import org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle.ChangelogStateBackendHandleImpl;
-import org.apache.flink.runtime.state.changelog.StateChangelogStorage;
import org.apache.flink.runtime.state.delegate.DelegatingStateBackend;
import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
-import org.apache.flink.runtime.taskmanager.AsynchronousException;
-import org.apache.flink.state.changelog.restore.ChangelogBackendRestoreOperation;
import org.apache.flink.state.changelog.restore.ChangelogBackendRestoreOperation.BaseBackendBuilder;
import org.apache.flink.util.Preconditions;
@@ -60,20 +53,20 @@ import java.util.stream.Collectors;
import static java.util.Collections.emptyList;
import static java.util.Collections.singletonList;
-import static org.apache.flink.util.Preconditions.checkNotNull;
/**
- * This state backend holds the working state in the underlying delegatedStateBackend, and forwards
- * state changes to State Changelog.
+ * An abstract base implementation of the {@link StateBackend} interface whose subclasses use
+ * delegatedStateBackend and State changes to restore.
*/
@Internal
-public class ChangelogStateBackend implements DelegatingStateBackend, ConfigurableStateBackend {
+public abstract class AbstractChangelogStateBackend
+ implements DelegatingStateBackend, StateBackend {
private static final long serialVersionUID = 1000L;
- private static final Logger LOG = LoggerFactory.getLogger(ChangelogStateBackend.class);
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractChangelogStateBackend.class);
- private final StateBackend delegatedStateBackend;
+ protected final StateBackend delegatedStateBackend;
/**
* Delegate a state backend by a ChangelogStateBackend.
@@ -82,7 +75,7 @@ public class ChangelogStateBackend implements DelegatingStateBackend, Configurab
*
* @param stateBackend the delegated state backend.
*/
- ChangelogStateBackend(StateBackend stateBackend) {
+ AbstractChangelogStateBackend(StateBackend stateBackend) {
this.delegatedStateBackend = Preconditions.checkNotNull(stateBackend);
Preconditions.checkArgument(
@@ -95,7 +88,7 @@ public class ChangelogStateBackend implements DelegatingStateBackend, Configurab
}
@Override
- public <K> ChangelogKeyedStateBackend<K> createKeyedStateBackend(
+ public <K> CheckpointableKeyedStateBackend<K> createKeyedStateBackend(
Environment env,
JobID jobID,
String operatorIdentifier,
@@ -113,7 +106,7 @@ public class ChangelogStateBackend implements DelegatingStateBackend, Configurab
operatorIdentifier,
keyGroupRange,
ttlTimeProvider,
- stateHandles,
+ castHandles(stateHandles),
baseHandles ->
(AbstractKeyedStateBackend<K>)
delegatedStateBackend.createKeyedStateBackend(
@@ -150,7 +143,7 @@ public class ChangelogStateBackend implements DelegatingStateBackend, Configurab
operatorIdentifier,
keyGroupRange,
ttlTimeProvider,
- stateHandles,
+ castHandles(stateHandles),
baseHandles ->
(AbstractKeyedStateBackend<K>)
delegatedStateBackend.createKeyedStateBackend(
@@ -189,79 +182,14 @@ public class ChangelogStateBackend implements DelegatingStateBackend, Configurab
return delegatedStateBackend;
}
- @Override
- public StateBackend configure(ReadableConfig config, ClassLoader classLoader)
- throws IllegalConfigurationException {
-
- if (delegatedStateBackend instanceof ConfigurableStateBackend) {
- return new ChangelogStateBackend(
- ((ConfigurableStateBackend) delegatedStateBackend)
- .configure(config, classLoader));
- }
-
- return this;
- }
-
- @SuppressWarnings({"unchecked", "rawtypes"})
- private <K> ChangelogKeyedStateBackend<K> restore(
+ protected abstract <K> CheckpointableKeyedStateBackend<K> restore(
Environment env,
String operatorIdentifier,
KeyGroupRange keyGroupRange,
TtlTimeProvider ttlTimeProvider,
- Collection<KeyedStateHandle> stateHandles,
+ Collection<ChangelogStateBackendHandle> stateBackendHandles,
BaseBackendBuilder<K> baseBackendBuilder)
- throws Exception {
- StateChangelogStorage<?> changelogStorage =
- Preconditions.checkNotNull(
- env.getTaskStateManager().getStateChangelogStorage(),
- "Changelog storage is null when creating and restoring"
- + " the ChangelogKeyedStateBackend.");
-
- String subtaskName = env.getTaskInfo().getTaskNameWithSubtasks();
- ExecutionConfig executionConfig = env.getExecutionConfig();
-
- Collection<ChangelogStateBackendHandle> stateBackendHandles = castHandles(stateHandles);
- ChangelogKeyedStateBackend<K> keyedStateBackend =
- ChangelogBackendRestoreOperation.restore(
- changelogStorage.createReader(),
- env.getUserCodeClassLoader().asClassLoader(),
- stateBackendHandles,
- baseBackendBuilder,
- (baseBackend, baseState) ->
- new ChangelogKeyedStateBackend(
- baseBackend,
- subtaskName,
- executionConfig,
- ttlTimeProvider,
- changelogStorage.createWriter(
- operatorIdentifier,
- keyGroupRange,
- env.getMainMailboxExecutor()),
- baseState,
- env.getCheckpointStorageAccess()));
-
- PeriodicMaterializationManager periodicMaterializationManager =
- new PeriodicMaterializationManager(
- checkNotNull(env.getMainMailboxExecutor()),
- checkNotNull(env.getAsyncOperationsThreadPool()),
- subtaskName,
- (message, exception) ->
- env.failExternally(new AsynchronousException(message, exception)),
- keyedStateBackend,
- executionConfig.getPeriodicMaterializeIntervalMillis(),
- executionConfig.getMaterializationMaxAllowedFailures(),
- operatorIdentifier);
-
- // keyedStateBackend is responsible to close periodicMaterializationManager
- // This indicates periodicMaterializationManager binds to the keyedStateBackend
- // However PeriodicMaterializationManager can not be part of keyedStateBackend
- // because of cyclic reference
- keyedStateBackend.registerCloseable(periodicMaterializationManager);
-
- periodicMaterializationManager.start();
-
- return keyedStateBackend;
- }
+ throws Exception;
private Collection<ChangelogStateBackendHandle> castHandles(
Collection<KeyedStateHandle> stateHandles) {
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 07e6bf32f79..eb6204d2ef5 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
@@ -55,14 +55,13 @@ import org.apache.flink.runtime.state.changelog.ChangelogStateHandle;
import org.apache.flink.runtime.state.changelog.SequenceNumber;
import org.apache.flink.runtime.state.changelog.StateChangelogWriter;
import org.apache.flink.runtime.state.heap.HeapPriorityQueueElement;
-import org.apache.flink.runtime.state.heap.InternalKeyContext;
import org.apache.flink.runtime.state.internal.InternalKvState;
import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot.BackendStateType;
import org.apache.flink.runtime.state.metrics.LatencyTrackingStateFactory;
import org.apache.flink.runtime.state.ttl.TtlStateFactory;
import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
+import org.apache.flink.state.changelog.restore.ChangelogRestoreTarget;
import org.apache.flink.state.changelog.restore.FunctionDelegationHelper;
-import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.shaded.guava30.com.google.common.io.Closer;
@@ -81,7 +80,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
-import java.util.NoSuchElementException;
import java.util.Optional;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
@@ -89,7 +87,6 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import java.util.stream.Collectors;
import java.util.stream.Stream;
import static java.util.Collections.emptyList;
@@ -119,25 +116,6 @@ public class ChangelogKeyedStateBackend<K>
new CheckpointOptions(
CheckpointType.CHECKPOINT, CheckpointStorageLocationReference.getDefault());
- private static final Map<StateDescriptor.Type, StateFactory> STATE_FACTORIES =
- Stream.of(
- Tuple2.of(
- StateDescriptor.Type.VALUE,
- (StateFactory) ChangelogValueState::create),
- Tuple2.of(
- StateDescriptor.Type.LIST,
- (StateFactory) ChangelogListState::create),
- Tuple2.of(
- StateDescriptor.Type.REDUCING,
- (StateFactory) ChangelogReducingState::create),
- Tuple2.of(
- StateDescriptor.Type.AGGREGATING,
- (StateFactory) ChangelogAggregatingState::create),
- Tuple2.of(
- StateDescriptor.Type.MAP,
- (StateFactory) ChangelogMapState::create))
- .collect(Collectors.toMap(t -> t.f0, t -> t.f1));
-
/** delegated keyedStateBackend. */
private final AbstractKeyedStateBackend<K> keyedStateBackend;
@@ -147,12 +125,7 @@ public class ChangelogKeyedStateBackend<K>
*/
private final Map<String, InternalKvState<K, ?, ?>> keyValueStatesByName;
- /**
- * Unwrapped changelog states used for recovery (not wrapped into e.g. TTL, latency tracking).
- */
- private final Map<String, ChangelogState> changelogStates;
-
- private final Map<String, ChangelogKeyGroupedPriorityQueue<?>> priorityQueueStatesByName;
+ private final ChangelogStateFactory changelogStateFactory;
private final ExecutionConfig executionConfig;
@@ -217,14 +190,34 @@ public class ChangelogKeyedStateBackend<K>
StateChangelogWriter<? extends ChangelogStateHandle> stateChangelogWriter,
Collection<ChangelogStateBackendHandle> initialState,
CheckpointStorageWorkerView checkpointStorageWorkerView) {
+ this(
+ keyedStateBackend,
+ subtaskName,
+ executionConfig,
+ ttlTimeProvider,
+ stateChangelogWriter,
+ initialState,
+ checkpointStorageWorkerView,
+ new ChangelogStateFactory());
+ }
+
+ @SuppressWarnings("rawtypes")
+ public ChangelogKeyedStateBackend(
+ AbstractKeyedStateBackend<K> keyedStateBackend,
+ String subtaskName,
+ ExecutionConfig executionConfig,
+ TtlTimeProvider ttlTimeProvider,
+ StateChangelogWriter<? extends ChangelogStateHandle> stateChangelogWriter,
+ Collection<ChangelogStateBackendHandle> initialState,
+ CheckpointStorageWorkerView checkpointStorageWorkerView,
+ ChangelogStateFactory changelogStateFactory) {
this.keyedStateBackend = keyedStateBackend;
this.subtaskName = subtaskName;
this.executionConfig = executionConfig;
this.ttlTimeProvider = ttlTimeProvider;
this.keyValueStatesByName = new HashMap<>();
- this.priorityQueueStatesByName = new HashMap<>();
+ this.changelogStateFactory = changelogStateFactory;
this.stateChangelogWriter = stateChangelogWriter;
- this.changelogStates = new HashMap<>();
this.changelogSnapshotState = completeRestore(initialState);
this.streamFactory =
new CheckpointStreamFactory() {
@@ -294,8 +287,7 @@ public class ChangelogKeyedStateBackend<K>
lastName = null;
lastState = null;
keyValueStatesByName.clear();
- changelogStates.clear();
- priorityQueueStatesByName.clear();
+ changelogStateFactory.dispose();
}
@Override
@@ -430,7 +422,9 @@ public class ChangelogKeyedStateBackend<K>
@Nonnull String stateName,
@Nonnull TypeSerializer<T> byteOrderedElementSerializer) {
ChangelogKeyGroupedPriorityQueue<T> queue =
- (ChangelogKeyGroupedPriorityQueue<T>) priorityQueueStatesByName.get(stateName);
+ (ChangelogKeyGroupedPriorityQueue<T>)
+ changelogStateFactory.getExistingState(
+ stateName, BackendStateType.PRIORITY_QUEUE);
if (queue == null) {
PriorityQueueStateChangeLoggerImpl<K, T> priorityQueueStateChangeLogger =
new PriorityQueueStateChangeLoggerImpl<>(
@@ -442,11 +436,11 @@ public class ChangelogKeyedStateBackend<K>
++lastCreatedStateId);
closer.register(priorityQueueStateChangeLogger);
queue =
- new ChangelogKeyGroupedPriorityQueue<>(
+ changelogStateFactory.create(
+ stateName,
keyedStateBackend.create(stateName, byteOrderedElementSerializer),
priorityQueueStateChangeLogger,
byteOrderedElementSerializer);
- priorityQueueStatesByName.put(stateName, queue);
}
return queue;
}
@@ -545,14 +539,6 @@ public class ChangelogKeyedStateBackend<K>
StateSnapshotTransformer.StateSnapshotTransformFactory<SEV>
snapshotTransformFactory)
throws Exception {
- StateFactory stateFactory = STATE_FACTORIES.get(stateDesc.getType());
- if (stateFactory == null) {
- String message =
- String.format(
- "State %s is not supported by %s",
- stateDesc.getClass(), this.getClass());
- throw new FlinkRuntimeException(message);
- }
RegisteredKeyValueStateBackendMetaInfo<N, SV> meta =
new RegisteredKeyValueStateBackendMetaInfo<>(
stateDesc.getType(),
@@ -577,13 +563,13 @@ public class ChangelogKeyedStateBackend<K>
stateDesc.getDefaultValue(),
++lastCreatedStateId);
closer.register(kvStateChangeLogger);
- IS is =
- stateFactory.create(
+ ChangelogState changelogState =
+ changelogStateFactory.create(
+ stateDesc,
state,
kvStateChangeLogger,
keyedStateBackend /* pass the nested backend as key context so that it get key updates on recovery*/);
- changelogStates.put(stateDesc.getName(), (ChangelogState) is);
- return is;
+ return (IS) changelogState;
}
public void registerCloseable(@Nullable Closeable closeable) {
@@ -653,14 +639,7 @@ public class ChangelogKeyedStateBackend<K>
upTo);
// log metadata after materialization is triggered
- for (ChangelogState changelogState : changelogStates.values()) {
- changelogState.resetWritingMetaFlag();
- }
-
- for (ChangelogKeyGroupedPriorityQueue<?> priorityQueueState :
- priorityQueueStatesByName.values()) {
- priorityQueueState.resetWritingMetaFlag();
- }
+ changelogStateFactory.resetAllWritingMetaFlags();
return Optional.of(materializationRunnable);
} else {
@@ -710,45 +689,41 @@ public class ChangelogKeyedStateBackend<K>
return keyedStateBackend.getDelegatedKeyedStateBackend(recursive);
}
- // Factory function interface
- private interface StateFactory {
- <K, N, SV, S extends State, IS extends S> IS create(
- InternalKvState<K, N, SV> kvState,
- KvStateChangeLogger<SV, N> changeLogger,
- InternalKeyContext<K> keyContext)
- throws Exception;
- }
+ public ChangelogRestoreTarget<K> getChangelogRestoreTarget() {
+ return new ChangelogRestoreTarget<K>() {
+ @Override
+ public KeyGroupRange getKeyGroupRange() {
+ return ChangelogKeyedStateBackend.this.getKeyGroupRange();
+ }
- /**
- * @param name state name
- * @param type state type (the only supported type currently are: {@link
- * BackendStateType#KEY_VALUE key value}, {@link BackendStateType#PRIORITY_QUEUE priority
- * queue})
- * @return an existing state, i.e. the one that was already created. The returned state will not
- * apply TTL to the passed values, regardless of the TTL settings. This prevents double
- * applying of TTL (recovered values are TTL values if TTL was enabled). The state will,
- * however, use TTL serializer if TTL is enabled. WARN: only valid during the recovery.
- * @throws NoSuchElementException if the state wasn't created
- * @throws UnsupportedOperationException if state type is not supported
- */
- public ChangelogState getExistingStateForRecovery(String name, BackendStateType type)
- throws NoSuchElementException, UnsupportedOperationException {
- ChangelogState state;
- switch (type) {
- case KEY_VALUE:
- state = changelogStates.get(name);
- break;
- case PRIORITY_QUEUE:
- state = priorityQueueStatesByName.get(name);
- break;
- default:
- throw new UnsupportedOperationException(
- String.format("Unknown state type %s (%s)", type, name));
- }
- if (state == null) {
- throw new NoSuchElementException(String.format("%s state %s not found", type, name));
- }
- return state;
+ @Override
+ public <N, S extends State, V> S createKeyedState(
+ TypeSerializer<N> namespaceSerializer, StateDescriptor<S, V> stateDescriptor)
+ throws Exception {
+ return ChangelogKeyedStateBackend.this.getOrCreateKeyedState(
+ namespaceSerializer, stateDescriptor);
+ }
+
+ @Nonnull
+ @Override
+ public <T extends HeapPriorityQueueElement & PriorityComparable<? super T> & Keyed<?>>
+ KeyGroupedInternalPriorityQueue<T> createPqState(
+ @Nonnull String stateName,
+ @Nonnull TypeSerializer<T> byteOrderedElementSerializer) {
+ return ChangelogKeyedStateBackend.this.create(
+ stateName, byteOrderedElementSerializer);
+ }
+
+ @Override
+ public ChangelogState getExistingState(String name, BackendStateType type) {
+ return changelogStateFactory.getExistingState(name, type);
+ }
+
+ @Override
+ public CheckpointableKeyedStateBackend<K> getRestoredKeyedStateBackend() {
+ return ChangelogKeyedStateBackend.this;
+ }
+ };
}
private static <T> RunnableFuture<T> toRunnableFuture(CompletableFuture<T> f) {
diff --git a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateBackend.java b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateBackend.java
index afcd17aa46c..28d3f93cfbb 100644
--- a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateBackend.java
+++ b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateBackend.java
@@ -20,46 +20,23 @@ package org.apache.flink.state.changelog;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.ReadableConfig;
-import org.apache.flink.core.execution.SavepointFormatType;
-import org.apache.flink.core.fs.CloseableRegistry;
-import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.execution.Environment;
-import org.apache.flink.runtime.query.TaskKvStateRegistry;
-import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
-import org.apache.flink.runtime.state.CheckpointBoundKeyedStateHandle;
import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
import org.apache.flink.runtime.state.ConfigurableStateBackend;
import org.apache.flink.runtime.state.KeyGroupRange;
-import org.apache.flink.runtime.state.KeyedStateHandle;
-import org.apache.flink.runtime.state.OperatorStateBackend;
-import org.apache.flink.runtime.state.OperatorStateHandle;
-import org.apache.flink.runtime.state.SavepointKeyedStateHandle;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle;
-import org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle.ChangelogStateBackendHandleImpl;
import org.apache.flink.runtime.state.changelog.StateChangelogStorage;
-import org.apache.flink.runtime.state.delegate.DelegatingStateBackend;
import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.runtime.taskmanager.AsynchronousException;
import org.apache.flink.state.changelog.restore.ChangelogBackendRestoreOperation;
import org.apache.flink.state.changelog.restore.ChangelogBackendRestoreOperation.BaseBackendBuilder;
import org.apache.flink.util.Preconditions;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.Nonnull;
-
import java.util.Collection;
-import java.util.Objects;
-import java.util.stream.Collectors;
-import static java.util.Collections.emptyList;
-import static java.util.Collections.singletonList;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
@@ -67,126 +44,13 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* state changes to State Changelog.
*/
@Internal
-public class ChangelogStateBackend implements DelegatingStateBackend, ConfigurableStateBackend {
+public class ChangelogStateBackend extends AbstractChangelogStateBackend
+ implements ConfigurableStateBackend {
private static final long serialVersionUID = 1000L;
- private static final Logger LOG = LoggerFactory.getLogger(ChangelogStateBackend.class);
-
- private final StateBackend delegatedStateBackend;
-
- /**
- * Delegate a state backend by a ChangelogStateBackend.
- *
- * <p>As FLINK-22678 mentioned, we currently hide this constructor from user.
- *
- * @param stateBackend the delegated state backend.
- */
ChangelogStateBackend(StateBackend stateBackend) {
- this.delegatedStateBackend = Preconditions.checkNotNull(stateBackend);
-
- Preconditions.checkArgument(
- !(stateBackend instanceof DelegatingStateBackend),
- "Recursive Delegation is not supported.");
-
- LOG.info(
- "ChangelogStateBackend is used, delegating {}.",
- delegatedStateBackend.getClass().getSimpleName());
- }
-
- @Override
- public <K> ChangelogKeyedStateBackend<K> createKeyedStateBackend(
- Environment env,
- JobID jobID,
- String operatorIdentifier,
- TypeSerializer<K> keySerializer,
- int numberOfKeyGroups,
- KeyGroupRange keyGroupRange,
- TaskKvStateRegistry kvStateRegistry,
- TtlTimeProvider ttlTimeProvider,
- MetricGroup metricGroup,
- @Nonnull Collection<KeyedStateHandle> stateHandles,
- CloseableRegistry cancelStreamRegistry)
- throws Exception {
- return restore(
- env,
- operatorIdentifier,
- keyGroupRange,
- ttlTimeProvider,
- stateHandles,
- baseHandles ->
- (AbstractKeyedStateBackend<K>)
- delegatedStateBackend.createKeyedStateBackend(
- env,
- jobID,
- operatorIdentifier,
- keySerializer,
- numberOfKeyGroups,
- keyGroupRange,
- kvStateRegistry,
- ttlTimeProvider,
- metricGroup,
- baseHandles,
- cancelStreamRegistry));
- }
-
- @Override
- public <K> CheckpointableKeyedStateBackend<K> createKeyedStateBackend(
- Environment env,
- JobID jobID,
- String operatorIdentifier,
- TypeSerializer<K> keySerializer,
- int numberOfKeyGroups,
- KeyGroupRange keyGroupRange,
- TaskKvStateRegistry kvStateRegistry,
- TtlTimeProvider ttlTimeProvider,
- MetricGroup metricGroup,
- @Nonnull Collection<KeyedStateHandle> stateHandles,
- CloseableRegistry cancelStreamRegistry,
- double managedMemoryFraction)
- throws Exception {
- return restore(
- env,
- operatorIdentifier,
- keyGroupRange,
- ttlTimeProvider,
- stateHandles,
- baseHandles ->
- (AbstractKeyedStateBackend<K>)
- delegatedStateBackend.createKeyedStateBackend(
- env,
- jobID,
- operatorIdentifier,
- keySerializer,
- numberOfKeyGroups,
- keyGroupRange,
- kvStateRegistry,
- ttlTimeProvider,
- metricGroup,
- baseHandles,
- cancelStreamRegistry,
- managedMemoryFraction));
- }
-
- @Override
- public OperatorStateBackend createOperatorStateBackend(
- Environment env,
- String operatorIdentifier,
- @Nonnull Collection<OperatorStateHandle> stateHandles,
- CloseableRegistry cancelStreamRegistry)
- throws Exception {
- return delegatedStateBackend.createOperatorStateBackend(
- env, operatorIdentifier, stateHandles, cancelStreamRegistry);
- }
-
- @Override
- public boolean useManagedMemory() {
- return delegatedStateBackend.useManagedMemory();
- }
-
- @Override
- public StateBackend getDelegatedStateBackend() {
- return delegatedStateBackend;
+ super(stateBackend);
}
@Override
@@ -202,13 +66,14 @@ public class ChangelogStateBackend implements DelegatingStateBackend, Configurab
return this;
}
+ @Override
@SuppressWarnings({"unchecked", "rawtypes"})
- private <K> ChangelogKeyedStateBackend<K> restore(
+ protected <K> CheckpointableKeyedStateBackend<K> restore(
Environment env,
String operatorIdentifier,
KeyGroupRange keyGroupRange,
TtlTimeProvider ttlTimeProvider,
- Collection<KeyedStateHandle> stateHandles,
+ Collection<ChangelogStateBackendHandle> stateBackendHandles,
BaseBackendBuilder<K> baseBackendBuilder)
throws Exception {
StateChangelogStorage<?> changelogStorage =
@@ -220,26 +85,29 @@ public class ChangelogStateBackend implements DelegatingStateBackend, Configurab
String subtaskName = env.getTaskInfo().getTaskNameWithSubtasks();
ExecutionConfig executionConfig = env.getExecutionConfig();
- Collection<ChangelogStateBackendHandle> stateBackendHandles = castHandles(stateHandles);
- ChangelogKeyedStateBackend<K> keyedStateBackend =
+ ChangelogStateFactory changelogStateFactory = new ChangelogStateFactory();
+ CheckpointableKeyedStateBackend<K> keyedStateBackend =
ChangelogBackendRestoreOperation.restore(
- changelogStorage.createReader(),
env.getUserCodeClassLoader().asClassLoader(),
stateBackendHandles,
baseBackendBuilder,
(baseBackend, baseState) ->
new ChangelogKeyedStateBackend(
- baseBackend,
- subtaskName,
- executionConfig,
- ttlTimeProvider,
- changelogStorage.createWriter(
- operatorIdentifier,
- keyGroupRange,
- env.getMainMailboxExecutor()),
- baseState,
- env.getCheckpointStorageAccess()));
-
+ baseBackend,
+ subtaskName,
+ executionConfig,
+ ttlTimeProvider,
+ changelogStorage.createWriter(
+ operatorIdentifier,
+ keyGroupRange,
+ env.getMainMailboxExecutor()),
+ baseState,
+ env.getCheckpointStorageAccess(),
+ changelogStateFactory)
+ .getChangelogRestoreTarget());
+
+ ChangelogKeyedStateBackend<K> changelogKeyedStateBackend =
+ (ChangelogKeyedStateBackend<K>) keyedStateBackend;
PeriodicMaterializationManager periodicMaterializationManager =
new PeriodicMaterializationManager(
checkNotNull(env.getMainMailboxExecutor()),
@@ -247,7 +115,7 @@ public class ChangelogStateBackend implements DelegatingStateBackend, Configurab
subtaskName,
(message, exception) ->
env.failExternally(new AsynchronousException(message, exception)),
- keyedStateBackend,
+ changelogKeyedStateBackend,
executionConfig.getPeriodicMaterializeIntervalMillis(),
executionConfig.getMaterializationMaxAllowedFailures(),
operatorIdentifier);
@@ -256,50 +124,10 @@ public class ChangelogStateBackend implements DelegatingStateBackend, Configurab
// This indicates periodicMaterializationManager binds to the keyedStateBackend
// However PeriodicMaterializationManager can not be part of keyedStateBackend
// because of cyclic reference
- keyedStateBackend.registerCloseable(periodicMaterializationManager);
+ changelogKeyedStateBackend.registerCloseable(periodicMaterializationManager);
periodicMaterializationManager.start();
return keyedStateBackend;
}
-
- private Collection<ChangelogStateBackendHandle> castHandles(
- Collection<KeyedStateHandle> stateHandles) {
- if (stateHandles.stream().anyMatch(h -> !(h instanceof ChangelogStateBackendHandle))) {
- LOG.warn(
- "Some state handles do not contain changelog: {} (ok if recovery from a savepoint)",
- stateHandles);
- }
- return stateHandles.stream()
- .filter(Objects::nonNull)
- .map(this::getChangelogStateBackendHandle)
- .collect(Collectors.toList());
- }
-
- private ChangelogStateBackendHandle getChangelogStateBackendHandle(
- KeyedStateHandle keyedStateHandle) {
- if (keyedStateHandle instanceof ChangelogStateBackendHandle) {
- return (ChangelogStateBackendHandle) keyedStateHandle;
- } else if (keyedStateHandle instanceof SavepointKeyedStateHandle) {
- return new ChangelogStateBackendHandleImpl(
- singletonList(keyedStateHandle),
- emptyList(),
- keyedStateHandle.getKeyGroupRange(),
- getMaterializationID(keyedStateHandle),
- 0L);
- } else {
- throw new IllegalStateException(
- String.format(
- "Recovery not supported from %s with Changelog enabled. Consider taking a savepoint in %s format.",
- keyedStateHandle.getClass(), SavepointFormatType.CANONICAL));
- }
- }
-
- private long getMaterializationID(KeyedStateHandle keyedStateHandle) {
- if (keyedStateHandle instanceof CheckpointBoundKeyedStateHandle) {
- return ((CheckpointBoundKeyedStateHandle) keyedStateHandle).getCheckpointId();
- } else {
- return 0L;
- }
- }
}
diff --git a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateFactory.java b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateFactory.java
new file mode 100644
index 00000000000..52aaa7fad3f
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateFactory.java
@@ -0,0 +1,158 @@
+/*
+ * 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.state.State;
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.runtime.state.heap.InternalKeyContext;
+import org.apache.flink.runtime.state.internal.InternalKvState;
+import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/** Maintains the lifecycle of all {@link ChangelogState}s. */
+public class ChangelogStateFactory {
+
+ /**
+ * Unwrapped changelog states used for recovery (not wrapped into e.g. TTL, latency tracking).
+ */
+ private final Map<String, ChangelogState> changelogStates;
+
+ private final Map<String, ChangelogKeyGroupedPriorityQueue<?>> priorityQueueStatesByName;
+
+ public ChangelogStateFactory() {
+ this.changelogStates = new HashMap<>();
+ this.priorityQueueStatesByName = new HashMap<>();
+ }
+
+ private static final Map<StateDescriptor.Type, StateFactory> STATE_FACTORIES =
+ Stream.of(
+ Tuple2.of(
+ StateDescriptor.Type.VALUE,
+ (StateFactory) ChangelogValueState::create),
+ Tuple2.of(
+ StateDescriptor.Type.LIST,
+ (StateFactory) ChangelogListState::create),
+ Tuple2.of(
+ StateDescriptor.Type.REDUCING,
+ (StateFactory) ChangelogReducingState::create),
+ Tuple2.of(
+ StateDescriptor.Type.AGGREGATING,
+ (StateFactory) ChangelogAggregatingState::create),
+ Tuple2.of(
+ StateDescriptor.Type.MAP,
+ (StateFactory) ChangelogMapState::create))
+ .collect(Collectors.toMap(t -> t.f0, t -> t.f1));
+
+ public <K, N, V, S extends State> ChangelogState create(
+ StateDescriptor<S, V> stateDescriptor,
+ InternalKvState<K, N, V> internalKvState,
+ KvStateChangeLogger<V, N> kvStateChangeLogger,
+ InternalKeyContext<K> keyContext)
+ throws Exception {
+ ChangelogState changelogState =
+ getStateFactory(stateDescriptor)
+ .create(internalKvState, kvStateChangeLogger, keyContext);
+ changelogStates.put(stateDescriptor.getName(), changelogState);
+ return changelogState;
+ }
+
+ public <T> ChangelogKeyGroupedPriorityQueue<T> create(
+ String stateName,
+ KeyGroupedInternalPriorityQueue<T> internalPriorityQueue,
+ StateChangeLogger<T, Void> logger,
+ TypeSerializer<T> serializer) {
+ ChangelogKeyGroupedPriorityQueue<T> changelogKeyGroupedPriorityQueue =
+ new ChangelogKeyGroupedPriorityQueue<>(internalPriorityQueue, logger, serializer);
+ priorityQueueStatesByName.put(stateName, changelogKeyGroupedPriorityQueue);
+ return changelogKeyGroupedPriorityQueue;
+ }
+
+ /**
+ * @param name state name
+ * @param type state type (the only supported type currently are: {@link
+ * StateMetaInfoSnapshot.BackendStateType#KEY_VALUE key value}, {@link
+ * StateMetaInfoSnapshot.BackendStateType#PRIORITY_QUEUE priority queue})
+ * @return an existing state, i.e. the one that was already created. The returned state will not
+ * apply TTL to the passed values, regardless of the TTL settings. This prevents double
+ * applying of TTL (recovered values are TTL values if TTL was enabled). The state will,
+ * however, use TTL serializer if TTL is enabled. WARN: only valid during the recovery.
+ * @throws UnsupportedOperationException if state type is not supported
+ */
+ public ChangelogState getExistingState(String name, StateMetaInfoSnapshot.BackendStateType type)
+ throws UnsupportedOperationException {
+ ChangelogState state;
+ switch (type) {
+ case KEY_VALUE:
+ state = changelogStates.get(name);
+ break;
+ case PRIORITY_QUEUE:
+ state = priorityQueueStatesByName.get(name);
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format("Unknown state type %s (%s)", type, name));
+ }
+ return state;
+ }
+
+ public void resetAllWritingMetaFlags() {
+ for (ChangelogState changelogState : changelogStates.values()) {
+ changelogState.resetWritingMetaFlag();
+ }
+
+ for (ChangelogKeyGroupedPriorityQueue<?> priorityQueueState :
+ priorityQueueStatesByName.values()) {
+ priorityQueueState.resetWritingMetaFlag();
+ }
+ }
+
+ public void dispose() {
+ changelogStates.clear();
+ priorityQueueStatesByName.clear();
+ }
+
+ private <S extends State, V> StateFactory getStateFactory(
+ StateDescriptor<S, V> stateDescriptor) {
+ StateFactory stateFactory = STATE_FACTORIES.get(stateDescriptor.getType());
+ if (stateFactory == null) {
+ String message =
+ String.format(
+ "State %s is not supported by %s",
+ stateDescriptor.getClass(), ChangelogKeyedStateBackend.class);
+ throw new FlinkRuntimeException(message);
+ }
+ return stateFactory;
+ }
+
+ // Factory function interface
+ private interface StateFactory {
+ <K, N, SV, S extends State, IS extends S> IS create(
+ InternalKvState<K, N, SV> kvState,
+ KvStateChangeLogger<SV, N> changeLogger,
+ InternalKeyContext<K> keyContext)
+ throws Exception;
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/DeactivatedChangelogStateBackend.java b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/DeactivatedChangelogStateBackend.java
new file mode 100644
index 00000000000..e0b8815dac3
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/DeactivatedChangelogStateBackend.java
@@ -0,0 +1,64 @@
+/*
+ * 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.annotation.Internal;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle;
+import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
+import org.apache.flink.state.changelog.restore.ChangelogBackendRestoreOperation;
+import org.apache.flink.state.changelog.restore.ChangelogBackendRestoreOperation.BaseBackendBuilder;
+import org.apache.flink.state.changelog.restore.ChangelogMigrationRestoreTarget;
+
+import java.util.Collection;
+
+/**
+ * This state backend use delegatedStateBackend and State changes to restore to the
+ * delegatedStateBackend in which switching Changelog from enabled to disabled.
+ */
+@Internal
+public class DeactivatedChangelogStateBackend extends AbstractChangelogStateBackend {
+
+ private static final long serialVersionUID = 1000L;
+
+ DeactivatedChangelogStateBackend(StateBackend stateBackend) {
+ super(stateBackend);
+ }
+
+ @Override
+ protected <K> CheckpointableKeyedStateBackend<K> restore(
+ Environment env,
+ String operatorIdentifier,
+ KeyGroupRange keyGroupRange,
+ TtlTimeProvider ttlTimeProvider,
+ Collection<ChangelogStateBackendHandle> stateBackendHandles,
+ BaseBackendBuilder<K> baseBackendBuilder)
+ throws Exception {
+ ChangelogStateFactory changelogStateFactory = new ChangelogStateFactory();
+ return ChangelogBackendRestoreOperation.restore(
+ env.getUserCodeClassLoader().asClassLoader(),
+ stateBackendHandles,
+ baseBackendBuilder,
+ (baseBackend, baseState) ->
+ new ChangelogMigrationRestoreTarget<>(baseBackend, changelogStateFactory));
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/KvStateChangeLogger.java b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/KvStateChangeLogger.java
index 8ff7c15fa40..256af90c656 100644
--- a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/KvStateChangeLogger.java
+++ b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/KvStateChangeLogger.java
@@ -17,6 +17,8 @@
package org.apache.flink.state.changelog;
+import org.apache.flink.annotation.Internal;
+
import java.io.IOException;
import java.util.Collection;
@@ -27,7 +29,8 @@ import java.util.Collection;
* @param <Value> the type of state values
* @param <Namespace> the type of namespace
*/
-interface KvStateChangeLogger<Value, Namespace> extends StateChangeLogger<Value, Namespace> {
+@Internal
+public interface KvStateChangeLogger<Value, Namespace> extends StateChangeLogger<Value, Namespace> {
void namespacesMerged(Namespace target, Collection<Namespace> sources) throws IOException;
}
diff --git a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/StateChangeLogger.java b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/StateChangeLogger.java
index 111af14f651..ed12624cb05 100644
--- a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/StateChangeLogger.java
+++ b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/StateChangeLogger.java
@@ -17,6 +17,7 @@
package org.apache.flink.state.changelog;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
import org.apache.flink.util.function.ThrowingConsumer;
@@ -42,7 +43,8 @@ import java.io.IOException;
* @param <Value> type of state (value)
* @param <Namespace> type of namespace
*/
-interface StateChangeLogger<Value, Namespace> extends Closeable {
+@Internal
+public interface StateChangeLogger<Value, Namespace> extends Closeable {
/** State updated, such as by {@link ListState#update}. */
void valueUpdated(Value newValue, Namespace ns) throws IOException;
diff --git a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogBackendLogApplier.java b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogBackendLogApplier.java
index 1e5681859cd..85535ea45fe 100644
--- a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogBackendLogApplier.java
+++ b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogBackendLogApplier.java
@@ -40,6 +40,7 @@ import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshotReadersWrite
import org.apache.flink.state.changelog.ChangelogKeyedStateBackend;
import org.apache.flink.state.changelog.ChangelogState;
import org.apache.flink.state.changelog.StateChangeOperation;
+import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -64,7 +65,7 @@ class ChangelogBackendLogApplier {
public static void apply(
StateChange stateChange,
- ChangelogKeyedStateBackend<?> changelogBackend,
+ ChangelogRestoreTarget<?> changelogRestoreTarget,
ClassLoader classLoader,
Map<Short, StateID> stateIds)
throws Exception {
@@ -73,7 +74,7 @@ class ChangelogBackendLogApplier {
applyOperation(
StateChangeOperation.byCode(in.readByte()),
stateChange.getKeyGroup(),
- changelogBackend,
+ changelogRestoreTarget,
in,
classLoader,
ChangelogApplierFactoryImpl.INSTANCE,
@@ -83,7 +84,7 @@ class ChangelogBackendLogApplier {
private static void applyOperation(
StateChangeOperation operation,
int keyGroup,
- ChangelogKeyedStateBackend<?> backend,
+ ChangelogRestoreTarget<?> changelogRestoreTarget,
DataInputView in,
ClassLoader classLoader,
ChangelogApplierFactory factory,
@@ -91,15 +92,15 @@ class ChangelogBackendLogApplier {
throws Exception {
LOG.debug("apply {} in key group {}", operation, keyGroup);
if (operation == METADATA) {
- applyMetaDataChange(in, backend, classLoader, stateIds);
- } else if (backend.getKeyGroupRange().contains(keyGroup)) {
- applyDataChange(in, factory, backend, operation, stateIds);
+ applyMetaDataChange(in, changelogRestoreTarget, classLoader, stateIds);
+ } else if (changelogRestoreTarget.getKeyGroupRange().contains(keyGroup)) {
+ applyDataChange(in, factory, changelogRestoreTarget, operation, stateIds);
}
}
private static void applyMetaDataChange(
DataInputView in,
- ChangelogKeyedStateBackend<?> backend,
+ ChangelogRestoreTarget<?> changelogRestoreTarget,
ClassLoader classLoader,
Map<Short, StateID> stateIds)
throws Exception {
@@ -108,10 +109,10 @@ class ChangelogBackendLogApplier {
RegisteredStateMetaInfoBase meta;
switch (snapshot.getBackendStateType()) {
case KEY_VALUE:
- meta = restoreKvMetaData(backend, snapshot, in);
+ meta = restoreKvMetaData(changelogRestoreTarget, snapshot, in);
break;
case PRIORITY_QUEUE:
- meta = restorePqMetaData(backend, snapshot);
+ meta = restorePqMetaData(changelogRestoreTarget, snapshot);
break;
default:
throw new RuntimeException(
@@ -147,7 +148,9 @@ class ChangelogBackendLogApplier {
}
private static RegisteredKeyValueStateBackendMetaInfo restoreKvMetaData(
- ChangelogKeyedStateBackend<?> backend, StateMetaInfoSnapshot snapshot, DataInputView in)
+ ChangelogRestoreTarget<?> changelogRestoreTarget,
+ StateMetaInfoSnapshot snapshot,
+ DataInputView in)
throws Exception {
RegisteredKeyValueStateBackendMetaInfo meta =
new RegisteredKeyValueStateBackendMetaInfo(snapshot);
@@ -162,7 +165,7 @@ class ChangelogBackendLogApplier {
if (ttlConfig.isEnabled()) {
stateDescriptor.enableTimeToLive(ttlConfig);
}
- backend.getOrCreateKeyedState(meta.getNamespaceSerializer(), stateDescriptor);
+ changelogRestoreTarget.createKeyedState(meta.getNamespaceSerializer(), stateDescriptor);
return meta;
}
@@ -194,10 +197,10 @@ class ChangelogBackendLogApplier {
}
private static RegisteredPriorityQueueStateBackendMetaInfo restorePqMetaData(
- ChangelogKeyedStateBackend<?> backend, StateMetaInfoSnapshot snapshot) {
+ ChangelogRestoreTarget<?> changelogRestoreTarget, StateMetaInfoSnapshot snapshot) {
RegisteredPriorityQueueStateBackendMetaInfo meta =
new RegisteredPriorityQueueStateBackendMetaInfo(snapshot);
- backend.create(meta.getName(), meta.getElementSerializer());
+ changelogRestoreTarget.createPqState(meta.getName(), meta.getElementSerializer());
return meta;
}
@@ -212,12 +215,14 @@ class ChangelogBackendLogApplier {
private static void applyDataChange(
DataInputView in,
ChangelogApplierFactory factory,
- ChangelogKeyedStateBackend<?> backend,
+ ChangelogRestoreTarget<?> changelogRestoreTarget,
StateChangeOperation operation,
Map<Short, StateID> stateIds)
throws Exception {
StateID id = checkNotNull(stateIds.get(in.readShort()));
- ChangelogState state = backend.getExistingStateForRecovery(id.stateName, id.stateType);
+ ChangelogState state = changelogRestoreTarget.getExistingState(id.stateName, id.stateType);
+ Preconditions.checkState(
+ state != null, String.format("%s state %s not found", id.stateType, id.stateName));
StateChangeApplier changeApplier = state.getChangeApplier(factory);
changeApplier.apply(operation, in);
}
diff --git a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogBackendRestoreOperation.java b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogBackendRestoreOperation.java
index d01ba0658b2..832ca0fff40 100644
--- a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogBackendRestoreOperation.java
+++ b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogBackendRestoreOperation.java
@@ -19,11 +19,13 @@ package org.apache.flink.state.changelog.restore;
import org.apache.flink.annotation.Internal;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle;
import org.apache.flink.runtime.state.changelog.ChangelogStateHandle;
import org.apache.flink.runtime.state.changelog.StateChange;
import org.apache.flink.runtime.state.changelog.StateChangelogHandleReader;
+import org.apache.flink.runtime.state.changelog.StateChangelogStorageLoader;
import org.apache.flink.state.changelog.ChangelogKeyedStateBackend;
import org.apache.flink.util.CloseableIterator;
import org.apache.flink.util.Preconditions;
@@ -48,50 +50,52 @@ public class ChangelogBackendRestoreOperation {
extends FunctionWithException<
Collection<KeyedStateHandle>, AbstractKeyedStateBackend<K>, Exception> {}
- /** Builds {@link ChangelogKeyedStateBackend} from the base backend and state. */
+ /** Builds {@link ChangelogRestoreTarget} from the base backend and state. */
@FunctionalInterface
- public interface DeltaBackendBuilder<K>
+ public interface ChangelogRestoreTargetBuilder<K>
extends BiFunctionWithException<
AbstractKeyedStateBackend<K>,
Collection<ChangelogStateBackendHandle>,
- ChangelogKeyedStateBackend<K>,
+ ChangelogRestoreTarget<K>,
Exception> {}
- public static <K, T extends ChangelogStateHandle> ChangelogKeyedStateBackend<K> restore(
- StateChangelogHandleReader<T> changelogHandleReader,
+ public static <K> CheckpointableKeyedStateBackend<K> restore(
ClassLoader classLoader,
Collection<ChangelogStateBackendHandle> stateHandles,
BaseBackendBuilder<K> baseBackendBuilder,
- DeltaBackendBuilder<K> changelogBackendBuilder)
+ ChangelogRestoreTargetBuilder<K> changelogRestoreTargetBuilder)
throws Exception {
Collection<KeyedStateHandle> baseState = extractBaseState(stateHandles);
AbstractKeyedStateBackend<K> baseBackend = baseBackendBuilder.apply(baseState);
- ChangelogKeyedStateBackend<K> changelogBackend =
- changelogBackendBuilder.apply(baseBackend, stateHandles);
+ ChangelogRestoreTarget<K> changelogRestoreTarget =
+ changelogRestoreTargetBuilder.apply(baseBackend, stateHandles);
for (ChangelogStateBackendHandle handle : stateHandles) {
if (handle != null) { // null is empty state (no change)
- readBackendHandle(changelogBackend, handle, changelogHandleReader, classLoader);
+ readBackendHandle(changelogRestoreTarget, handle, classLoader);
}
}
- return changelogBackend;
+ return changelogRestoreTarget.getRestoredKeyedStateBackend();
}
@SuppressWarnings("unchecked")
private static <T extends ChangelogStateHandle> void readBackendHandle(
- ChangelogKeyedStateBackend<?> backend,
+ ChangelogRestoreTarget<?> changelogRestoreTarget,
ChangelogStateBackendHandle backendHandle,
- StateChangelogHandleReader<T> changelogHandleReader,
ClassLoader classLoader)
throws Exception {
Map<Short, StateID> stateIds = new HashMap<>();
for (ChangelogStateHandle changelogHandle :
backendHandle.getNonMaterializedStateHandles()) {
+ StateChangelogHandleReader<T> changelogHandleReader =
+ (StateChangelogHandleReader<T>)
+ StateChangelogStorageLoader.loadFromStateHandle(changelogHandle)
+ .createReader();
try (CloseableIterator<StateChange> changes =
changelogHandleReader.getChanges((T) changelogHandle)) {
while (changes.hasNext()) {
ChangelogBackendLogApplier.apply(
- changes.next(), backend, classLoader, stateIds);
+ changes.next(), changelogRestoreTarget, classLoader, stateIds);
}
}
}
diff --git a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogMigrationRestoreTarget.java b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogMigrationRestoreTarget.java
new file mode 100644
index 00000000000..50b1fe8361b
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogMigrationRestoreTarget.java
@@ -0,0 +1,298 @@
+/*
+ * 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.restore;
+
+import org.apache.flink.api.common.state.State;
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.runtime.state.Keyed;
+import org.apache.flink.runtime.state.KeyedStateHandle;
+import org.apache.flink.runtime.state.PriorityComparable;
+import org.apache.flink.runtime.state.SavepointResources;
+import org.apache.flink.runtime.state.SnapshotResult;
+import org.apache.flink.runtime.state.StateSnapshotTransformer;
+import org.apache.flink.runtime.state.heap.HeapPriorityQueueElement;
+import org.apache.flink.runtime.state.internal.InternalKvState;
+import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
+import org.apache.flink.state.changelog.ChangelogKeyGroupedPriorityQueue;
+import org.apache.flink.state.changelog.ChangelogState;
+import org.apache.flink.state.changelog.ChangelogStateFactory;
+import org.apache.flink.state.changelog.KvStateChangeLogger;
+import org.apache.flink.state.changelog.StateChangeLogger;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.concurrent.RunnableFuture;
+import java.util.stream.Stream;
+
+/** A {@link ChangelogRestoreTarget} supports to migrate to the delegated keyed state backend. */
+public class ChangelogMigrationRestoreTarget<K> implements ChangelogRestoreTarget<K> {
+
+ private final AbstractKeyedStateBackend<K> keyedStateBackend;
+
+ private final ChangelogStateFactory changelogStateFactory;
+
+ private final FunctionDelegationHelper functionDelegationHelper =
+ new FunctionDelegationHelper();
+
+ public ChangelogMigrationRestoreTarget(
+ AbstractKeyedStateBackend<K> keyedStateBackend,
+ ChangelogStateFactory changelogStateFactory) {
+ this.keyedStateBackend = keyedStateBackend;
+ this.changelogStateFactory = changelogStateFactory;
+ }
+
+ @Override
+ public KeyGroupRange getKeyGroupRange() {
+ return keyedStateBackend.getKeyGroupRange();
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <N, S extends State, V> S createKeyedState(
+ TypeSerializer<N> namespaceSerializer, StateDescriptor<S, V> stateDescriptor)
+ throws Exception {
+ S keyedState =
+ keyedStateBackend.getOrCreateKeyedState(namespaceSerializer, stateDescriptor);
+ functionDelegationHelper.addOrUpdate(stateDescriptor);
+ final InternalKvState<K, N, V> kvState = (InternalKvState<K, N, V>) keyedState;
+ ChangelogState changelogState =
+ changelogStateFactory.create(
+ stateDescriptor,
+ kvState,
+ VoidStateChangeLogger.getInstance(),
+ keyedStateBackend);
+ return (S) changelogState;
+ }
+
+ @Nonnull
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends HeapPriorityQueueElement & PriorityComparable<? super T> & Keyed<?>>
+ KeyGroupedInternalPriorityQueue<T> createPqState(
+ @Nonnull String stateName,
+ @Nonnull TypeSerializer<T> byteOrderedElementSerializer) {
+ ChangelogKeyGroupedPriorityQueue<T> queue =
+ (ChangelogKeyGroupedPriorityQueue<T>)
+ changelogStateFactory.getExistingState(
+ stateName, StateMetaInfoSnapshot.BackendStateType.PRIORITY_QUEUE);
+ if (queue == null) {
+ KeyGroupedInternalPriorityQueue<T> internalPriorityQueue =
+ keyedStateBackend.create(stateName, byteOrderedElementSerializer);
+ queue =
+ changelogStateFactory.create(
+ stateName,
+ internalPriorityQueue,
+ VoidStateChangeLogger.getInstance(),
+ byteOrderedElementSerializer);
+ }
+ return queue;
+ }
+
+ @Override
+ public ChangelogState getExistingState(
+ String name, StateMetaInfoSnapshot.BackendStateType type) {
+ return changelogStateFactory.getExistingState(name, type);
+ }
+
+ @Override
+ public CheckpointableKeyedStateBackend<K> getRestoredKeyedStateBackend() {
+ // TODO: This inner class make the behaviour of the method of create consistent with
+ // the method of getOrCreateKeyedState currently which could be removed
+ // after we support state migration (in FLINK-23143).
+ // It is also used to maintain FunctionDelegationHelper in the delegated state backend.
+ return wrapKeyedStateBackend(
+ keyedStateBackend, changelogStateFactory, functionDelegationHelper);
+ }
+
+ private static class VoidStateChangeLogger<Value, Namespace>
+ implements KvStateChangeLogger<Value, Namespace>, StateChangeLogger<Value, Namespace> {
+
+ private static final VoidStateChangeLogger<Object, Object> INSTANCE =
+ new VoidStateChangeLogger<>();
+
+ @SuppressWarnings("unchecked")
+ public static <Value, Namespace> VoidStateChangeLogger<Value, Namespace> getInstance() {
+ return (VoidStateChangeLogger<Value, Namespace>) INSTANCE;
+ }
+
+ private VoidStateChangeLogger() {}
+
+ @Override
+ public void namespacesMerged(Namespace target, Collection<Namespace> sources)
+ throws IOException {}
+
+ @Override
+ public void valueUpdated(Value newValue, Namespace ns) throws IOException {}
+
+ @Override
+ public void valueUpdatedInternal(Value newValue, Namespace ns) throws IOException {}
+
+ @Override
+ public void valueAdded(Value addedValue, Namespace ns) throws IOException {}
+
+ @Override
+ public void valueCleared(Namespace ns) throws IOException {}
+
+ @Override
+ public void valueElementAdded(
+ ThrowingConsumer<DataOutputViewStreamWrapper, IOException> dataSerializer,
+ Namespace ns)
+ throws IOException {}
+
+ @Override
+ public void valueElementAddedOrUpdated(
+ ThrowingConsumer<DataOutputViewStreamWrapper, IOException> dataSerializer,
+ Namespace ns)
+ throws IOException {}
+
+ @Override
+ public void valueElementRemoved(
+ ThrowingConsumer<DataOutputViewStreamWrapper, IOException> dataSerializer,
+ Namespace ns)
+ throws IOException {}
+
+ @Override
+ public void resetWritingMetaFlag() {}
+
+ @Override
+ public void close() throws IOException {}
+ }
+
+ private static <K> AbstractKeyedStateBackend<K> wrapKeyedStateBackend(
+ AbstractKeyedStateBackend<K> keyedStateBackend,
+ ChangelogStateFactory changelogStateFactory,
+ FunctionDelegationHelper functionDelegationHelper) {
+ return new AbstractKeyedStateBackend<K>(keyedStateBackend) {
+
+ @Override
+ public void setCurrentKey(K newKey) {
+ keyedStateBackend.setCurrentKey(newKey);
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ keyedStateBackend.notifyCheckpointComplete(checkpointId);
+ }
+
+ @Nonnull
+ @Override
+ public SavepointResources<K> savepoint() throws Exception {
+ return keyedStateBackend.savepoint();
+ }
+
+ @Override
+ public int numKeyValueStateEntries() {
+ return keyedStateBackend.numKeyValueStateEntries();
+ }
+
+ @Override
+ public <N> Stream<K> getKeys(String state, N namespace) {
+ return keyedStateBackend.getKeys(state, namespace);
+ }
+
+ @Override
+ public <N> Stream<Tuple2<K, N>> getKeysAndNamespaces(String state) {
+ return keyedStateBackend.getKeysAndNamespaces(state);
+ }
+
+ @Nonnull
+ @Override
+ public <N, SV, SEV, S extends State, IS extends S> IS createInternalState(
+ @Nonnull TypeSerializer<N> namespaceSerializer,
+ @Nonnull StateDescriptor<S, SV> stateDesc,
+ @Nonnull
+ StateSnapshotTransformer.StateSnapshotTransformFactory<SEV>
+ snapshotTransformFactory)
+ throws Exception {
+ return keyedStateBackend.createInternalState(
+ namespaceSerializer, stateDesc, snapshotTransformFactory);
+ }
+
+ @Override
+ public <N, S extends State> S getPartitionedState(
+ N namespace,
+ TypeSerializer<N> namespaceSerializer,
+ StateDescriptor<S, ?> stateDescriptor)
+ throws Exception {
+ S partitionedState =
+ keyedStateBackend.getPartitionedState(
+ namespace, namespaceSerializer, stateDescriptor);
+ functionDelegationHelper.addOrUpdate(stateDescriptor);
+ return partitionedState;
+ }
+
+ @Override
+ public <N, S extends State, V> S getOrCreateKeyedState(
+ TypeSerializer<N> namespaceSerializer, StateDescriptor<S, V> stateDescriptor)
+ throws Exception {
+ S keyedState =
+ keyedStateBackend.getOrCreateKeyedState(
+ namespaceSerializer, stateDescriptor);
+ functionDelegationHelper.addOrUpdate(stateDescriptor);
+ return keyedState;
+ }
+
+ @Nonnull
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends HeapPriorityQueueElement & PriorityComparable<? super T> & Keyed<?>>
+ KeyGroupedInternalPriorityQueue<T> create(
+ @Nonnull String stateName,
+ @Nonnull TypeSerializer<T> byteOrderedElementSerializer) {
+ ChangelogKeyGroupedPriorityQueue<T> existingState =
+ (ChangelogKeyGroupedPriorityQueue<T>)
+ changelogStateFactory.getExistingState(
+ stateName,
+ StateMetaInfoSnapshot.BackendStateType.PRIORITY_QUEUE);
+ return existingState == null
+ ? keyedStateBackend.create(stateName, byteOrderedElementSerializer)
+ : existingState;
+ }
+
+ @Nonnull
+ @Override
+ public RunnableFuture<SnapshotResult<KeyedStateHandle>> snapshot(
+ long checkpointId,
+ long timestamp,
+ @Nonnull CheckpointStreamFactory streamFactory,
+ @Nonnull CheckpointOptions checkpointOptions)
+ throws Exception {
+ return keyedStateBackend.snapshot(
+ checkpointId, timestamp, streamFactory, checkpointOptions);
+ }
+
+ @Override
+ public void dispose() {
+ keyedStateBackend.dispose();
+ changelogStateFactory.dispose();
+ }
+ };
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogRestoreTarget.java b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogRestoreTarget.java
new file mode 100644
index 00000000000..94453a5f935
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogRestoreTarget.java
@@ -0,0 +1,72 @@
+/*
+ * 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.restore;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.state.State;
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.runtime.state.Keyed;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.PriorityComparable;
+import org.apache.flink.runtime.state.PriorityQueueSetFactory;
+import org.apache.flink.runtime.state.heap.HeapPriorityQueueElement;
+import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot.BackendStateType;
+import org.apache.flink.state.changelog.ChangelogState;
+
+import javax.annotation.Nonnull;
+
+/** Maintains metadata operation related to Changelog recovery. */
+@Internal
+public interface ChangelogRestoreTarget<K> {
+
+ /** Returns the key groups which this restore procedure covers. */
+ KeyGroupRange getKeyGroupRange();
+
+ /**
+ * Creates a keyed state which could be retrieved by {@link #getExistingState(String,
+ * BackendStateType)} in the restore procedure. The interface comes from {@link
+ * KeyedStateBackend#getOrCreateKeyedState(TypeSerializer, StateDescriptor)}.
+ */
+ <N, S extends State, V> S createKeyedState(
+ TypeSerializer<N> namespaceSerializer, StateDescriptor<S, V> stateDescriptor)
+ throws Exception;
+
+ /**
+ * Creates a {@link KeyGroupedInternalPriorityQueue} which could be retrieved by {@link
+ * #getExistingState(String, BackendStateType)} in the restore procedure. The interface comes
+ * from {@link PriorityQueueSetFactory#create(String, TypeSerializer)}.
+ */
+ @Nonnull
+ <T extends HeapPriorityQueueElement & PriorityComparable<? super T> & Keyed<?>>
+ KeyGroupedInternalPriorityQueue<T> createPqState(
+ @Nonnull String stateName,
+ @Nonnull TypeSerializer<T> byteOrderedElementSerializer);
+
+ /**
+ * Returns the existing state created by {@link #createKeyedState(TypeSerializer,
+ * StateDescriptor)} or {@link #createPqState(String, TypeSerializer)} in the restore procedure.
+ */
+ ChangelogState getExistingState(String name, BackendStateType type);
+
+ /** Returns keyed state backend restored finally. */
+ CheckpointableKeyedStateBackend<K> getRestoredKeyedStateBackend();
+}
diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateBackendLoadingTest.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateBackendLoadingTest.java
index 5fa4a3763fc..6d65b625be4 100644
--- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateBackendLoadingTest.java
+++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateBackendLoadingTest.java
@@ -34,6 +34,7 @@ import org.apache.flink.runtime.jobgraph.tasks.JobCheckpointingSettings;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.AbstractStateBackend;
+import org.apache.flink.runtime.state.ChangelogTestUtils;
import org.apache.flink.runtime.state.CheckpointStorage;
import org.apache.flink.runtime.state.CheckpointStorageAccess;
import org.apache.flink.runtime.state.CheckpointStorageLoader;
@@ -64,6 +65,7 @@ import org.junit.rules.TemporaryFolder;
import javax.annotation.Nonnull;
import java.util.Collection;
+import java.util.Collections;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
@@ -144,6 +146,18 @@ public class ChangelogStateBackendLoadingTest {
assertTrue(backend instanceof HashMapStateBackend);
}
+ @Test
+ public void testLoadingChangelogForRecovery() throws Exception {
+ final StateBackend backend =
+ StateBackendLoader.loadStateBackendFromKeyedStateHandles(
+ new MockStateBackend(),
+ cl,
+ Collections.singletonList(
+ ChangelogTestUtils.createChangelogStateBackendHandle()));
+
+ assertTrue(backend instanceof DeactivatedChangelogStateBackend);
+ }
+
@Test(expected = IllegalArgumentException.class)
public void testRecursiveDelegation() throws Exception {
final StateBackend appBackend =
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java
index 3df0b50564c..18d91f828c0 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java
@@ -66,6 +66,7 @@ import java.util.NoSuchElementException;
import java.util.OptionalLong;
import java.util.stream.StreamSupport;
+import static org.apache.flink.runtime.state.StateBackendLoader.loadStateBackendFromKeyedStateHandles;
import static org.apache.flink.runtime.state.StateUtil.unexpectedStateHandleException;
/**
@@ -326,19 +327,25 @@ public class StreamTaskStateInitializerImpl implements StreamTaskStateInitialize
backendRestorer =
new BackendRestorerProcedure<>(
(stateHandles) ->
- stateBackend.createKeyedStateBackend(
- environment,
- environment.getJobID(),
- operatorIdentifierText,
- keySerializer,
- taskInfo.getMaxNumberOfParallelSubtasks(),
- keyGroupRange,
- environment.getTaskKvStateRegistry(),
- ttlTimeProvider,
- metricGroup,
- stateHandles,
- cancelStreamRegistryForRestore,
- managedMemoryFraction),
+ loadStateBackendFromKeyedStateHandles(
+ stateBackend,
+ environment
+ .getUserCodeClassLoader()
+ .asClassLoader(),
+ stateHandles)
+ .createKeyedStateBackend(
+ environment,
+ environment.getJobID(),
+ operatorIdentifierText,
+ keySerializer,
+ taskInfo.getMaxNumberOfParallelSubtasks(),
+ keyGroupRange,
+ environment.getTaskKvStateRegistry(),
+ ttlTimeProvider,
+ metricGroup,
+ stateHandles,
+ cancelStreamRegistryForRestore,
+ managedMemoryFraction),
backendCloseableRegistry,
logDescription);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationITCase.java
index fb4f8752562..b65a1226df1 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationITCase.java
@@ -17,6 +17,8 @@
package org.apache.flink.test.checkpointing;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.minicluster.MiniCluster;
import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.StateHandleID;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -46,6 +48,8 @@ public class ChangelogPeriodicMaterializationITCase
@Test
public void testNonMaterialization() throws Exception {
File checkpointFolder = TEMPORARY_FOLDER.newFolder();
+ SharedReference<JobID> jobID = sharedObjects.add(generateJobID());
+ SharedReference<MiniCluster> miniCluster = sharedObjects.add(cluster.getMiniCluster());
SharedReference<AtomicBoolean> hasMaterialization =
sharedObjects.add(new AtomicBoolean(true));
StreamExecutionEnvironment env =
@@ -64,13 +68,14 @@ public class ChangelogPeriodicMaterializationITCase
.get()
.compareAndSet(
true,
- !getAllStateHandleId(checkpointFolder)
+ !getAllStateHandleId(
+ jobID.get(), miniCluster.get())
.isEmpty());
throwArtificialFailure();
}
}
},
- generateJobID()));
+ jobID.get()));
Preconditions.checkState(!hasMaterialization.get().get());
}
@@ -78,6 +83,8 @@ public class ChangelogPeriodicMaterializationITCase
@Test
public void testMaterialization() throws Exception {
File checkpointFolder = TEMPORARY_FOLDER.newFolder();
+ SharedReference<JobID> jobID = sharedObjects.add(generateJobID());
+ SharedReference<MiniCluster> miniCluster = sharedObjects.add(cluster.getMiniCluster());
SharedReference<AtomicInteger> currentCheckpointNum =
sharedObjects.add(new AtomicInteger());
SharedReference<Set<StateHandleID>> currentMaterializationId =
@@ -101,7 +108,8 @@ public class ChangelogPeriodicMaterializationITCase
return true;
}
Set<StateHandleID> allMaterializationId =
- getAllStateHandleId(checkpointFolder);
+ getAllStateHandleId(
+ jobID.get(), miniCluster.get());
if (!allMaterializationId.isEmpty()) {
currentMaterializationId
.get()
@@ -126,7 +134,8 @@ public class ChangelogPeriodicMaterializationITCase
return true;
}
Set<StateHandleID> allMaterializationId =
- getAllStateHandleId(checkpointFolder);
+ getAllStateHandleId(
+ jobID.get(), miniCluster.get());
return allMaterializationId.isEmpty()
|| currentMaterializationId
.get()
@@ -136,12 +145,14 @@ public class ChangelogPeriodicMaterializationITCase
}
}
},
- generateJobID()));
+ jobID.get()));
}
@Test
public void testFailedMaterialization() throws Exception {
File checkpointFolder = TEMPORARY_FOLDER.newFolder();
+ SharedReference<JobID> jobID = sharedObjects.add(generateJobID());
+ SharedReference<MiniCluster> miniCluster = sharedObjects.add(cluster.getMiniCluster());
SharedReference<AtomicBoolean> hasFailed = sharedObjects.add(new AtomicBoolean());
SharedReference<Set<StateHandleID>> currentMaterializationId =
sharedObjects.add(ConcurrentHashMap.newKeySet());
@@ -175,7 +186,8 @@ public class ChangelogPeriodicMaterializationITCase
waitWhile(
() -> {
Set<StateHandleID> allMaterializationId =
- getAllStateHandleId(checkpointFolder);
+ getAllStateHandleId(
+ jobID.get(), miniCluster.get());
if (!allMaterializationId.isEmpty()) {
currentMaterializationId
.get()
@@ -188,7 +200,8 @@ public class ChangelogPeriodicMaterializationITCase
waitWhile(
() -> {
Set<StateHandleID> allMaterializationId =
- getAllStateHandleId(checkpointFolder);
+ getAllStateHandleId(
+ jobID.get(), miniCluster.get());
return allMaterializationId.isEmpty()
|| currentMaterializationId
.get()
@@ -197,6 +210,6 @@ public class ChangelogPeriodicMaterializationITCase
}
}
},
- generateJobID()));
+ jobID.get()));
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationRescaleITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationRescaleITCase.java
index 7cbc24650c6..9fbb657983c 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationRescaleITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationRescaleITCase.java
@@ -17,31 +17,18 @@
package org.apache.flink.test.checkpointing;
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.runtime.state.AbstractStateBackend;
-import org.apache.flink.runtime.state.StateHandleID;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.testutils.junit.SharedReference;
-import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.Preconditions;
import org.junit.Test;
-import java.io.File;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-
-import static org.apache.flink.test.util.TestUtils.getMostRecentCompletedCheckpoint;
-
/**
* This verifies that rescale works correctly for Changelog state backend with materialized state /
* non-materialized state.
*/
public class ChangelogPeriodicMaterializationRescaleITCase
- extends ChangelogPeriodicMaterializationTestBase {
+ extends ChangelogPeriodicMaterializationSwitchEnvTestBase {
public ChangelogPeriodicMaterializationRescaleITCase(
AbstractStateBackend delegatedStateBackend) {
@@ -50,90 +37,16 @@ public class ChangelogPeriodicMaterializationRescaleITCase
@Test
public void testRescaleOut() throws Exception {
- testRescale(NUM_SLOTS / 2, NUM_SLOTS);
+ testSwitchEnv(getEnv(NUM_SLOTS / 2), getEnv(NUM_SLOTS));
}
@Test
public void testRescaleIn() throws Exception {
- testRescale(NUM_SLOTS, NUM_SLOTS / 2);
- }
-
- private void testRescale(int firstParallelism, int secondParallelism) throws Exception {
- File firstCheckpointFolder = TEMPORARY_FOLDER.newFolder();
- JobID firstJobID = generateJobID();
- SharedReference<Set<StateHandleID>> currentMaterializationId =
- sharedObjects.add(ConcurrentHashMap.newKeySet());
-
- JobGraph firstJobGraph =
- buildJobGraph(
- getEnv(firstCheckpointFolder, firstParallelism),
- new ControlledSource() {
- @Override
- protected void beforeElement(SourceContext<Integer> ctx)
- throws Exception {
- if (currentIndex == TOTAL_ELEMENTS / 4) {
- waitWhile(
- () -> {
- if (completedCheckpointNum.get() <= 0) {
- return true;
- }
- Set<StateHandleID> allMaterializationId =
- getAllStateHandleId(firstCheckpointFolder);
- if (!allMaterializationId.isEmpty()) {
- currentMaterializationId
- .get()
- .addAll(allMaterializationId);
- return false;
- }
- return true;
- });
- } else if (currentIndex > TOTAL_ELEMENTS / 4) {
- throwArtificialFailure();
- }
- }
- },
- firstJobID);
-
- try {
- cluster.getMiniCluster().submitJob(firstJobGraph).get();
- cluster.getMiniCluster().requestJobResult(firstJobGraph.getJobID()).get();
- } catch (Exception ex) {
- Preconditions.checkState(
- ExceptionUtils.findThrowable(ex, ArtificialFailure.class).isPresent());
- }
-
- File secondCheckpointFolder = TEMPORARY_FOLDER.newFolder();
- JobID secondJobId = generateJobID();
- JobGraph jobGraph =
- buildJobGraph(
- getEnv(secondCheckpointFolder, secondParallelism),
- new ControlledSource() {
- @Override
- protected void beforeElement(SourceContext<Integer> ctx)
- throws Exception {
- if (currentIndex == TOTAL_ELEMENTS / 2) {
- waitWhile(
- () -> {
- Set<StateHandleID> allMaterializationId =
- getAllStateHandleId(secondCheckpointFolder);
- return allMaterializationId.isEmpty()
- || currentMaterializationId
- .get()
- .equals(allMaterializationId);
- });
- }
- }
- },
- secondJobId);
- File checkpoint = getMostRecentCompletedCheckpoint(firstCheckpointFolder);
- jobGraph.setSavepointRestoreSettings(
- SavepointRestoreSettings.forPath(checkpoint.getPath()));
- waitAndAssert(jobGraph);
+ testSwitchEnv(getEnv(NUM_SLOTS), getEnv(NUM_SLOTS / 2));
}
- private StreamExecutionEnvironment getEnv(File checkpointFolder, int parallelism) {
- StreamExecutionEnvironment env =
- getEnv(delegatedStateBackend, checkpointFolder, 50, 0, 20, 0);
+ private StreamExecutionEnvironment getEnv(int parallelism) {
+ StreamExecutionEnvironment env = getEnv(delegatedStateBackend, 50, 0, 20, 0);
env.getCheckpointConfig()
.setExternalizedCheckpointCleanup(
CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationSwitchEnvTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationSwitchEnvTestBase.java
new file mode 100644
index 00000000000..113ecf7594a
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationSwitchEnvTestBase.java
@@ -0,0 +1,133 @@
+/*
+ * 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.test.checkpointing;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.state.AbstractStateBackend;
+import org.apache.flink.runtime.state.StateHandleID;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.testutils.junit.SharedReference;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.Preconditions;
+
+import java.io.File;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.flink.test.util.TestUtils.getMostRecentCompletedCheckpoint;
+
+/** Base class for tests related to switching environment for Changelog state backend. */
+public abstract class ChangelogPeriodicMaterializationSwitchEnvTestBase
+ extends ChangelogPeriodicMaterializationTestBase {
+
+ public ChangelogPeriodicMaterializationSwitchEnvTestBase(
+ AbstractStateBackend delegatedStateBackend) {
+ super(delegatedStateBackend);
+ }
+
+ protected void testSwitchEnv(
+ StreamExecutionEnvironment firstEnv, StreamExecutionEnvironment secondEnv)
+ throws Exception {
+ File firstCheckpointFolder = TEMPORARY_FOLDER.newFolder();
+ SharedReference<MiniCluster> miniCluster = sharedObjects.add(cluster.getMiniCluster());
+ SharedReference<Set<StateHandleID>> currentMaterializationId =
+ sharedObjects.add(ConcurrentHashMap.newKeySet());
+ firstEnv.getCheckpointConfig().setCheckpointStorage(firstCheckpointFolder.toURI());
+ JobGraph firstJobGraph =
+ firstNormalJobGraph(firstEnv, miniCluster, currentMaterializationId);
+
+ try {
+ cluster.getMiniCluster().submitJob(firstJobGraph).get();
+ cluster.getMiniCluster().requestJobResult(firstJobGraph.getJobID()).get();
+ } catch (Exception ex) {
+ Preconditions.checkState(
+ ExceptionUtils.findThrowable(ex, ArtificialFailure.class).isPresent());
+ }
+
+ File secondCheckpointFolder = TEMPORARY_FOLDER.newFolder();
+ secondEnv.getCheckpointConfig().setCheckpointStorage(secondCheckpointFolder.toURI());
+ JobGraph jobGraph = nextNormalJobGraph(secondEnv, miniCluster, currentMaterializationId);
+ File checkpointFile = getMostRecentCompletedCheckpoint(firstCheckpointFolder);
+ jobGraph.setSavepointRestoreSettings(
+ SavepointRestoreSettings.forPath(checkpointFile.getPath()));
+ waitAndAssert(jobGraph);
+ }
+
+ private JobGraph firstNormalJobGraph(
+ StreamExecutionEnvironment env,
+ SharedReference<MiniCluster> miniCluster,
+ SharedReference<Set<StateHandleID>> currentMaterializationId) {
+ SharedReference<JobID> jobID = sharedObjects.add(generateJobID());
+ return buildJobGraph(
+ env,
+ new ControlledSource() {
+ @Override
+ protected void beforeElement(SourceContext<Integer> ctx) throws Exception {
+ if (currentIndex == TOTAL_ELEMENTS / 4) {
+ waitWhile(
+ () -> {
+ if (completedCheckpointNum.get() <= 0) {
+ return true;
+ }
+ Set<StateHandleID> allMaterializationId =
+ getAllStateHandleId(jobID.get(), miniCluster.get());
+ if (!allMaterializationId.isEmpty()) {
+ currentMaterializationId
+ .get()
+ .addAll(allMaterializationId);
+ return false;
+ }
+ return true;
+ });
+ } else if (currentIndex > TOTAL_ELEMENTS / 3) {
+ throwArtificialFailure();
+ }
+ }
+ },
+ jobID.get());
+ }
+
+ private JobGraph nextNormalJobGraph(
+ StreamExecutionEnvironment env,
+ SharedReference<MiniCluster> miniCluster,
+ SharedReference<Set<StateHandleID>> currentMaterializationId) {
+ SharedReference<JobID> jobID = sharedObjects.add(generateJobID());
+ return buildJobGraph(
+ env,
+ new ControlledSource() {
+ @Override
+ protected void beforeElement(SourceContext<Integer> ctx) throws Exception {
+ if (currentIndex == TOTAL_ELEMENTS / 2) {
+ waitWhile(
+ () -> {
+ Set<StateHandleID> allMaterializationId =
+ getAllStateHandleId(jobID.get(), miniCluster.get());
+ return allMaterializationId.isEmpty()
+ || currentMaterializationId
+ .get()
+ .equals(allMaterializationId);
+ });
+ }
+ }
+ },
+ jobID.get());
+ }
+}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationSwitchStateBackendITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationSwitchStateBackendITCase.java
new file mode 100644
index 00000000000..c2e06bf22ec
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationSwitchStateBackendITCase.java
@@ -0,0 +1,66 @@
+/*
+ * 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.test.checkpointing;
+
+import org.apache.flink.runtime.state.AbstractStateBackend;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import org.junit.Test;
+
+/**
+ * This verifies that switching state backend works correctly for Changelog state backend with
+ * materialized state / non-materialized state.
+ */
+public class ChangelogPeriodicMaterializationSwitchStateBackendITCase
+ extends ChangelogPeriodicMaterializationSwitchEnvTestBase {
+
+ public ChangelogPeriodicMaterializationSwitchStateBackendITCase(
+ AbstractStateBackend delegatedStateBackend) {
+ super(delegatedStateBackend);
+ }
+
+ @Test
+ public void testSwitchFromEnablingToDisabling() throws Exception {
+ testSwitchEnv(getEnv(true), getEnv(false));
+ }
+
+ @Test
+ public void testSwitchFromEnablingToDisablingWithRescalingOut() throws Exception {
+ testSwitchEnv(getEnv(true, NUM_SLOTS / 2), getEnv(false, NUM_SLOTS));
+ }
+
+ @Test
+ public void testSwitchFromEnablingToDisablingWithRescalingIn() throws Exception {
+ testSwitchEnv(getEnv(true, NUM_SLOTS), getEnv(false, NUM_SLOTS / 2));
+ }
+
+ private StreamExecutionEnvironment getEnv(boolean enableChangelog) {
+ return getEnv(enableChangelog, NUM_SLOTS);
+ }
+
+ private StreamExecutionEnvironment getEnv(boolean enableChangelog, int parallelism) {
+ StreamExecutionEnvironment env = getEnv(delegatedStateBackend, 100, 0, 500, 0);
+ env.enableChangelogStateBackend(enableChangelog);
+ env.setParallelism(parallelism);
+ env.getCheckpointConfig()
+ .setExternalizedCheckpointCleanup(
+ CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
+ return env;
+ }
+}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationTestBase.java
index 4c5e7603efc..e05f966d4f4 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationTestBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogPeriodicMaterializationTestBase.java
@@ -19,7 +19,7 @@ package org.apache.flink.test.checkpointing;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobSubmissionResult;
-import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.CheckpointListener;
import org.apache.flink.api.common.state.ListState;
@@ -31,7 +31,6 @@ import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.changelog.fs.FsStateChangelogStorageFactory;
-import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.StateChangelogOptions;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
@@ -43,6 +42,8 @@ import org.apache.flink.runtime.clusterframework.ApplicationStatus;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.runtime.messages.FlinkJobNotFoundException;
+import org.apache.flink.runtime.minicluster.MiniCluster;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.AbstractStateBackend;
@@ -67,12 +68,20 @@ import org.apache.flink.runtime.state.heap.HeapPriorityQueueElement;
import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
+import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.test.util.MiniClusterWithClientResource;
import org.apache.flink.testutils.junit.SharedObjects;
import org.apache.flink.util.AbstractID;
+import org.apache.flink.util.Collector;
import org.apache.flink.util.TestLogger;
import org.apache.flink.util.function.FunctionWithException;
@@ -87,6 +96,7 @@ import org.junit.runners.Parameterized;
import javax.annotation.Nonnull;
import java.io.File;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.Serializable;
import java.time.Duration;
@@ -99,14 +109,15 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import static org.apache.flink.runtime.testutils.CommonTestUtils.getLatestCompletedCheckpointPath;
import static org.apache.flink.shaded.guava30.com.google.common.collect.Iterables.get;
-import static org.apache.flink.test.util.TestUtils.getMostRecentCompletedCheckpointMaybe;
import static org.apache.flink.test.util.TestUtils.loadCheckpointMetadata;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
@@ -128,7 +139,7 @@ public abstract class ChangelogPeriodicMaterializationTestBase extends TestLogge
@Rule public final SharedObjects sharedObjects = SharedObjects.create();
- @Parameterized.Parameters(name = "delegated state backend type ={0}")
+ @Parameterized.Parameters(name = "delegated state backend type = {0}")
public static Collection<AbstractStateBackend> parameter() {
return Arrays.asList(
new HashMapStateBackend(),
@@ -162,16 +173,12 @@ public abstract class ChangelogPeriodicMaterializationTestBase extends TestLogge
protected StreamExecutionEnvironment getEnv(
StateBackend stateBackend,
- File checkpointFile,
long checkpointInterval,
int restartAttempts,
long materializationInterval,
int materializationMaxFailure) {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.enableCheckpointing(checkpointInterval)
- .enableChangelogStateBackend(true)
- .getCheckpointConfig()
- .setCheckpointStorage(checkpointFile.toURI());
+ env.enableCheckpointing(checkpointInterval).enableChangelogStateBackend(true);
env.getCheckpointConfig().enableUnalignedCheckpoints(false);
env.setStateBackend(stateBackend)
.setRestartStrategy(RestartStrategies.fixedDelayRestart(restartAttempts, 0));
@@ -186,13 +193,43 @@ public abstract class ChangelogPeriodicMaterializationTestBase extends TestLogge
return env;
}
+ protected StreamExecutionEnvironment getEnv(
+ StateBackend stateBackend,
+ File checkpointFile,
+ long checkpointInterval,
+ int restartAttempts,
+ long materializationInterval,
+ int materializationMaxFailure) {
+ StreamExecutionEnvironment env =
+ getEnv(
+ stateBackend,
+ checkpointInterval,
+ restartAttempts,
+ materializationInterval,
+ materializationMaxFailure);
+ env.getCheckpointConfig().setCheckpointStorage(checkpointFile.toURI());
+ return env;
+ }
+
protected JobGraph buildJobGraph(
StreamExecutionEnvironment env, ControlledSource controlledSource, JobID jobId) {
- env.addSource(controlledSource)
- .keyBy(element -> element)
- .map(new CountMapper())
- .addSink(new CollectionSink())
- .setParallelism(1);
+ KeyedStream<Integer, Integer> keyedStream =
+ env.addSource(controlledSource)
+ .assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps())
+ .keyBy(element -> element);
+ keyedStream.process(new CountFunction()).addSink(new CollectionSink()).setParallelism(1);
+ keyedStream
+ .window(TumblingProcessingTimeWindows.of(Time.milliseconds(10)))
+ .process(
+ new ProcessWindowFunction<Integer, Integer, Integer, TimeWindow>() {
+ @Override
+ public void process(
+ Integer integer,
+ Context context,
+ Iterable<Integer> elements,
+ Collector<Integer> out) {}
+ })
+ .addSink(new DiscardingSink<>());
return env.getStreamGraph().getJobGraph(jobId);
}
@@ -207,15 +244,23 @@ public abstract class ChangelogPeriodicMaterializationTestBase extends TestLogge
return JobID.fromByteArray(randomBytes);
}
- protected static Set<StateHandleID> getAllStateHandleId(File checkpointFile)
- throws IOException {
- Optional<File> mostRecentCompletedCheckpoint =
- getMostRecentCompletedCheckpointMaybe(checkpointFile);
- if (!mostRecentCompletedCheckpoint.isPresent()) {
+ protected static Set<StateHandleID> getAllStateHandleId(JobID jobID, MiniCluster miniCluster)
+ throws IOException, FlinkJobNotFoundException, ExecutionException,
+ InterruptedException {
+ Optional<String> mostRecentCompletedCheckpointPath =
+ getLatestCompletedCheckpointPath(jobID, miniCluster);
+ if (!mostRecentCompletedCheckpointPath.isPresent()) {
+ return Collections.emptySet();
+ }
+
+ CheckpointMetadata checkpointMetadata;
+ try {
+ checkpointMetadata = loadCheckpointMetadata(mostRecentCompletedCheckpointPath.get());
+ } catch (FileNotFoundException fileNotFoundException) {
+ // return empty result when the metadata file do not exist due to subsumed checkpoint.
return Collections.emptySet();
}
- CheckpointMetadata checkpointMetadata =
- loadCheckpointMetadata(mostRecentCompletedCheckpoint.get().toString());
+
Set<StateHandleID> materializationIds =
checkpointMetadata.getOperatorStates().stream()
.flatMap(operatorState -> operatorState.getStates().stream())
@@ -224,8 +269,10 @@ public abstract class ChangelogPeriodicMaterializationTestBase extends TestLogge
operatorSubtaskState.getManagedKeyedState().stream())
.flatMap(
keyedStateHandle ->
- ((ChangelogStateBackendHandle) keyedStateHandle)
- .getMaterializedStateHandles().stream())
+ keyedStateHandle instanceof ChangelogStateBackendHandle
+ ? ((ChangelogStateBackendHandle) keyedStateHandle)
+ .getMaterializedStateHandles().stream()
+ : Stream.of(keyedStateHandle))
.map(KeyedStateHandle::getStateHandleId)
.collect(Collectors.toSet());
if (!materializationIds.isEmpty()) {
@@ -247,7 +294,6 @@ public abstract class ChangelogPeriodicMaterializationTestBase extends TestLogge
private Configuration configure() throws IOException {
Configuration configuration = new Configuration();
- configuration.setInteger(CheckpointingOptions.MAX_RETAINED_CHECKPOINTS, 200);
FsStateChangelogStorageFactory.configure(configuration, TEMPORARY_FOLDER.newFolder());
return configuration;
}
@@ -375,8 +421,9 @@ public abstract class ChangelogPeriodicMaterializationTestBase extends TestLogge
}
}
- /** A Map function with value state mocks word count. */
- protected static class CountMapper extends RichMapFunction<Integer, Tuple2<Integer, Integer>> {
+ /** A Process function with value state mocks word count. */
+ protected static class CountFunction
+ extends KeyedProcessFunction<Integer, Integer, Tuple2<Integer, Integer>> {
private static final long serialVersionUID = 1L;
@@ -390,11 +437,13 @@ public abstract class ChangelogPeriodicMaterializationTestBase extends TestLogge
}
@Override
- public Tuple2<Integer, Integer> map(Integer value) throws Exception {
+ public void processElement(
+ Integer value, Context ctx, Collector<Tuple2<Integer, Integer>> out)
+ throws Exception {
Integer count = countState.value();
Integer currentCount = count == null ? 1 : count + 1;
countState.update(currentCount);
- return Tuple2.of(value, currentCount);
+ out.collect(Tuple2.of(value, currentCount));
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogCompatibilityITCase.java b/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogCompatibilityITCase.java
index aafa0966878..5176cd523e2 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogCompatibilityITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogCompatibilityITCase.java
@@ -66,11 +66,16 @@ public class ChangelogCompatibilityITCase {
@Parameterized.Parameters(name = "{0}")
public static List<TestCase> parameters() {
return Arrays.asList(
- // disable changelog - allow restore only from CANONICAL_SAVEPOINT
+ // disable changelog - allow restore from CANONICAL_SAVEPOINT
TestCase.startWithChangelog(true)
.restoreWithChangelog(false)
.from(RestoreSource.CANONICAL_SAVEPOINT)
.allowRestore(true),
+ // disable changelog - allow restore from CHECKPOINT
+ TestCase.startWithChangelog(true)
+ .restoreWithChangelog(false)
+ .from(RestoreSource.CHECKPOINT)
+ .allowRestore(true),
// enable changelog - allow restore only from CANONICAL_SAVEPOINT
TestCase.startWithChangelog(false)
.restoreWithChangelog(true)
[flink] 02/02: [FLINK-23252][changelog] Introduce CheckpointBoundKeyedStateHandle#rebound
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit ece745c39eaee7caaa1a9d077f079543cb088f67
Author: Hangxiang Yu <ma...@gmail.com>
AuthorDate: Fri Apr 22 15:13:38 2022 +0800
[FLINK-23252][changelog] Introduce CheckpointBoundKeyedStateHandle#rebound
---
.../metadata/MetadataV2V3SerializerBase.java | 16 ++++++--
.../state/CheckpointBoundKeyedStateHandle.java | 10 ++++-
.../state/IncrementalLocalKeyedStateHandle.java | 11 ++++++
.../state/IncrementalRemoteKeyedStateHandle.java | 13 +++++++
.../changelog/ChangelogStateBackendHandle.java | 44 +++++++++++++++++++++-
.../flink/runtime/state/ChangelogTestUtils.java | 1 +
.../runtime/state/SharedStateRegistryTest.java | 3 ++
.../changelog/ChangelogStateBackendHandleTest.java | 18 ++++++++-
.../changelog/AbstractChangelogStateBackend.java | 1 +
.../changelog/ChangelogKeyedStateBackend.java | 9 ++++-
.../DeactivatedChangelogStateBackend.java | 20 ++++++++++
11 files changed, 137 insertions(+), 9 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java
index 8bb9db305d1..37f22ee95ad 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java
@@ -119,6 +119,8 @@ public abstract class MetadataV2V3SerializerBase {
private static final byte KEY_GROUPS_HANDLE_V2 = 12;
// CHANGELOG_FILE_INCREMENT_HANDLE_V2 is introduced to add new field of storageIdentifier.
private static final byte CHANGELOG_FILE_INCREMENT_HANDLE_V2 = 13;
+ // CHANGELOG_HANDLE_V2 is introduced to add new field of checkpointId.
+ private static final byte CHANGELOG_HANDLE_V2 = 14;
// ------------------------------------------------------------------------
// (De)serialization entry points
@@ -346,7 +348,7 @@ public abstract class MetadataV2V3SerializerBase {
} else if (stateHandle instanceof ChangelogStateBackendHandle) {
ChangelogStateBackendHandle handle = (ChangelogStateBackendHandle) stateHandle;
- dos.writeByte(CHANGELOG_HANDLE);
+ dos.writeByte(CHANGELOG_HANDLE_V2);
dos.writeInt(handle.getKeyGroupRange().getStartKeyGroup());
dos.writeInt(handle.getKeyGroupRange().getNumberOfKeyGroups());
@@ -363,6 +365,7 @@ public abstract class MetadataV2V3SerializerBase {
}
dos.writeLong(handle.getMaterializationID());
+ dos.writeLong(handle.getCheckpointId());
writeStateHandleId(handle, dos);
} else if (stateHandle instanceof InMemoryChangelogStateHandle) {
@@ -441,7 +444,7 @@ public abstract class MetadataV2V3SerializerBase {
} else if (INCREMENTAL_KEY_GROUPS_HANDLE == type
|| INCREMENTAL_KEY_GROUPS_HANDLE_V2 == type) {
return deserializeIncrementalStateHandle(dis, context, type);
- } else if (CHANGELOG_HANDLE == type) {
+ } else if (CHANGELOG_HANDLE == type || CHANGELOG_HANDLE_V2 == type) {
int startKeyGroup = dis.readInt();
int numKeyGroups = dis.readInt();
@@ -467,9 +470,16 @@ public abstract class MetadataV2V3SerializerBase {
}
long materializationID = dis.readLong();
+ long checkpointId = CHANGELOG_HANDLE_V2 == type ? dis.readLong() : materializationID;
StateHandleID stateHandleId = new StateHandleID(dis.readUTF());
return ChangelogStateBackendHandleImpl.restore(
- base, delta, keyGroupRange, materializationID, checkpointedSize, stateHandleId);
+ base,
+ delta,
+ keyGroupRange,
+ checkpointId,
+ materializationID,
+ checkpointedSize,
+ stateHandleId);
} else if (CHANGELOG_BYTE_INCREMENT_HANDLE == type) {
int start = dis.readInt();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointBoundKeyedStateHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointBoundKeyedStateHandle.java
index 6688f0222c5..7b4762d98be 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointBoundKeyedStateHandle.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointBoundKeyedStateHandle.java
@@ -17,8 +17,16 @@
package org.apache.flink.runtime.state;
/** {@link KeyedStateHandle} that is bound to a specific checkpoint. */
-public interface CheckpointBoundKeyedStateHandle {
+public interface CheckpointBoundKeyedStateHandle extends KeyedStateHandle {
/** Returns the ID of the checkpoint for which the handle was created or used. */
long getCheckpointId();
+
+ /**
+ * Returns a new {@link CheckpointBoundKeyedStateHandle} Rebounding checkpoint id to a specific
+ * checkpoint id.
+ *
+ * @param checkpointId rebounded checkpoint id.
+ */
+ CheckpointBoundKeyedStateHandle rebound(long checkpointId);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/IncrementalLocalKeyedStateHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/IncrementalLocalKeyedStateHandle.java
index 16f18a22167..85405dc649e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/IncrementalLocalKeyedStateHandle.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/IncrementalLocalKeyedStateHandle.java
@@ -74,6 +74,17 @@ public class IncrementalLocalKeyedStateHandle extends DirectoryKeyedStateHandle
return checkpointId;
}
+ @Override
+ public CheckpointBoundKeyedStateHandle rebound(long checkpointId) {
+ return new IncrementalLocalKeyedStateHandle(
+ backendIdentifier,
+ checkpointId,
+ getDirectoryStateHandle(),
+ getKeyGroupRange(),
+ getMetaDataState(),
+ getSharedStateHandles());
+ }
+
@Override
@Nonnull
public UUID getBackendIdentifier() {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/IncrementalRemoteKeyedStateHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/IncrementalRemoteKeyedStateHandle.java
index 602cf6d65c8..28813e7bd97 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/IncrementalRemoteKeyedStateHandle.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/IncrementalRemoteKeyedStateHandle.java
@@ -190,6 +190,19 @@ public class IncrementalRemoteKeyedStateHandle implements IncrementalKeyedStateH
return checkpointId;
}
+ @Override
+ public CheckpointBoundKeyedStateHandle rebound(long checkpointId) {
+ return new IncrementalRemoteKeyedStateHandle(
+ backendIdentifier,
+ keyGroupRange,
+ checkpointId,
+ sharedState,
+ privateState,
+ metaStateHandle,
+ persistedSizeOfThisCheckpoint,
+ stateHandleId);
+ }
+
public Map<StateHandleID, StreamStateHandle> getSharedState() {
return sharedState;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/ChangelogStateBackendHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/ChangelogStateBackendHandle.java
index c9116691bdf..b1f5b0d61ec 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/ChangelogStateBackendHandle.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/ChangelogStateBackendHandle.java
@@ -19,6 +19,7 @@ package org.apache.flink.runtime.state.changelog;
import org.apache.flink.annotation.Internal;
import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.runtime.state.CheckpointBoundKeyedStateHandle;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.SharedStateRegistry;
@@ -45,13 +46,17 @@ import static org.apache.flink.util.Preconditions.checkArgument;
* can be no state or multiple states (e.g. after rescaling).
*/
@Internal
-public interface ChangelogStateBackendHandle extends KeyedStateHandle {
+public interface ChangelogStateBackendHandle
+ extends KeyedStateHandle, CheckpointBoundKeyedStateHandle {
List<KeyedStateHandle> getMaterializedStateHandles();
List<ChangelogStateHandle> getNonMaterializedStateHandles();
long getMaterializationID();
+ @Override
+ ChangelogStateBackendHandle rebound(long checkpointId);
+
class ChangelogStateBackendHandleImpl implements ChangelogStateBackendHandle {
private static final long serialVersionUID = 1L;
@@ -60,6 +65,7 @@ public interface ChangelogStateBackendHandle extends KeyedStateHandle {
private final KeyGroupRange keyGroupRange;
private final long materializationID;
+ private final long checkpointId;
private final long persistedSizeOfThisCheckpoint;
private final StateHandleID stateHandleID;
@@ -67,12 +73,14 @@ public interface ChangelogStateBackendHandle extends KeyedStateHandle {
List<KeyedStateHandle> materialized,
List<ChangelogStateHandle> nonMaterialized,
KeyGroupRange keyGroupRange,
+ long checkpointId,
long materializationID,
long persistedSizeOfThisCheckpoint) {
this(
materialized,
nonMaterialized,
keyGroupRange,
+ checkpointId,
materializationID,
persistedSizeOfThisCheckpoint,
StateHandleID.randomStateHandleId());
@@ -82,6 +90,7 @@ public interface ChangelogStateBackendHandle extends KeyedStateHandle {
List<KeyedStateHandle> materialized,
List<ChangelogStateHandle> nonMaterialized,
KeyGroupRange keyGroupRange,
+ long checkpointId,
long materializationID,
long persistedSizeOfThisCheckpoint,
StateHandleID stateHandleId) {
@@ -90,6 +99,7 @@ public interface ChangelogStateBackendHandle extends KeyedStateHandle {
this.keyGroupRange = keyGroupRange;
this.persistedSizeOfThisCheckpoint = persistedSizeOfThisCheckpoint;
checkArgument(keyGroupRange.getNumberOfKeyGroups() > 0);
+ this.checkpointId = checkpointId;
this.materializationID = materializationID;
this.stateHandleID = stateHandleId;
}
@@ -98,6 +108,7 @@ public interface ChangelogStateBackendHandle extends KeyedStateHandle {
List<KeyedStateHandle> materialized,
List<ChangelogStateHandle> nonMaterialized,
KeyGroupRange keyGroupRange,
+ long checkpointId,
long materializationID,
long persistedSizeOfThisCheckpoint,
StateHandleID stateHandleId) {
@@ -105,6 +116,7 @@ public interface ChangelogStateBackendHandle extends KeyedStateHandle {
materialized,
nonMaterialized,
keyGroupRange,
+ checkpointId,
materializationID,
persistedSizeOfThisCheckpoint,
stateHandleId);
@@ -163,6 +175,7 @@ public interface ChangelogStateBackendHandle extends KeyedStateHandle {
basePart,
deltaPart,
intersection,
+ checkpointId,
materializationID,
persistedSizeOfThisCheckpoint);
}
@@ -205,6 +218,35 @@ public interface ChangelogStateBackendHandle extends KeyedStateHandle {
keyGroupRange, materialized.size(), nonMaterialized.size());
}
+ @Override
+ public long getCheckpointId() {
+ return checkpointId;
+ }
+
+ @Override
+ public ChangelogStateBackendHandleImpl rebound(long checkpointId) {
+ List<KeyedStateHandle> reboundedMaterialized =
+ materialized.stream()
+ .map(
+ keyedStateHandle ->
+ keyedStateHandle
+ instanceof
+ CheckpointBoundKeyedStateHandle
+ ? ((CheckpointBoundKeyedStateHandle)
+ keyedStateHandle)
+ .rebound(checkpointId)
+ : keyedStateHandle)
+ .collect(Collectors.toList());
+ return new ChangelogStateBackendHandleImpl(
+ reboundedMaterialized,
+ nonMaterialized,
+ keyGroupRange,
+ checkpointId,
+ materializationID,
+ persistedSizeOfThisCheckpoint,
+ stateHandleID);
+ }
+
/**
* This wrapper class is introduced as current {@link SharedStateRegistry} only accept
* StreamStateHandle to register, remove it once FLINK-25862 is resolved.
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChangelogTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChangelogTestUtils.java
index 1897cf92d18..62cd99a83ca 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChangelogTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChangelogTestUtils.java
@@ -48,6 +48,7 @@ public class ChangelogTestUtils {
Collections.emptyList(),
new KeyGroupRange(0, 1),
1L,
+ 1L,
0L);
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/SharedStateRegistryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/SharedStateRegistryTest.java
index e2da9f673bd..9d8d9e4a20f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/SharedStateRegistryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/SharedStateRegistryTest.java
@@ -122,6 +122,7 @@ public class SharedStateRegistryTest {
Collections.singletonList(materializedState1),
Collections.singletonList(nonMaterializedState1),
materializedStateBase1.getKeyGroupRange(),
+ checkpointId1,
materializationId,
nonMaterializedState1.getStateSize());
changelogStateBackendHandle1.registerSharedStates(sharedStateRegistry, checkpointId1);
@@ -136,6 +137,7 @@ public class SharedStateRegistryTest {
Collections.singletonList(materializedState2),
Collections.singletonList(nonMaterializedState2),
materializedStateBase1.getKeyGroupRange(),
+ checkpointId2,
materializationId,
nonMaterializedState2.getStateSize());
changelogStateBackendHandle2.registerSharedStates(sharedStateRegistry, checkpointId2);
@@ -160,6 +162,7 @@ public class SharedStateRegistryTest {
Collections.singletonList(materializedState3),
Collections.singletonList(nonMaterializedState2),
materializedState3.getKeyGroupRange(),
+ checkpointId3,
materializationId2,
0L);
changelogStateBackendHandle3.registerSharedStates(sharedStateRegistry, checkpointId3);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/changelog/ChangelogStateBackendHandleTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/changelog/ChangelogStateBackendHandleTest.java
index b2f9c54988a..de5fec4d9cc 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/changelog/ChangelogStateBackendHandleTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/changelog/ChangelogStateBackendHandleTest.java
@@ -30,33 +30,47 @@ public class ChangelogStateBackendHandleTest {
@Test
public void testPublicConstructor() {
+ long checkpointId = 2L;
long materializationID = 1L;
long size = 2L;
validateHandle(
+ checkpointId,
materializationID,
size,
new ChangelogStateBackendHandleImpl(
- emptyList(), emptyList(), KeyGroupRange.of(1, 2), materializationID, size));
+ emptyList(),
+ emptyList(),
+ KeyGroupRange.of(1, 2),
+ checkpointId,
+ materializationID,
+ size));
}
@Test
public void testRestore() {
+ long checkpointId = 2L;
long materializationID = 1L;
long size = 2L;
validateHandle(
+ checkpointId,
materializationID,
size,
ChangelogStateBackendHandleImpl.restore(
emptyList(),
emptyList(),
KeyGroupRange.of(1, 2),
+ checkpointId,
materializationID,
size,
randomStateHandleId()));
}
private void validateHandle(
- long materializationID, long size, ChangelogStateBackendHandleImpl handle) {
+ long checkpointId,
+ long materializationID,
+ long size,
+ ChangelogStateBackendHandleImpl handle) {
+ assertEquals(checkpointId, handle.getCheckpointId());
assertEquals(materializationID, handle.getMaterializationID());
assertEquals(size, handle.getCheckpointedSize());
}
diff --git a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/AbstractChangelogStateBackend.java b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/AbstractChangelogStateBackend.java
index 826ad4828a3..d5c04179343 100644
--- a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/AbstractChangelogStateBackend.java
+++ b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/AbstractChangelogStateBackend.java
@@ -214,6 +214,7 @@ public abstract class AbstractChangelogStateBackend
emptyList(),
keyedStateHandle.getKeyGroupRange(),
getMaterializationID(keyedStateHandle),
+ getMaterializationID(keyedStateHandle),
0L);
} else {
throw new IllegalStateException(
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 eb6204d2ef5..35849546c91 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
@@ -385,11 +385,15 @@ public class ChangelogKeyedStateBackend<K>
.thenApply(
delta ->
buildSnapshotResult(
- delta, changelogStateBackendStateCopy)));
+ checkpointId,
+ delta,
+ changelogStateBackendStateCopy)));
}
private SnapshotResult<KeyedStateHandle> buildSnapshotResult(
- ChangelogStateHandle delta, ChangelogSnapshotState changelogStateBackendStateCopy) {
+ long checkpointId,
+ ChangelogStateHandle delta,
+ ChangelogSnapshotState changelogStateBackendStateCopy) {
// collections don't change once started and handles are immutable
List<ChangelogStateHandle> prevDeltaCopy =
@@ -409,6 +413,7 @@ public class ChangelogKeyedStateBackend<K>
changelogStateBackendStateCopy.getMaterializedSnapshot(),
prevDeltaCopy,
getKeyGroupRange(),
+ checkpointId,
changelogStateBackendStateCopy.materializationID,
persistedSizeOfThisCheckpoint));
}
diff --git a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/DeactivatedChangelogStateBackend.java b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/DeactivatedChangelogStateBackend.java
index e0b8815dac3..48af8890615 100644
--- a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/DeactivatedChangelogStateBackend.java
+++ b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/DeactivatedChangelogStateBackend.java
@@ -30,6 +30,7 @@ import org.apache.flink.state.changelog.restore.ChangelogBackendRestoreOperation
import org.apache.flink.state.changelog.restore.ChangelogMigrationRestoreTarget;
import java.util.Collection;
+import java.util.stream.Collectors;
/**
* This state backend use delegatedStateBackend and State changes to restore to the
@@ -53,6 +54,15 @@ public class DeactivatedChangelogStateBackend extends AbstractChangelogStateBack
Collection<ChangelogStateBackendHandle> stateBackendHandles,
BaseBackendBuilder<K> baseBackendBuilder)
throws Exception {
+ // ChangelogKeyedStateBackend will use materialization id as the checkpoint id of delegated
+ // state backend.
+ // While switching Changelog from enabled to disabled,
+ // the materialization id will be passed into the delegated keyed state backend whose
+ // KeyedStateHandle implements CheckpointBoundKeyedStateHandle.
+ // It will cause unforeseen exception because we couldn't know what the delegated keyed
+ // state backend will do with checkpoint id.
+ // So we need to rebound the checkpoint id to the real checkpoint id here.
+ stateBackendHandles = reboundCheckpoint(stateBackendHandles);
ChangelogStateFactory changelogStateFactory = new ChangelogStateFactory();
return ChangelogBackendRestoreOperation.restore(
env.getUserCodeClassLoader().asClassLoader(),
@@ -61,4 +71,14 @@ public class DeactivatedChangelogStateBackend extends AbstractChangelogStateBack
(baseBackend, baseState) ->
new ChangelogMigrationRestoreTarget<>(baseBackend, changelogStateFactory));
}
+
+ private Collection<ChangelogStateBackendHandle> reboundCheckpoint(
+ Collection<ChangelogStateBackendHandle> stateBackendHandles) {
+ return stateBackendHandles.stream()
+ .map(
+ changelogStateBackendHandle ->
+ changelogStateBackendHandle.rebound(
+ changelogStateBackendHandle.getCheckpointId()))
+ .collect(Collectors.toList());
+ }
}