Merge branch 'v1.8.0_dev' into feat_1.8_codereview

# Conflicts:
#	cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraOutputFormat.java
#	core/src/main/java/com/dtstack/flink/sql/Main.java
#	core/src/main/java/com/dtstack/flink/sql/constrant/ConfigConstrant.java
#	core/src/main/java/com/dtstack/flink/sql/environment/StreamEnvConfigManager.java
#	core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java
#	core/src/main/java/com/dtstack/flink/sql/side/BaseAsyncReqRow.java
#	core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java
#	elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/ElasticsearchSink.java
#	hbase/hbase-side/hbase-side-core/src/main/java/com/dtstack/flink/sql/side/hbase/enums/EReplaceType.java
#	hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseOutputFormat.java
#	kafka/kafka-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java
#	kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java
#	kafka10/kafka10-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java
#	kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java
#	kudu/kudu-side/kudu-all-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAllReqRow.java
#	mongo/mongo-side/mongo-all-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAllReqRow.java
#	mongo/mongo-side/mongo-async-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAsyncReqRow.java
#	mongo/mongo-sink/src/main/java/com/dtstack/flink/sql/sink/mongo/MongoOutputFormat.java
#	rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/AbstractRdbAllReqRow.java
#	rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/writer/AbstractUpsertWriter.java
#	rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/writer/AppendOnlyWriter.java
#	redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisOutputFormat.java

Author: zoudaokoulife

README.md

@@ -149,6 +149,10 @@ 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）
+    	* logLevel: 日志级别动态配置（默认info）
         * [prometheus 相关参数](docs/prometheus.md) per_job可指定metric写入到外部监控组件,以prometheus pushgateway举例
 
 
@@ -181,6 +185,7 @@ sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack
 	* 描述：yarn session 模式下指定的运行的一些参数，[可参考](https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/cli.html),目前只支持指定yid
 	* 必选：否
 	* 默认值：false	
+	
 
 ## 2 结构
 ### 2.1 源表插件

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

@@ -18,6 +18,12 @@
 
 package com.dtstack.flink.sql.side.cassandra;
 
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.table.runtime.types.CRow;
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Collector;
+
 import com.datastax.driver.core.Cluster;
 import com.datastax.driver.core.ConsistencyLevel;
 import com.datastax.driver.core.HostDistance;
@@ -33,15 +39,11 @@
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo;
-import org.apache.calcite.sql.JoinType;
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
-import org.apache.flink.table.runtime.types.CRow;
-import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
-import org.apache.flink.types.Row;
-import org.apache.flink.util.Collector;
+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;
 
@@ -220,9 +222,9 @@ private Session getConn(CassandraSideTableInfo tableInfo) {
                 //重试策略
                 RetryPolicy retryPolicy = DowngradingConsistencyRetryPolicy.INSTANCE;
 
-                for (String server : address.split(",")) {
-                    cassandraPort = Integer.parseInt(server.split(":")[1]);
-                    serversList.add(InetAddress.getByName(server.split(":")[0]));
+                for (String server : StringUtils.split(address, ",")) {
+                    cassandraPort = Integer.parseInt(StringUtils.split(server, ":")[1]);
+                    serversList.add(InetAddress.getByName(StringUtils.split(server, ":")[0]));
                 }
 
                 if (userName == null || userName.isEmpty() || password == null || password.isEmpty()) {
@@ -276,7 +278,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 = sideInfo.getSideSelectFields().split(",");
+            String[] sideFieldNames = StringUtils.split(sideInfo.getSideSelectFields(), ",");
             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

@@ -19,6 +19,13 @@
 
 package com.dtstack.flink.sql.side.cassandra;
 
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.async.ResultFuture;
+import org.apache.flink.table.runtime.types.CRow;
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
+import org.apache.flink.types.Row;
+
 import com.datastax.driver.core.Cluster;
 import com.datastax.driver.core.ConsistencyLevel;
 import com.datastax.driver.core.HostDistance;
@@ -38,18 +45,13 @@
 import com.dtstack.flink.sql.side.cache.CacheObj;
 import com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo;
 import com.google.common.base.Function;
+import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.AsyncFunction;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import io.vertx.core.json.JsonArray;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import org.apache.flink.configuration.Configuration;
-import com.google.common.collect.Lists;
-import org.apache.flink.streaming.api.functions.async.ResultFuture;
-import org.apache.flink.table.runtime.types.CRow;
-import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
-import org.apache.flink.types.Row;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -133,9 +135,9 @@ private void connCassandraDB(CassandraSideTableInfo tableInfo) {
                 //重试策略
                 RetryPolicy retryPolicy = DowngradingConsistencyRetryPolicy.INSTANCE;
 
-                for (String server : address.split(",")) {
-                    cassandraPort = Integer.parseInt(server.split(":")[1]);
-                    serversList.add(InetAddress.getByName(server.split(":")[0]));
+                for (String server : StringUtils.split(address, ",")) {
+                    cassandraPort = Integer.parseInt(StringUtils.split(server, ":")[1]);
+                    serversList.add(InetAddress.getByName(StringUtils.split(server, ":")[0]));
                 }
 
                 if (userName == null || userName.isEmpty() || password == null || password.isEmpty()) {

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

@@ -38,6 +38,12 @@
 
 package com.dtstack.flink.sql.sink.cassandra;
 
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.types.Row;
+
 import com.datastax.driver.core.Cluster;
 import com.datastax.driver.core.ConsistencyLevel;
 import com.datastax.driver.core.HostDistance;
@@ -48,14 +54,11 @@
 import com.datastax.driver.core.SocketOptions;
 import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
 import com.datastax.driver.core.policies.RetryPolicy;
-import com.dtstack.flink.sql.outputformat.AbstractDtRichOutputFormat;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.types.Row;
+import com.dtstack.flink.sql.outputformat.DtRichOutputFormat;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
 import java.io.IOException;
 import java.net.InetAddress;
 import java.sql.DriverManager;
@@ -145,9 +148,9 @@ public void open(int taskNumber, int numTasks) {
                 //重试策略
                 RetryPolicy retryPolicy = DowngradingConsistencyRetryPolicy.INSTANCE;
 
-                for (String server : address.split(",")) {
-                    cassandraPort = Integer.parseInt(server.split(":")[1]);
-                    serversList.add(InetAddress.getByName(server.split(":")[0]));
+                for (String server : StringUtils.split(address, ",")) {
+                    cassandraPort = Integer.parseInt(StringUtils.split(server, ":")[1]);
+                    serversList.add(InetAddress.getByName(StringUtils.split(server, ":")[0]));
                 }
 
                 if (userName == null || userName.isEmpty() || password == null || password.isEmpty()) {
@@ -216,7 +219,7 @@ private void insertWrite(Row row) {
             }
         } catch (Exception e) {
             if(outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0){
-                LOG.error("record insert failed ..", row.toString().substring(0, 100));
+                LOG.error("record insert failed, total dirty num:{}, current record:{}", outDirtyRecords.getCount(), row.toString());
                 LOG.error("", e);
             }
 

clickhouse/clickhouse-side/clickhouse-all-side/pom.xml

@@ -41,7 +41,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-
+                                    <exclude>org.slf4j</exclude>
                                 </excludes>
                             </artifactSet>
                             <filters>

clickhouse/clickhouse-side/clickhouse-async-side/pom.xml

@@ -41,7 +41,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-
+                                    <exclude>org.slf4j</exclude>
                                 </excludes>
                             </artifactSet>
                             <filters>

clickhouse/clickhouse-sink/pom.xml

@@ -40,7 +40,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-
+                                    <exclude>org.slf4j</exclude>
                                 </excludes>
                             </artifactSet>
                             <filters>

core/pom.xml

@@ -20,6 +20,7 @@
         <calcite.server.version>1.16.0</calcite.server.version>
         <jackson.version>2.7.9</jackson.version>
         <guava.version>19.0</guava.version>
+        <logger.tool.version>1.0.0-SNAPSHOT</logger.tool.version>
     </properties>
 
     <dependencies>
@@ -122,11 +123,10 @@
             <version>4.12</version>
         </dependency>
         <dependency>
-            <groupId>ch.qos.logback</groupId>
-            <artifactId>logback-classic</artifactId>
-            <version>1.1.7</version>
+            <groupId>com.aiweiergou</groupId>
+            <artifactId>tools-logger</artifactId>
+            <version>${logger.tool.version}</version>
         </dependency>
-
     </dependencies>
 
     <build>

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

@@ -27,7 +27,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-
 /**
  * Date: 2018/6/26
  * Company: www.dtstack.com
@@ -39,6 +38,7 @@ public class Main {
 
     public static void main(String[] args) throws Exception {
         ParamsInfo paramsInfo = ExecuteProcessHelper.parseParams(args);
+        ExecuteProcessHelper.setLogLevel(paramsInfo);
         StreamExecutionEnvironment env = ExecuteProcessHelper.getStreamExecution(paramsInfo);
         env.execute(paramsInfo.getName());
         LOG.info("program {} execution success", paramsInfo.getName());

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

@@ -63,6 +63,10 @@ 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";
+
+    public static final String LOG_LEVEL_KEY = "logLevel";
+
 
     // restart plocy
     public static final int FAILUEE_RATE = 3;
@@ -71,4 +75,9 @@ public class ConfigConstrant {
 
     public static final int DELAY_INTERVAL = 10; //sec
 
+    public static final String FAILUREINTERVAL = "failure.interval"; //min
+
+    public static final String  DELAYINTERVAL= "delay.interval"; //sec
+
+
 }