UPSERT mode for DELETE operation not send external version

confluentinc · Sep 11, 2023 · c60bd1d · c60bd1d
1 parent e4629e6
commit c60bd1d
Showing 1 changed file with 7 additions and 1 deletion.
diff --git a/src/main/java/io/confluent/connect/elasticsearch/DataConverter.java b/src/main/java/io/confluent/connect/elasticsearch/DataConverter.java
@@ -52,6 +52,8 @@
 import java.util.List;
 import java.util.Map;
 
+import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.WriteMethod.UPSERT;
+
 public class DataConverter {
 
   private static final Logger log = LoggerFactory.getLogger(DataConverter.class);
@@ -161,7 +163,11 @@ public DocWriteRequest<?> convertRecord(SinkRecord record, String index) {
 
     // delete
     if (record.value() == null) {
-      return maybeAddExternalVersioning(new DeleteRequest(index).id(id), record);
+      if (config.writeMethod().name().equals(UPSERT.name())) {
+        return new DeleteRequest(index).id(id);
+      } else {
+        return maybeAddExternalVersioning(new DeleteRequest(index).id(id), record);
+      }
     }
 
     String payload = getPayload(record);