Merge branch '1.8_test_3.10.x' of ssh://git.dtstack.cn:10022/dt-insight-engine/flinkStreamSQL into test_1.8_3.1.0x_elasticsearch6-sink

Author: dengdejiushini

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

@@ -219,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);
             }
 

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>
+        <logback.version>1.1.7</logback.version>
     </properties>
 
     <dependencies>
@@ -121,10 +122,11 @@
             <artifactId>junit</artifactId>
             <version>4.12</version>
         </dependency>
+
         <dependency>
             <groupId>ch.qos.logback</groupId>
             <artifactId>logback-classic</artifactId>
-            <version>1.1.7</version>
+            <version>${logback.version}</version>
         </dependency>
 
     </dependencies>

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

@@ -27,8 +27,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import ch.qos.logback.classic.Level;
-import ch.qos.logback.classic.LoggerContext;
 
 /**
  * Date: 2018/6/26
@@ -45,10 +43,4 @@ public static void main(String[] args) throws Exception {
         env.execute(paramsInfo.getName());
         LOG.info("program {} execution success", paramsInfo.getName());
     }
-    private static void setLogLevel(String level){
-        LoggerContext loggerContext= (LoggerContext) LoggerFactory.getILoggerFactory();
-        //设置全局日志级别
-        ch.qos.logback.classic.Logger logger = loggerContext.getLogger("root");
-        logger.setLevel(Level.toLevel(level, Level.INFO));
-    }
 }

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduOutputFormat.java

@@ -23,13 +23,7 @@
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.types.Row;
-import org.apache.kudu.client.AsyncKuduClient;
-import org.apache.kudu.client.AsyncKuduSession;
-import org.apache.kudu.client.KuduClient;
-import org.apache.kudu.client.KuduException;
-import org.apache.kudu.client.KuduTable;
-import org.apache.kudu.client.Operation;
-import org.apache.kudu.client.PartialRow;
+import org.apache.kudu.client.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -123,7 +117,7 @@ public void writeRecord(Tuple2 record) throws IOException {
         Row row = tupleTrans.getField(1);
         if (row.getArity() != fieldNames.length) {
             if(outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0) {
-                LOG.error("record insert failed ..{}", row.toString());
+                LOG.error("record insert failed:{}", row.toString());
                 LOG.error("cause by row.getArity() != fieldNames.length");
             }
             outDirtyRecords.inc();
@@ -142,7 +136,7 @@ public void writeRecord(Tuple2 record) throws IOException {
             outRecords.inc();
         } catch (KuduException 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 record:{} current row:{}", outDirtyRecords.getCount(), row.toString());
                 LOG.error("", e);
             }
             outDirtyRecords.inc();