Merge branch 'feat_1.8_esnullcheck' into '1.8_test_3.10.x'

es id null check



See merge request !279

Author: todd5167

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

@@ -96,11 +96,14 @@ public void setOutRecords(Counter outRecords) {
     }
 
     private IndexRequest createIndexRequest(Row element) {
-        // index start at 1,
-        String idFieldStr = idFieldIndexList.stream()
-                .filter(index -> index > 0 && index <= element.getArity())
-                .map(index -> element.getField(index - 1).toString())
-                .collect(Collectors.joining(ID_VALUE_SPLIT));
+        String idFieldStr = "";
+        if (null != idFieldIndexList) {
+            // index start at 1,
+            idFieldStr = idFieldIndexList.stream()
+                    .filter(index -> index > 0 && index <= element.getArity())
+                    .map(index -> element.getField(index - 1).toString())
+                    .collect(Collectors.joining(ID_VALUE_SPLIT));
+        }
 
         Map<String, Object> dataMap = EsUtil.rowToJsonMap(element,fieldNames,fieldTypes);
         int length = Math.min(element.getArity(), fieldNames.size());

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

@@ -95,11 +95,14 @@ public void setOutDirtyRecords(Counter outDirtyRecords) {
     }
 
     private IndexRequest createIndexRequest(Row element) {
-        // index start at 1,
-        String idFieldStr = idFieldIndexList.stream()
-                .filter(index -> index > 0 && index <= element.getArity())
-                .map(index -> element.getField(index - 1).toString())
-                .collect(Collectors.joining(ID_VALUE_SPLIT));
+        String idFieldStr = "";
+        if (null != idFieldIndexList) {
+            // index start at 1,
+            idFieldStr = idFieldIndexList.stream()
+                    .filter(index -> index > 0 && index <= element.getArity())
+                    .map(index -> element.getField(index - 1).toString())
+                    .collect(Collectors.joining(ID_VALUE_SPLIT));
+        }
 
         Map<String, Object> dataMap = Es6Util.rowToJsonMap(element, fieldNames, fieldTypes);
         int length = Math.min(element.getArity(), fieldNames.size());