Skip to content

Commit dbc9211

Browse files
committed
[feat-269][elasticsearch5&6][sink]ignore check when ids is null.
1 parent 70f68ad commit dbc9211

File tree

2 files changed

+6
-2
lines changed
  • elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch
  • elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch

2 files changed

+6
-2
lines changed

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -119,7 +119,9 @@ public TypeInformation<?>[] getFieldTypes() {
119119
private RichSinkFunction createEsSinkFunction(){
120120

121121
// check whether id fields is exists in columns if not use position to generate doc's id
122-
if (!usePosition) {
122+
if (!usePosition
123+
&& ids != null
124+
&& ids.size() != 0) {
123125
List<String> filedNamesLists = Arrays.asList(fieldNames);
124126
Preconditions.checkState(filedNamesLists.containsAll(ids), "elasticsearch6 type of id %s is should be exists in columns %s.", ids, filedNamesLists);
125127
}

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -111,7 +111,9 @@ public TypeInformation<?>[] getFieldTypes() {
111111
private RichSinkFunction createEsSinkFunction() {
112112

113113
// check whether id fields is exists in columns if not use position to generate doc's id
114-
if (!usePosition) {
114+
if (!usePosition
115+
&& ids != null
116+
&& ids.size() != 0) {
115117
List<String> filedNamesLists = Arrays.asList(fieldNames);
116118
Preconditions.checkState(filedNamesLists.containsAll(ids), "elasticsearch6 type of id %s is should be exists in columns %s.", ids, filedNamesLists);
117119
}

0 commit comments

Comments
 (0)