解决elasticsearch-side分支与test_1.8_3.1.0x的冲突

Author: dengdejiushini

README.md

@@ -149,9 +149,6 @@ sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack
         * taskmanager.slots：per_job模式下指定每个taskmanager对应的slot数量(默认1)
         * savePointPath：任务恢复点的路径（默认无）
         * allowNonRestoredState：指示保存点是否允许非还原状态的标志（默认false）
-        * restore.enable：是否失败重启（默认是true）
-        * failure.interval：衡量失败率的时间段，单位分钟（默认6m）
-        * delay.interval：连续两次重启尝试间的间隔，单位是秒（默认10s）
         * [prometheus 相关参数](docs/prometheus.md) per_job可指定metric写入到外部监控组件,以prometheus pushgateway举例
 
 

cassandra/cassandra-side/cassandra-all-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAllReqRow.java

@@ -43,7 +43,6 @@
 import com.google.common.collect.Maps;
 import org.apache.calcite.sql.JoinType;
 import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -224,9 +223,9 @@ private Session getConn(CassandraSideTableInfo tableInfo) {
                 //重试策略
                 RetryPolicy retryPolicy = DowngradingConsistencyRetryPolicy.INSTANCE;
 
-                for (String server : StringUtils.split(address, ",")) {
-                    cassandraPort = Integer.parseInt(StringUtils.split(server, ":")[1]);
-                    serversList.add(InetAddress.getByName(StringUtils.split(server, ":")[0]));
+                for (String server : address.split(",")) {
+                    cassandraPort = Integer.parseInt(server.split(":")[1]);
+                    serversList.add(InetAddress.getByName(server.split(":")[0]));
                 }
 
                 if (userName == null || userName.isEmpty() || password == null || password.isEmpty()) {
@@ -280,7 +279,7 @@ private void loadData(Map<String, List<Map<String, Object>>> tmpCache) throws SQ
             //load data from table
             String sql = sideInfo.getSqlCondition() + " limit " + FETCH_SIZE;
             ResultSet resultSet = session.execute(sql);
-            String[] sideFieldNames = StringUtils.split(sideInfo.getSideSelectFields(), ",");
+            String[] sideFieldNames = sideInfo.getSideSelectFields().split(",");
             for (com.datastax.driver.core.Row row : resultSet) {
                 Map<String, Object> oneRow = Maps.newHashMap();
                 for (String fieldName : sideFieldNames) {

cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncReqRow.java

@@ -51,7 +51,6 @@
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import io.vertx.core.json.JsonArray;
-import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -135,9 +134,9 @@ private void connCassandraDB(CassandraSideTableInfo tableInfo) {
                 //重试策略
                 RetryPolicy retryPolicy = DowngradingConsistencyRetryPolicy.INSTANCE;
 
-                for (String server : StringUtils.split(address, ",")) {
-                    cassandraPort = Integer.parseInt(StringUtils.split(server, ":")[1]);
-                    serversList.add(InetAddress.getByName(StringUtils.split(server, ":")[0]));
+                for (String server : address.split(",")) {
+                    cassandraPort = Integer.parseInt(server.split(":")[1]);
+                    serversList.add(InetAddress.getByName(server.split(":")[0]));
                 }
 
                 if (userName == null || userName.isEmpty() || password == null || password.isEmpty()) {

cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraOutputFormat.java

@@ -55,7 +55,6 @@
 import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
 import com.datastax.driver.core.policies.RetryPolicy;
 import com.dtstack.flink.sql.outputformat.DtRichOutputFormat;
-import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -148,9 +147,9 @@ public void open(int taskNumber, int numTasks) {
                 //重试策略
                 RetryPolicy retryPolicy = DowngradingConsistencyRetryPolicy.INSTANCE;
 
-                for (String server : StringUtils.split(address, ",")) {
-                    cassandraPort = Integer.parseInt(StringUtils.split(server, ":")[1]);
-                    serversList.add(InetAddress.getByName(StringUtils.split(server, ":")[0]));
+                for (String server : address.split(",")) {
+                    cassandraPort = Integer.parseInt(server.split(":")[1]);
+                    serversList.add(InetAddress.getByName(server.split(":")[0]));
                 }
 
                 if (userName == null || userName.isEmpty() || password == null || password.isEmpty()) {

core/src/main/java/com/dtstack/flink/sql/constrant/ConfigConstrant.java

@@ -63,14 +63,12 @@ public class ConfigConstrant {
     public static final String CHECKPOINTS_DIRECTORY_KEY = "state.checkpoints.dir";
     public static final String STATE_BACKEND_INCREMENTAL_KEY = "state.backend.incremental";
 
-    public static final String RESTOREENABLE = "restore.enable";
-
 
     // restart plocy
     public static final int failureRate = 3;
 
-    public static final String FAILUREINTERVAL = "failure.interval"; //min
+    public static final int failureInterval = 6; //min
 
-    public static final String  DELAYINTERVAL= "delay.interval"; //sec
+    public static final int delayInterval = 10; //sec
 
 }

core/src/main/java/com/dtstack/flink/sql/environment/StreamEnvConfigManager.java

@@ -18,12 +18,6 @@
 
 package com.dtstack.flink.sql.environment;
 
-import com.dtstack.flink.sql.constrant.ConfigConstrant;
-import com.dtstack.flink.sql.enums.EStateBackend;
-import com.dtstack.flink.sql.util.MathUtil;
-import com.dtstack.flink.sql.util.PropertiesUtils;
-import org.apache.commons.lang3.BooleanUtils;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.api.common.time.Time;
@@ -40,6 +34,13 @@
 import org.apache.flink.table.api.StreamQueryConfig;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
 
+import com.dtstack.flink.sql.constrant.ConfigConstrant;
+import com.dtstack.flink.sql.enums.EStateBackend;
+import com.dtstack.flink.sql.util.MathUtil;
+import com.dtstack.flink.sql.util.PropertiesUtils;
+import org.apache.commons.lang3.BooleanUtils;
+import org.apache.commons.lang3.StringUtils;
+
 import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
@@ -101,15 +102,11 @@ public static void streamExecutionEnvironmentConfig(StreamExecutionEnvironment s
             }
         });
 
-        if(isRestore(confProperties).get()){
-            streamEnv.setRestartStrategy(RestartStrategies.failureRateRestart(
-                    ConfigConstrant.failureRate,
-                    Time.of(getFailureInterval(confProperties).get(), TimeUnit.MINUTES),
-                    Time.of(getDelayInterval(confProperties).get(), TimeUnit.SECONDS)
-            ));
-        } else {
-            streamEnv.setRestartStrategy(RestartStrategies.noRestart());
-        }
+        streamEnv.setRestartStrategy(RestartStrategies.failureRateRestart(
+                ConfigConstrant.failureRate,
+                Time.of(ConfigConstrant.failureInterval, TimeUnit.MINUTES),
+                Time.of(ConfigConstrant.delayInterval, TimeUnit.SECONDS)
+        ));
 
         // checkpoint config
         Optional<Boolean> checkpointingEnabled = isCheckpointingEnabled(confProperties);
@@ -167,20 +164,6 @@ public static Optional<Long> getAutoWatermarkInterval(Properties properties) {
         return StringUtils.isNotBlank(autoWatermarkInterval) ? Optional.of(Long.valueOf(autoWatermarkInterval)) : Optional.empty();
     }
 
-    public static Optional<Boolean> isRestore(Properties properties){
-        String restoreEnable = properties.getProperty(ConfigConstrant.RESTOREENABLE, "true");
-        return Optional.of(Boolean.valueOf(restoreEnable));
-    }
-
-    public static Optional<Integer> getDelayInterval(Properties properties){
-        String delayInterval = properties.getProperty(ConfigConstrant.DELAYINTERVAL, "10");
-        return Optional.of(Integer.valueOf(delayInterval));
-    }
-    public static Optional<Integer> getFailureInterval(Properties properties){
-        String failureInterval = properties.getProperty(ConfigConstrant.FAILUREINTERVAL, "6");
-        return Optional.of(Integer.valueOf(failureInterval));
-    }
-
     /**
      * #ProcessingTime(默认), IngestionTime, EventTime
      * @param properties

core/src/main/java/com/dtstack/flink/sql/table/TableInfoParser.java

@@ -29,7 +29,6 @@
 import com.dtstack.flink.sql.util.MathUtil;
 import com.google.common.base.Strings;
 import com.google.common.collect.Maps;
-import org.apache.commons.lang3.StringUtils;
 
 import java.util.Map;
 import java.util.regex.Matcher;
@@ -111,7 +110,7 @@ public TableInfo parseWithTableType(int tableType, CreateTableParser.SqlParserRe
      * @return
      */
     private static boolean checkIsSideTable(String tableField){
-        String[] fieldInfos = StringUtils.split(tableField, ",");
+        String[] fieldInfos = tableField.split(",");
         for(String field : fieldInfos){
             Matcher matcher = SIDE_PATTERN.matcher(field.trim());
             if(matcher.find()){

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

@@ -180,14 +180,14 @@ public static String addJdbcParam(String dbUrl, Map<String, String> addParams, b
             return dbUrl;
         }
 
-        String[] splits = StringUtils.split(dbUrl, "\\?");
+        String[] splits = dbUrl.split("\\?");
         String preStr = splits[0];
         Map<String, String> params = Maps.newHashMap();
         if(splits.length > 1){
             String existsParamStr = splits[1];
-            String[] existsParams = StringUtils.split(existsParamStr, "&");
+            String[] existsParams = existsParamStr.split("&");
             for(String oneParam : existsParams){
-                String[] kv = StringUtils.split(oneParam, "=");
+                String[] kv = oneParam.split("=");
                 if(kv.length != 2){
                     throw new RuntimeException("illegal dbUrl:" + dbUrl);
                 }

elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/ElasticsearchSink.java

@@ -118,7 +118,7 @@ private RichSinkFunction createEsSinkFunction(){
         List<InetSocketAddress> transports = new ArrayList<>();
 
         for(String address : esAddressList){
-            String[] infoArray = StringUtils.split(address, ":");
+            String[] infoArray = address.split(":");
             int port = 9300;
             String host = infoArray[0];
             if(infoArray.length > 1){
@@ -169,7 +169,7 @@ public ElasticsearchSink genStreamSink(TargetTableInfo targetTableInfo) {
         esTableInfo = elasticsearchTableInfo;
         clusterName = elasticsearchTableInfo.getClusterName();
         String address = elasticsearchTableInfo.getAddress();
-        String[] addr = StringUtils.split(address, ",");
+        String[] addr = address.split(",");
         esAddressList = Arrays.asList(addr);
         index = elasticsearchTableInfo.getIndex();
         type = elasticsearchTableInfo.getEsType();

elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/EsUtil.java

@@ -22,7 +22,6 @@
 import org.apache.flink.util.Preconditions;
 
 import com.dtstack.flink.sql.util.DtStringUtil;
-import org.apache.commons.lang3.StringUtils;
 
 import java.util.HashMap;
 import java.util.List;
@@ -42,7 +41,7 @@ public static Map<String, Object> rowToJsonMap(Row row, List<String> fields, Lis
         int i = 0;
         for(; i < fields.size(); ++i) {
             String field = fields.get(i);
-            String[] parts = StringUtils.split(field, "\\.");
+            String[] parts = field.split("\\.");
             Map<String, Object> currMap = jsonMap;
             for(int j = 0; j < parts.length - 1; ++j) {
                 String key = parts[j];