zhoujinsong commented on code in PR #2676:
URL: https://github.com/apache/incubator-amoro/pull/2676#discussion_r1549183261
##########
ams/server/src/main/java/com/netease/arctic/server/manager/FlinkOptimizerContainer.java:
##########
@@ -128,30 +170,41 @@ public void init(String name, Map<String, String>
containerProperties) {
@Override
protected Map<String, String> doScaleOut(Resource resource) {
- String startUpArgs = this.buildOptimizerStartupArgsString(resource);
- Runtime runtime = Runtime.getRuntime();
- try {
- String exportCmd = String.join(" && ", exportSystemProperties());
- String startUpCmd = String.format("%s && %s", exportCmd, startUpArgs);
- String[] cmd = {"/bin/sh", "-c", startUpCmd};
- LOG.info("Starting flink optimizer using command : {}", startUpCmd);
- Process exec = runtime.exec(cmd);
- Map<String, String> startUpStatesMap = Maps.newHashMap();
- switch (target) {
- case YARN_PER_JOB:
- case YARN_APPLICATION:
- String applicationId = fetchCommandOutput(exec,
yarnApplicationIdReader);
- if (applicationId != null) {
- startUpStatesMap.put(YARN_APPLICATION_ID_PROPERTY, applicationId);
- }
- break;
- case KUBERNETES_APPLICATION:
- startUpStatesMap.put(KUBERNETES_CLUSTER_ID_PROPERTY,
kubernetesClusterId(resource));
- break;
+ if (target.runByFlinkRestClient()) {
+ try {
+ JobID jobID = runFlinkOptimizerJob(resource, loadFlinkConfig());
Review Comment:
Should we support reloading session cluster information by using flink-conf.
properties in the container and group?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
