异常时将数据写入日志

Author: dengdejiushini

docs/elasticsearch6Sink.md

@@ -48,7 +48,7 @@ CREATE TABLE MyResult(
     aa INT,
     bb INT
  )WITH(
-    type ='elasticsearch',
+    type ='elasticsearch6',
     address ='172.16.10.47:9500',
     cluster='es_47_menghan',
     esType ='type1',

elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch6/CustomerSinkFunc.java

@@ -68,19 +68,19 @@ public CustomerSinkFunc(String index, String type, List<String> fieldNames, List
 
     @Override
     public void process(Tuple2 tuple2, RuntimeContext ctx, RequestIndexer indexer) {
-        try{
+        try {
             Tuple2<Boolean, Row> tupleTrans = tuple2;
             Boolean retract = tupleTrans.getField(0);
             Row element = tupleTrans.getField(1);
-            if(!retract){
+            if (!retract) {
                 return;
             }
 
-
             indexer.add(createIndexRequest(element));
             outRecords.inc();
-        }catch (Throwable e){
-            logger.error("", e);
+        } catch (Throwable e) {
+            logger.error("Failed to store source data {}. ", tuple2.getField(1));
+            logger.error("Failed to create index request exception. ", e);
         }
     }