添加对high-availability.cluster.id的判断，为空则为applicationId

Author: WTZ468071157

launcher/src/main/java/com/dtstack/flink/sql/launcher/ClusterClientFactory.java

@@ -44,6 +44,8 @@
 import org.apache.hadoop.yarn.client.api.YarnClient;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.util.StringHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.net.InetSocketAddress;
 import java.net.URLDecoder;
@@ -59,6 +61,8 @@
  */
 public class ClusterClientFactory {
 
+    private static final Logger LOG = LoggerFactory.getLogger(ClusterClientFactory.class);
+
     public static ClusterClient createClusterClient(Options launcherOptions) throws Exception {
         String mode = launcherOptions.getMode();
         if (mode.equals(ClusterMode.standalone.name())) {
@@ -100,6 +104,9 @@ public static ClusterClient createYarnSessionClient(Options launcherOptions) {
                 String yarnSessionConf = launcherOptions.getYarnSessionConf();
                 yarnSessionConf = URLDecoder.decode(yarnSessionConf, Charsets.UTF_8.toString());
                 Properties yarnSessionConfProperties = PluginUtil.jsonStrToObject(yarnSessionConf, Properties.class);
+
+                LOG.info("current yarn config:\n{}", yarnSessionConfProperties);
+
                 Object yid = yarnSessionConfProperties.get("yid");
 
                 if (null != yid) {
@@ -108,12 +115,18 @@ public static ClusterClient createYarnSessionClient(Options launcherOptions) {
                     applicationId = getYarnClusterApplicationId(yarnClient);
                 }
 
-                Log.info("applicationId={}", applicationId.toString());
+                LOG.info("applicationId={}", applicationId.toString());
+
+                if (config.getString("high-availability.cluster-id", null) == null) {
+                    config.setString("high-availability.cluster-id", applicationId.toString());
+                }
 
                 if (StringUtils.isEmpty(applicationId.toString())) {
                     throw new RuntimeException("No flink session found on yarn cluster.");
                 }
 
+                LOG.info("current config detail:\n{}", config);
+
                 AbstractYarnClusterDescriptor clusterDescriptor = new YarnClusterDescriptor(config, yarnConf, flinkConfDir, yarnClient, false);
                 ClusterClient clusterClient = clusterDescriptor.retrieve(applicationId);
                 clusterClient.setDetached(true);