Merge branch 'feat_1.8_elasticsearch6-sink' into 'v1.8.0_dev'

添加对es6 结果表的支持（es添加脏数据指标的输出）

es添加脏数据指标的输出

See merge request !233

Author: zoudaokoulife

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

@@ -53,7 +53,9 @@ public class CustomerSinkFunc implements ElasticsearchSinkFunction<Tuple2> {
 
     private List<String> fieldTypes;
 
-    public transient Counter outRecords;
+    private transient Counter outRecords;
+
+    private transient Counter outDirtyRecords;
 
     /** 默认分隔符为'_' */
     private char sp = '_';
@@ -79,6 +81,7 @@ public void process(Tuple2 tuple2, RuntimeContext ctx, RequestIndexer indexer) {
             indexer.add(createIndexRequest(element));
             outRecords.inc();
         } catch (Throwable e) {
+            outDirtyRecords.inc();
             logger.error("Failed to store source data {}. ", tuple2.getField(1));
             logger.error("Failed to create index request exception. ", e);
         }
@@ -88,6 +91,10 @@ public void setOutRecords(Counter outRecords) {
         this.outRecords = outRecords;
     }
 
+    public void setOutDirtyRecords(Counter outDirtyRecords) {
+        this.outDirtyRecords = outDirtyRecords;
+    }
+
     private IndexRequest createIndexRequest(Row element) {
 
         List<String> idFieldList = new ArrayList<>();

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

@@ -63,7 +63,10 @@ public void open(Configuration parameters) throws Exception {
 
     public void initMetric() {
         Counter counter = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_RECORDS_OUT);
+        Counter outDirtyRecords = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_DIRTY_RECORDS_OUT);
+
         customerSinkFunc.setOutRecords(counter);
+        customerSinkFunc.setOutDirtyRecords(outDirtyRecords);
         outRecordsRate = getRuntimeContext().getMetricGroup().meter(MetricConstant.DT_NUM_RECORDS_OUT_RATE, new MeterView(counter, 20));
     }
 }