Merge remote-tracking branch 'origin/1.8_release_3.9.x' into hotfix_3.9.4_23229

# Conflicts:
#	launcher/src/main/java/com/dtstack/flink/sql/launcher/perjob/PerJobClusterClientBuilder.java
#	rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java

Author: todd5167

README.md

@@ -48,6 +48,18 @@
 * Java: JDK8及以上
 * Flink集群: 1.4,1.5,1.8（单机模式不需要安装Flink集群）
 * 操作系统：理论上不限
+* kerberos环境需要在flink-conf.yaml配置security.kerberos.login.keytab以及security.kerberos.login.principal参数，配置案例:
+```
+## hadoop配置文件路径
+fs.hdfs.hadoopconf: /Users/maqi/tmp/hadoopconf/hadoop_250
+security.kerberos.login.use-ticket-cache: true
+security.kerberos.login.keytab: /Users/maqi/tmp/hadoopconf/hadoop_250/maqi.keytab
+security.kerberos.login.principal: maqi@DTSTACK.COM
+security.kerberos.login.contexts: Client,KafkaClient
+zookeeper.sasl.service-name: zookeeper
+zookeeper.sasl.login-context-name: Client
+
+```
 
 ### 1.3 打包
 
@@ -56,9 +68,16 @@
 ```
 mvn clean package -Dmaven.test.skip
 
-打包结束后，项目根目录下会产生plugins目录，plugins目录下存放编译好的数据同步插件包,在lib目下存放job提交的包
 ```
 
+打包完成后的包结构：
+
+> * dt-center-flinkStreamSQL  
+>  >  * bin: 任务启动脚本
+>  >  * lib: launcher包存储路径，是任务提交的入口
+>  >  * plugins:  插件包存储路径
+>  >  * ........   :  core及插件代码
+
 ### 1.4 启动
 
 #### 1.4.1 启动命令

cassandra/cassandra-side/cassandra-all-side/pom.xml

@@ -36,6 +36,7 @@
                             <goal>shade</goal>
                         </goals>
                         <configuration>
+                            <createDependencyReducedPom>false</createDependencyReducedPom>
                             <artifactSet>
                                 <excludes>
                                     <exclude>org.slf4j</exclude>

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

@@ -268,7 +268,7 @@ private void loadData(Map<String, List<Map<String, Object>>> tmpCache) throws SQ
                         LOG.warn("get conn fail, wait for 5 sec and try again, connInfo:" + connInfo);
                         Thread.sleep(5 * 1000);
                     } catch (InterruptedException e1) {
-                        e1.printStackTrace();
+                        LOG.error("", e1);
                     }
                 }
 

cassandra/cassandra-side/cassandra-async-side/pom.xml

@@ -52,6 +52,7 @@
                             <goal>shade</goal>
                         </goals>
                         <configuration>
+                            <createDependencyReducedPom>false</createDependencyReducedPom>
                             <artifactSet>
                                 <excludes>
                                     <exclude>org.slf4j</exclude>

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

@@ -161,16 +161,16 @@ private void connCassandraDB(CassandraSideTableInfo tableInfo) {
 
     @Override
     public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Exception {
-
+        Row inputRow = Row.copy(input);
         JsonArray inputParams = new JsonArray();
         StringBuffer stringBuffer = new StringBuffer();
         String sqlWhere = " where ";
 
         for (int i = 0; i < sideInfo.getEqualFieldList().size(); i++) {
             Integer conValIndex = sideInfo.getEqualValIndex().get(i);
-            Object equalObj = input.getField(conValIndex);
+            Object equalObj = inputRow.getField(conValIndex);
             if (equalObj == null) {
-                dealMissKey(input, resultFuture);
+                dealMissKey(inputRow, resultFuture);
                 return;
             }
             inputParams.add(equalObj);
@@ -194,12 +194,12 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
             if (val != null) {
 
                 if (ECacheContentType.MissVal == val.getType()) {
-                    dealMissKey(input, resultFuture);
+                    dealMissKey(inputRow, resultFuture);
                     return;
                 } else if (ECacheContentType.MultiLine == val.getType()) {
                     List<Row> rowList = Lists.newArrayList();
                     for (Object jsonArray : (List) val.getContent()) {
-                        Row row = fillData(input, jsonArray);
+                        Row row = fillData(inputRow, jsonArray);
                         rowList.add(row);
                     }
                     resultFuture.complete(rowList);
@@ -240,7 +240,7 @@ public void onSuccess(List<com.datastax.driver.core.Row> rows) {
                     List<com.datastax.driver.core.Row> cacheContent = Lists.newArrayList();
                     List<Row> rowList = Lists.newArrayList();
                     for (com.datastax.driver.core.Row line : rows) {
-                        Row row = fillData(input, line);
+                        Row row = fillData(inputRow, line);
                         if (openCache()) {
                             cacheContent.add(line);
                         }
@@ -251,7 +251,7 @@ public void onSuccess(List<com.datastax.driver.core.Row> rows) {
                         putCache(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, cacheContent));
                     }
                 } else {
-                    dealMissKey(input, resultFuture);
+                    dealMissKey(inputRow, resultFuture);
                     if (openCache()) {
                         putCache(key, CacheMissVal.getMissKeyObj());
                     }

cassandra/cassandra-side/cassandra-side-core/src/main/java/com/dtstack/flink/sql/side/cassandra/table/CassandraSideParser.java

@@ -68,9 +68,8 @@ public class CassandraSideParser extends AbsSideTableParser {
 
     public static final String POOL_TIMEOUT_MILLIS_KEY = "poolTimeoutMillis";
 
-    static {
-        keyPatternMap.put(SIDE_SIGN_KEY, SIDE_TABLE_SIGN);
-        keyHandlerMap.put(SIDE_SIGN_KEY, CassandraSideParser::dealSideSign);
+    public CassandraSideParser() {
+        addParserHandler(SIDE_SIGN_KEY, SIDE_TABLE_SIGN, this::dealSideSign);
     }
 
     @Override
@@ -97,7 +96,7 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         return cassandraSideTableInfo;
     }
 
-    private static void dealSideSign(Matcher matcher, TableInfo tableInfo) {
+    private void dealSideSign(Matcher matcher, TableInfo tableInfo) {
     }
 
     public Class dbTypeConvertToJavaType(String fieldType) {

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

@@ -48,7 +48,7 @@
 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.sink.MetricOutputFormat;
+import com.dtstack.flink.sql.outputformat.DtRichOutputFormat;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -69,7 +69,7 @@
  * @see Tuple
  * @see DriverManager
  */
-public class CassandraOutputFormat extends MetricOutputFormat {
+public class CassandraOutputFormat extends DtRichOutputFormat {
     private static final long serialVersionUID = -7994311331389155692L;
 
     private static final Logger LOG = LoggerFactory.getLogger(CassandraOutputFormat.class);
@@ -193,7 +193,6 @@ public void writeRecord(Tuple2 tuple2) throws IOException {
         try {
             if (retract) {
                 insertWrite(row);
-                outRecords.inc();
             } else {
                 //do nothing
             }
@@ -204,14 +203,24 @@ public void writeRecord(Tuple2 tuple2) throws IOException {
 
     private void insertWrite(Row row) {
         try {
+
+            if(outRecords.getCount() % ROW_PRINT_FREQUENCY == 0){
+                LOG.info("Receive data : {}", row);
+            }
+
             String cql = buildSql(row);
             if (cql != null) {
                 ResultSet resultSet = session.execute(cql);
                 resultSet.wasApplied();
+                outRecords.inc();
             }
         } catch (Exception e) {
+            if(outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0){
+                LOG.error("record insert failed, total dirty num:{}, current record:{}", outDirtyRecords.getCount(), row.toString());
+                LOG.error("", e);
+            }
+
             outDirtyRecords.inc();
-            LOG.error("[upsert] is error:" + e.getMessage());
         }
     }
 

console/console-sink/src/main/java/com/dtstack/flink/sql/sink/console/ConsoleOutputFormat.java

@@ -18,7 +18,7 @@
 
 package com.dtstack.flink.sql.sink.console;
 
-import com.dtstack.flink.sql.sink.MetricOutputFormat;
+import com.dtstack.flink.sql.outputformat.DtRichOutputFormat;
 import com.dtstack.flink.sql.sink.console.table.TablePrintUtil;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -37,7 +37,7 @@
  *
  * @author xuqianjin
  */
-public class ConsoleOutputFormat extends MetricOutputFormat {
+public class ConsoleOutputFormat extends DtRichOutputFormat {
 
     private static final Logger LOG = LoggerFactory.getLogger(ConsoleOutputFormat.class);
 
@@ -69,7 +69,11 @@ public void writeRecord(Tuple2 tuple2) throws IOException {
 
         List<String[]> data = new ArrayList<>();
         data.add(fieldNames);
-        data.add(record.toString().split(","));
+        String[] recordStr = new String[record.getArity()];
+        for (int i=0; i < record.getArity(); i++) {
+            recordStr[i] = (String.valueOf(record.getField(i)));
+        }
+        data.add(recordStr);
         TablePrintUtil.build(data).print();
 
         outRecords.inc();