use RestClusterClient

Author: WTZ468071157

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

@@ -6,9 +6,9 @@
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
- *
+ * <p>
  * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * <p>
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -26,6 +26,8 @@
 import org.apache.commons.lang.StringUtils;
 import org.apache.flink.client.program.ClusterClient;
 import org.apache.flink.client.program.MiniClusterClient;
+import org.apache.flink.client.program.rest.RestClusterClient;
+import org.apache.flink.client.program.rest.RestClusterClientConfiguration;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.GlobalConfiguration;
 import org.apache.flink.configuration.JobManagerOptions;
@@ -70,10 +72,7 @@ public static ClusterClient createClusterClient(Options launcherOptions) throws
     public static ClusterClient createStandaloneClient(Options launcherOptions) throws Exception {
         String flinkConfDir = launcherOptions.getFlinkconf();
         Configuration config = GlobalConfiguration.loadConfiguration(flinkConfDir);
-        MiniClusterConfiguration.Builder configBuilder = new MiniClusterConfiguration.Builder();
-        configBuilder.setConfiguration(config);
-        MiniCluster miniCluster = new MiniCluster(configBuilder.build());
-        MiniClusterClient clusterClient = new MiniClusterClient(config, miniCluster);
+        RestClusterClient clusterClient = new RestClusterClient<>(config, "clusterClient");
         LeaderConnectionInfo connectionInfo = clusterClient.getClusterConnectionInfo();
         InetSocketAddress address = AkkaUtils.getInetSocketAddressFromAkkaURL(connectionInfo.getAddress());
         config.setString(JobManagerOptions.ADDRESS, address.getAddress().getHostName());
@@ -122,7 +121,7 @@ public static ClusterClient createYarnSessionClient(Options launcherOptions) {
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
-        }else{
+        } else {
             throw new RuntimeException("yarn mode must set param of 'yarnconf'!!!");
         }
     }

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

@@ -89,8 +89,7 @@ public static void main(String[] args) throws Exception {
         confProp = URLDecoder.decode(confProp, Charsets.UTF_8.toString());
         Properties confProperties = PluginUtil.jsonStrToObject(confProp, Properties.class);
 
-//        LOG.info("current mode is {}", mode);
-        System.out.println("current mode is " + mode);
+        LOG.info("current mode is {}", mode);
 
         if (mode.equals(ClusterMode.local.name())) {
             String[] localArgs = argList.toArray(new String[0]);