You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2022/02/17 15:04:35 UTC
[flink] 01/02: [hotfix][state] Explicitly validate native savepoints support
This is an automated email from the ASF dual-hosted git repository.
roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 77811884f693d0899b0bf25fb35a6542744b693e
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Tue Feb 15 18:12:48 2022 +0100
[hotfix][state] Explicitly validate native savepoints support
---
.../java/org/apache/flink/streaming/tests/StubStateBackend.java | 6 ++++++
.../main/java/org/apache/flink/runtime/state/StateBackend.java | 5 +++++
.../org/apache/flink/runtime/state/filesystem/FsStateBackend.java | 6 ++++++
.../apache/flink/runtime/state/hashmap/HashMapStateBackend.java | 6 ++++++
.../org/apache/flink/runtime/state/memory/MemoryStateBackend.java | 6 ++++++
.../contrib/streaming/state/EmbeddedRocksDBStateBackend.java | 6 ++++++
.../apache/flink/contrib/streaming/state/RocksDBStateBackend.java | 6 ++++++
.../java/org/apache/flink/streaming/runtime/tasks/StreamTask.java | 8 ++++++++
8 files changed, 49 insertions(+)
diff --git a/flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/StubStateBackend.java b/flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/StubStateBackend.java
index d78208e..5f10794 100644
--- a/flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/StubStateBackend.java
+++ b/flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/StubStateBackend.java
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.tests;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+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;
@@ -99,4 +100,9 @@ final class StubStateBackend implements StateBackend {
public boolean supportsNoClaimRestoreMode() {
return backend.supportsNoClaimRestoreMode();
}
+
+ @Override
+ public boolean supportsSavepointFormat(SavepointFormatType formatType) {
+ return true;
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackend.java
index a551cfa..4f904be 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackend.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.state;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+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.checkpoint.CheckpointOptions;
@@ -188,4 +189,8 @@ public interface StateBackend extends java.io.Serializable {
default boolean supportsNoClaimRestoreMode() {
return false;
}
+
+ default boolean supportsSavepointFormat(SavepointFormatType formatType) {
+ return formatType == SavepointFormatType.CANONICAL;
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
index cacc1c2..2f7910c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
@@ -502,6 +503,11 @@ public class FsStateBackend extends AbstractFileStateBackend implements Configur
return true;
}
+ @Override
+ public boolean supportsSavepointFormat(SavepointFormatType formatType) {
+ return true;
+ }
+
// ------------------------------------------------------------------------
// Reconfiguration
// ------------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/hashmap/HashMapStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/hashmap/HashMapStateBackend.java
index 8e44be8..6f00622 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/hashmap/HashMapStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/hashmap/HashMapStateBackend.java
@@ -23,6 +23,7 @@ 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;
@@ -98,6 +99,11 @@ public class HashMapStateBackend extends AbstractStateBackend implements Configu
}
@Override
+ public boolean supportsSavepointFormat(SavepointFormatType formatType) {
+ return true;
+ }
+
+ @Override
public <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(
Environment env,
JobID jobID,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java
index c98995b..cc1a447 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.fs.Path;
import org.apache.flink.metrics.MetricGroup;
@@ -292,6 +293,11 @@ public class MemoryStateBackend extends AbstractFileStateBackend
return true;
}
+ @Override
+ public boolean supportsSavepointFormat(SavepointFormatType formatType) {
+ return true;
+ }
+
// ------------------------------------------------------------------------
// Reconfiguration
// ------------------------------------------------------------------------
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackend.java
index cb7e8d9..bd906e2 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackend.java
@@ -30,6 +30,7 @@ import org.apache.flink.configuration.DescribedEnum;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.configuration.description.InlineElement;
+import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.fs.Path;
import org.apache.flink.metrics.MetricGroup;
@@ -309,6 +310,11 @@ public class EmbeddedRocksDBStateBackend extends AbstractManagedMemoryStateBacke
return true;
}
+ @Override
+ public boolean supportsSavepointFormat(SavepointFormatType formatType) {
+ return true;
+ }
+
private void lazyInitializeForJob(
Environment env, @SuppressWarnings("unused") String operatorIdentifier)
throws IOException {
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
index a3d5429..20be1e0 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
@@ -21,6 +21,7 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.fs.Path;
import org.apache.flink.metrics.MetricGroup;
@@ -288,6 +289,11 @@ public class RocksDBStateBackend extends AbstractManagedMemoryStateBackend
return true;
}
+ @Override
+ public boolean supportsSavepointFormat(SavepointFormatType formatType) {
+ return true;
+ }
+
// ------------------------------------------------------------------------
// Checkpoint initialization and persistent storage
// ------------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index 5fa34b0..ec0e334 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -1295,6 +1295,14 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
RestoreMode.NO_CLAIM,
RestoreMode.CLAIM,
RestoreMode.LEGACY));
+ } else if (checkpointOptions.getCheckpointType().isSavepoint()) {
+ SavepointType savepointType = (SavepointType) checkpointOptions.getCheckpointType();
+ if (!stateBackend.supportsSavepointFormat(savepointType.getFormatType())) {
+ throw new IllegalStateException(
+ String.format(
+ "Configured state backend (%s) does not support %s savepoints",
+ stateBackend, savepointType.getFormatType()));
+ }
}
}