[code optimize]

Author: [email protected]

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -21,6 +21,7 @@
 package com.dtstack.flink.sql;
 
 import com.dtstack.flink.sql.classloader.DtClassLoader;
+import com.dtstack.flink.sql.constrant.ConfigConstrant;
 import com.dtstack.flink.sql.enums.ClusterMode;
 import com.dtstack.flink.sql.enums.ECacheType;
 import com.dtstack.flink.sql.environment.MyLocalStreamEnvironment;
@@ -99,12 +100,6 @@ public class Main {
 
     private static final Logger LOG = LoggerFactory.getLogger(Main.class);
 
-    private static final int failureRate = 3;
-
-    private static final int failureInterval = 6; //min
-
-    private static final int delayInterval = 10; //sec
-
     public static void main(String[] args) throws Exception {
 
         OptionParser optionParser = new OptionParser(args);
@@ -337,9 +332,9 @@ private static StreamExecutionEnvironment getStreamExeEnv(Properties confPropert
             env.setBufferTimeout(FlinkUtil.getBufferTimeoutMillis(confProperties));
         }
         env.setRestartStrategy(RestartStrategies.failureRateRestart(
-                failureRate,
-                Time.of(failureInterval, TimeUnit.MINUTES),
-                Time.of(delayInterval, TimeUnit.SECONDS)
+                ConfigConstrant.failureRate,
+                Time.of(ConfigConstrant.failureInterval, TimeUnit.MINUTES),
+                Time.of(ConfigConstrant.delayInterval, TimeUnit.SECONDS)
         ));
         FlinkUtil.setStreamTimeCharacteristic(env, confProperties);
         FlinkUtil.openCheckpoint(env, confProperties);

…tack/flink/sql/util/ConfigConstrant.java …flink/sql/constrant/ConfigConstrant.javacore/src/main/java/com/dtstack/flink/sql/util/ConfigConstrant.java renamed to core/src/main/java/com/dtstack/flink/sql/constrant/ConfigConstrant.java core/src/main/java/com/dtstack/flink/sql/util/ConfigConstrant.java renamed to core/src/main/java/com/dtstack/flink/sql/constrant/ConfigConstrant.java

@@ -18,7 +18,7 @@
 
 
 
-package com.dtstack.flink.sql.util;
+package com.dtstack.flink.sql.constrant;
 
 
 /**
@@ -51,4 +51,11 @@ public class ConfigConstrant {
 
     public static final String FLINK_TIME_CHARACTERISTIC_KEY = "time.characteristic";
 
+    // restart plocy
+    public static final int failureRate = 3;
+
+    public static final int failureInterval = 6; //min
+
+    public static final int delayInterval = 10; //sec
+
 }

…k/sql/threadFactory/DTThreadFactory.java …k/flink/sql/factory/DTThreadFactory.javacore/src/main/java/com/dtstack/flink/sql/threadFactory/DTThreadFactory.java renamed to core/src/main/java/com/dtstack/flink/sql/factory/DTThreadFactory.java core/src/main/java/com/dtstack/flink/sql/threadFactory/DTThreadFactory.java renamed to core/src/main/java/com/dtstack/flink/sql/factory/DTThreadFactory.java

@@ -18,7 +18,7 @@
 
 
 
-package com.dtstack.flink.sql.threadFactory;
+package com.dtstack.flink.sql.factory;
 
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicInteger;

core/src/main/java/com/dtstack/flink/sql/side/AllReqRow.java

@@ -20,7 +20,7 @@
 
 package com.dtstack.flink.sql.side;
 
-import com.dtstack.flink.sql.threadFactory.DTThreadFactory;
+import com.dtstack.flink.sql.factory.DTThreadFactory;
 import org.apache.flink.api.common.functions.RichFlatMapFunction;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.types.Row;

core/src/main/java/com/dtstack/flink/sql/util/FlinkUtil.java

@@ -21,6 +21,7 @@
 package com.dtstack.flink.sql.util;
 
 
+import com.dtstack.flink.sql.constrant.ConfigConstrant;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.runtime.state.filesystem.FsStateBackend;

hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAsyncReqRow.java

@@ -30,7 +30,7 @@
 import com.dtstack.flink.sql.side.hbase.rowkeydealer.PreRowKeyModeDealerDealer;
 import com.dtstack.flink.sql.side.hbase.rowkeydealer.RowKeyEqualModeDealer;
 import com.dtstack.flink.sql.side.hbase.table.HbaseSideTableInfo;
-import com.dtstack.flink.sql.threadFactory.DTThreadFactory;
+import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.google.common.collect.Maps;
 import com.stumbleupon.async.Deferred;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;

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

@@ -20,8 +20,8 @@
 
 import com.dtstack.flink.sql.option.Options;
 import com.dtstack.flink.sql.util.PluginUtil;
-import com.dtstack.flink.yarn.JobParameter;
-import com.dtstack.flink.yarn.YarnClusterConfiguration;
+import com.dtstack.flink.sql.yarn.JobParameter;
+import com.dtstack.flink.sql.yarn.YarnClusterConfiguration;
 import org.apache.commons.io.Charsets;
 import org.apache.commons.lang.StringUtils;
 import org.apache.flink.client.program.ClusterClient;
@@ -129,7 +129,7 @@ public static ClusterClient createYarnClient(Options launcherOptions, String mod
                         //jobmanager+taskmanager param
                         JobParameter appConf = new JobParameter(confProperties);
 
-                        com.dtstack.flink.yarn.YarnClusterDescriptor clusterDescriptor = new com.dtstack.flink.yarn.YarnClusterDescriptor(
+                        com.dtstack.flink.sql.yarn.YarnClusterDescriptor clusterDescriptor = new com.dtstack.flink.sql.yarn.YarnClusterDescriptor(
                                 clusterConf, yarnClient, appConf,applicationId,  launcherOptions.getName(),null );
                         clusterClient = clusterDescriptor.deploy();
 

…com/dtstack/flink/yarn/JobParameter.java …dtstack/flink/sql/yarn/JobParameter.javacore/src/main/java/com/dtstack/flink/yarn/JobParameter.java renamed to launcher/src/main/java/com/dtstack/flink/sql/yarn/JobParameter.java core/src/main/java/com/dtstack/flink/yarn/JobParameter.java renamed to launcher/src/main/java/com/dtstack/flink/sql/yarn/JobParameter.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.dtstack.flink.yarn;
+package com.dtstack.flink.sql.yarn;
 
 import java.util.Objects;
 import java.util.Properties;

…flink/yarn/YarnClusterConfiguration.java …k/sql/yarn/YarnClusterConfiguration.javacore/src/main/java/com/dtstack/flink/yarn/YarnClusterConfiguration.java renamed to launcher/src/main/java/com/dtstack/flink/sql/yarn/YarnClusterConfiguration.java core/src/main/java/com/dtstack/flink/yarn/YarnClusterConfiguration.java renamed to launcher/src/main/java/com/dtstack/flink/sql/yarn/YarnClusterConfiguration.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.dtstack.flink.yarn;
+package com.dtstack.flink.sql.yarn;
 
 import org.apache.flink.configuration.Configuration;
 import org.apache.hadoop.fs.Path;

…ck/flink/yarn/YarnClusterDescriptor.java …link/sql/yarn/YarnClusterDescriptor.javacore/src/main/java/com/dtstack/flink/yarn/YarnClusterDescriptor.java renamed to launcher/src/main/java/com/dtstack/flink/sql/yarn/YarnClusterDescriptor.java core/src/main/java/com/dtstack/flink/yarn/YarnClusterDescriptor.java renamed to launcher/src/main/java/com/dtstack/flink/sql/yarn/YarnClusterDescriptor.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.dtstack.flink.yarn;
+package com.dtstack.flink.sql.yarn;
 
 import org.apache.flink.client.deployment.ClusterDeploymentException;
 import org.apache.flink.client.deployment.ClusterSpecification;
@@ -42,7 +42,6 @@
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import static java.util.Objects.requireNonNull;
 import static org.apache.hadoop.yarn.api.records.YarnApplicationState.NEW;