在standalone模式下实现基于save point的断点续跑
- Dominant language
- Java
- Stars
- 4.1k
- Forks
- 1.7k
- PR merge metrics
- No merged PRs in 30d
Description
**Describe the bug**
只能在yarn引擎上续跑,standalone无法续跑
**To Reproduce**
修改JobGraphUtil类buildJobGraph方法

public static JobGraph buildJobGraph(Options launcherOptions, String[] programArgs)
throws Exception {
String pluginRoot = launcherOptions.getFlinkxDistDir();
String coreJarPath = PluginInfoUtil.getCoreJarPath(pluginRoot);
File jarFile = new File(coreJarPath);
Configuration flinkConf = launcherOptions.loadFlinkConfiguration();
PackagedProgram program = null;
if (flinkConf.containsKey("execution.savepoint.path")) {
String savepointPath = flinkConf.getString("execution.savepoint.path", null);
program =
PackagedProgram.newBuilder()
.setJarFile(jarFile)
.setEntryPointClassName(PluginInfoUtil.getMainClass())
.setConfiguration(launcherOptions.loadFlinkConfiguration())
.setSavepointRestoreSettings(
SavepointRestoreSettings.forPath(savepointPath))
.setArguments(programArgs)
.build();
} else {
program =
PackagedProgram.newBuilder()
.setJarFile(jarFile)
.setEntryPointClassName(PluginInfoUtil.getMainClass())
.setConfiguration(launcherOptions.loadFlinkConfiguration())
.setArguments(programArgs)
.build();
}
JobGraph jobGraph =
PackagedProgramUtils.createJobGraph(
program,
launcherOptions.loadFlinkConfiguration(),
flinkConf.getInteger(DEFAULT_PARALLELISM),
false);
List pluginClassPath =
jobGraph.getUserArtifacts().entrySet().stream()
.filter(tmp -> tmp.getKey().startsWith("class_path"))
.map(tmp -> new File(tmp.getValue().filePath))
.map(
file -> {
try {
return file.toURI().toURL();
} catch (MalformedURLException e) {
LOG.error(e.getMessage());
}
return null;
})
.collect(Collectors.toList());
jobGraph.setClasspaths(pluginClassPath);
return jobGraph;
}
Contributor guide
Assessment
This issue has not been assessed yet.