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

# Conflicts:
#   hbase/hbase-side/hbase-side-core/src/main/java/com/dtstack/flink/sql/side/hbase/RowKeyBuilder.java

Author: zoudaokoulife

.gitlab-ci.yml

@@ -2,7 +2,7 @@ build:
   stage: test
   script:
     - mvn clean org.jacoco:jacoco-maven-plugin:0.7.8:prepare-agent package -Dmaven.test.failure.ignore=true -q
-    - mvn sonar:sonar -Dsonar.projectKey="dt-insight-engine/flinkStreamSQL" -Dsonar.branch.name="v1.8.0_dev" -Dsonar.login=11974c5e9a29625efa09fdc3c3fdc031efb1aab1 -Dsonar.host.url=http://172.16.100.198:9000 -Dsonar.jdbc.url=jdbc:postgresql://172.16.100.198:5432/sonar -Dsonar.java.binaries=target/sonar
+    - mvn sonar:sonar -Dsonar.projectKey="dt-insight-engine/flinkStreamSQL"  -Dsonar.login=11974c5e9a29625efa09fdc3c3fdc031efb1aab1 -Dsonar.host.url=http://172.16.100.198:9000 -Dsonar.jdbc.url=jdbc:postgresql://172.16.100.198:5432/sonar -Dsonar.java.binaries=target/sonar
     - sh ci/sonar_notify.sh
   only:
     - v1.8.0_dev

README.md

@@ -7,34 +7,11 @@
 >  >  * 支持原生FLinkSQL所有的语法
 >  >  * 扩展了输入和输出的性能指标到promethus
 
- ## 新特性:
-  * 1.kafka源表支持not null语法,支持字符串类型的时间转换。
-  * 2.rdb维表与DB建立连接时，周期进行连接，防止连接断开。rdbsink写入时，对连接进行检查。
-  * 3.异步维表支持非等值连接，比如：<>,<,>。
-  * 4.增加kafka数组解析
-  * 5.增加kafka1.0以上版本的支持
-  * 6.增加postgresql、kudu、clickhouse维表、结果表的支持
-  * 7.支持插件的依赖方式,参考pluginLoadMode参数
-  * 8.支持cep处理
-  * 9.支持udaf
-  * 10.支持谓词下移
-  * 11.支持状态的ttl
-  
- ## BUG修复：
-  * 1.修复不能解析sql中orderby,union语法。
-  * 2.修复yarnPer模式提交失败的异常。
-  * 3.一些bug的修复
- 
 # 已支持
   * 源表：kafka 0.9、0.10、0.11、1.x版本
   * 维表：mysql, SQlServer,oracle, hbase, mongo, redis, cassandra, serversocket, kudu, postgresql, clickhouse, impala, db2, sqlserver
   * 结果表：mysql, SQlServer, oracle, hbase, elasticsearch5.x, mongo, redis, cassandra, console, kudu, postgresql, clickhouse, impala, db2, sqlserver
 
-# 后续开发计划
-  * 维表快照
-  * kafka avro格式
-  * topN
-
 ## 1 快速起步
 ### 1.1 运行模式
 
@@ -205,6 +182,7 @@ sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack
 * [impala 结果表插件](docs/impalaSink.md)
 * [db2 结果表插件](docs/db2Sink.md)
 * [sqlserver 结果表插件](docs/sqlserverSink.md)
+* [kafka 结果表插件](docs/kafkaSink.md)
 
 ### 2.3 维表插件
 * [hbase 维表插件](docs/hbaseSide.md)

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

@@ -267,7 +267,7 @@ private void loadData(Map<String, List<Map<String, Object>>> tmpCache) throws SQ
                         String connInfo = "address:" + tableInfo.getAddress() + ";userName:" + tableInfo.getUserName()
                                 + ",pwd:" + tableInfo.getPassword();
                         LOG.warn("get conn fail, wait for 5 sec and try again, connInfo:" + connInfo);
-                        Thread.sleep(5 * 1000);
+                        Thread.sleep(LOAD_DATA_ERROR_SLEEP_TIME);
                     } catch (InterruptedException e1) {
                         LOG.error("", e1);
                     }

core/pom.xml

@@ -116,12 +116,6 @@
             <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
             <version>${flink.version}</version>
         </dependency>
-
-        <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-            <version>4.12</version>
-        </dependency>
         <dependency>
             <groupId>com.aiweiergou</groupId>
             <artifactId>tools-logger</artifactId>

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

@@ -113,14 +113,13 @@ public JobExecutionResult execute(String jobName) throws Exception {
             LOG.info("Running job on local embedded Flink mini cluster");
         }
 
-        MiniCluster exec = new MiniCluster(configBuilder.build());
-        try {
+        try (MiniCluster exec = new MiniCluster(configBuilder.build());) {
             exec.start();
-            return exec.executeJobBlocking(jobGraph);
-        }
-        finally {
+            JobExecutionResult jobExecutionResult = exec.executeJobBlocking(jobGraph);
             transformations.clear();
-            exec.closeAsync();
+            return jobExecutionResult;
+        } catch (Exception e) {
+            throw new RuntimeException(e);
         }
     }
 }

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -282,9 +282,7 @@ public static Set<URL> registerTable(SqlTree sqlTree, StreamExecutionEnvironment
 
                 RowTypeInfo typeInfo = new RowTypeInfo(adaptTable.getSchema().getFieldTypes(), adaptTable.getSchema().getFieldNames());
                 DataStream adaptStream = tableEnv.toRetractStream(adaptTable, typeInfo)
-                        .map((Tuple2<Boolean, Row> f0) -> {
-                            return f0.f1;
-                        })
+                        .map((Tuple2<Boolean, Row> f0) -> f0.f1)
                         .returns(typeInfo);
 
                 String fields = String.join(",", typeInfo.getFieldNames());

core/src/main/java/com/dtstack/flink/sql/format/SerializationMetricWrapper.java

@@ -24,7 +24,7 @@
 import org.apache.flink.metrics.Counter;
 import org.apache.flink.metrics.Meter;
 import org.apache.flink.metrics.MeterView;
-import org.apache.flink.types.Row;
+import org.apache.flink.table.runtime.types.CRow;
 
 
 /**
@@ -34,11 +34,11 @@
  * author: toutian
  * create: 2019/12/24
  */
-public class SerializationMetricWrapper implements SerializationSchema<Row> {
+public class SerializationMetricWrapper implements SerializationSchema<CRow> {
 
     private static final long serialVersionUID = 1L;
 
-    private SerializationSchema<Row> serializationSchema;
+    private SerializationSchema<CRow> serializationSchema;
 
     private transient RuntimeContext runtimeContext;
 
@@ -47,7 +47,7 @@ public class SerializationMetricWrapper implements SerializationSchema<Row> {
     protected transient Meter dtNumRecordsOutRate;
 
 
-    public SerializationMetricWrapper(SerializationSchema<Row> serializationSchema) {
+    public SerializationMetricWrapper(SerializationSchema<CRow> serializationSchema) {
         this.serializationSchema = serializationSchema;
     }
 
@@ -57,7 +57,7 @@ public void initMetric() {
     }
 
     @Override
-    public byte[] serialize(Row element) {
+    public byte[] serialize(CRow element) {
         beforeSerialize();
         byte[] row = serializationSchema.serialize(element);
         afterSerialize();
@@ -79,7 +79,7 @@ public void setRuntimeContext(RuntimeContext runtimeContext) {
         this.runtimeContext = runtimeContext;
     }
 
-    public SerializationSchema<Row> getSerializationSchema() {
+    public SerializationSchema<CRow> getSerializationSchema() {
         return serializationSchema;
     }
 

core/src/main/java/com/dtstack/flink/sql/option/OptionParser.java

@@ -26,12 +26,11 @@
 import org.apache.commons.lang.StringUtils;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Field;
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Map;
 import java.io.File;
-import java.io.FileInputStream;
 import java.net.URLEncoder;
-import java.util.stream.Stream;
 
 import org.apache.commons.codec.Charsets;
 import org.apache.flink.util.FileUtils;
@@ -104,8 +103,8 @@ public List<String> getProgramExeArgList() throws Exception {
                 continue;
             } else if (OPTION_SQL.equalsIgnoreCase(key)) {
                 File file = new File(value.toString());
-                String content = FileUtils.readFile(file, "UTF-8");
-                value = URLEncoder.encode(content, Charsets.UTF_8.name());
+                String content = FileUtils.readFile(file, StandardCharsets.UTF_8.name());
+                value = URLEncoder.encode(content, StandardCharsets.UTF_8.name());
             }
             args.add("-" + key);
             args.add(value.toString());

core/src/main/java/com/dtstack/flink/sql/parser/CreateTmpTableParser.java

@@ -22,7 +22,11 @@
 
 import com.dtstack.flink.sql.util.DtStringUtil;
 import org.apache.calcite.config.Lex;
-import org.apache.calcite.sql.*;
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlJoin;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlSelect;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
 import com.google.common.collect.Lists;

core/src/main/java/com/dtstack/flink/sql/parser/InsertSqlParser.java

@@ -21,7 +21,14 @@
 package com.dtstack.flink.sql.parser;
 
 import org.apache.calcite.config.Lex;
-import org.apache.calcite.sql.*;
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlInsert;
+import org.apache.calcite.sql.SqlJoin;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlMatchRecognize;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOrderBy;
+import org.apache.calcite.sql.SqlSelect;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
 import org.apache.commons.lang3.StringUtils;

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

@@ -57,11 +57,11 @@ public abstract class AbstractSideTableInfo extends AbstractTableInfo implements
 
     public static final String ASYNC_REQ_POOL_KEY = "asyncPoolSize";
 
-    private String cacheType = "none";//None or LRU or ALL
+    private String cacheType = "none";
 
     private int cacheSize = 10000;
 
-    private long cacheTimeout = 60 * 1000;//
+    private long cacheTimeout = 60_000L;
 
     private int  asyncCapacity=100;
 

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

@@ -47,6 +47,8 @@ public abstract class BaseAllReqRow extends RichFlatMapFunction<CRow, CRow> impl
 
     private static final Logger LOG = LoggerFactory.getLogger(BaseAllReqRow.class);
 
+    public static final long LOAD_DATA_ERROR_SLEEP_TIME = 5_000L;
+
     protected BaseSideInfo sideInfo;
 
     private ScheduledExecutorService es;

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

@@ -25,7 +25,6 @@
 import com.google.common.base.Strings;
 
 import java.io.Serializable;
-import java.util.Map;
 
 /**
  * Join信息

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

@@ -121,7 +121,7 @@ public void exec(String sql, Map<String, AbstractSideTableInfo> sideTableMap, St
         SideSQLParser sideSQLParser = new SideSQLParser();
         sideSQLParser.setLocalTableCache(localTableCache);
         Queue<Object> exeQueue = sideSQLParser.getExeQueue(sql, sideTableMap.keySet());
-        Object pollObj = null;
+        Object pollObj;
 
         //need clean
         boolean preIsSideJoin = false;