You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by le...@apache.org on 2024/11/19 02:03:57 UTC
(flink-cdc) 03/09: [FLINK-36105][flink-cdc-cli] Fix unable to restore from checkpoint with Flink 1.20
This is an automated email from the ASF dual-hosted git repository.
leonard pushed a commit to branch release-3.2
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
commit a92dba65306417244abe7a2c7d0214ec650b4754
Author: yuxiqian <34...@users.noreply.github.com>
AuthorDate: Wed Aug 21 10:00:28 2024 +0800
[FLINK-36105][flink-cdc-cli] Fix unable to restore from checkpoint with Flink 1.20
This closes #3701.
---
.../java/org/apache/flink/cdc/cli/CliFrontend.java | 35 ++++++++++++++++++----
.../flink/cdc/cli/utils/ConfigurationUtils.java | 12 ++++++++
2 files changed, 42 insertions(+), 5 deletions(-)
diff --git a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliFrontend.java b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliFrontend.java
index ac746e224..27cb95ec9 100644
--- a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliFrontend.java
+++ b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliFrontend.java
@@ -22,7 +22,6 @@ import org.apache.flink.cdc.cli.utils.FlinkEnvironmentUtils;
import org.apache.flink.cdc.common.annotation.VisibleForTesting;
import org.apache.flink.cdc.common.configuration.Configuration;
import org.apache.flink.cdc.composer.PipelineExecution;
-import org.apache.flink.runtime.jobgraph.RestoreMode;
import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
@@ -34,6 +33,7 @@ import org.apache.commons.cli.Options;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.lang.reflect.InvocationTargetException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Arrays;
@@ -120,18 +120,43 @@ public class CliFrontend {
String savepointPath = commandLine.getOptionValue(SAVEPOINT_PATH_OPTION.getOpt());
boolean allowNonRestoredState =
commandLine.hasOption(SAVEPOINT_ALLOW_NON_RESTORED_OPTION.getOpt());
- final RestoreMode restoreMode;
+ final Object restoreMode;
if (commandLine.hasOption(SAVEPOINT_CLAIM_MODE)) {
restoreMode =
org.apache.flink.configuration.ConfigurationUtils.convertValue(
commandLine.getOptionValue(SAVEPOINT_CLAIM_MODE),
- RestoreMode.class);
+ ConfigurationUtils.getClaimModeClass());
} else {
restoreMode = SavepointConfigOptions.RESTORE_MODE.defaultValue();
}
// allowNonRestoredState is always false because all operators are predefined.
- return SavepointRestoreSettings.forPath(
- savepointPath, allowNonRestoredState, restoreMode);
+
+ return (SavepointRestoreSettings)
+ Arrays.stream(SavepointRestoreSettings.class.getMethods())
+ .filter(
+ method ->
+ method.getName().equals("forPath")
+ && method.getParameterCount() == 3)
+ .findFirst()
+ .map(
+ method -> {
+ try {
+ return method.invoke(
+ null,
+ savepointPath,
+ allowNonRestoredState,
+ restoreMode);
+ } catch (IllegalAccessException
+ | InvocationTargetException e) {
+ throw new RuntimeException(
+ "Failed to invoke SavepointRestoreSettings#forPath nethod.",
+ e);
+ }
+ })
+ .orElseThrow(
+ () ->
+ new RuntimeException(
+ "Failed to resolve SavepointRestoreSettings#forPath method."));
} else {
return SavepointRestoreSettings.none();
}
diff --git a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/ConfigurationUtils.java b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/ConfigurationUtils.java
index 6d2100aa5..af1bc7318 100644
--- a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/ConfigurationUtils.java
+++ b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/ConfigurationUtils.java
@@ -75,4 +75,16 @@ public class ConfigurationUtils {
&& !target.equalsIgnoreCase(LocalExecutor.NAME)
&& !target.equalsIgnoreCase(RemoteExecutor.NAME);
}
+
+ public static Class<?> getClaimModeClass() {
+ try {
+ return Class.forName("org.apache.flink.core.execution.RestoreMode");
+ } catch (ClassNotFoundException ignored) {
+ try {
+ return Class.forName("org.apache.flink.runtime.jobgraph.RestoreMode");
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
}