Add option to use auto-generated IDs on indexing (#803)

Author: srishti-saraswat

src/main/java/io/confluent/connect/elasticsearch/DataConverter.java

@@ -181,10 +181,12 @@ public DocWriteRequest<?> convertRecord(SinkRecord record, String index) {
             .retryOnConflict(Math.min(config.maxInFlightRequests(), 5));
       case INSERT:
         OpType opType = config.isDataStream() ? OpType.CREATE : OpType.INDEX;
-        return maybeAddExternalVersioning(
-            new IndexRequest(index).id(id).source(payload, XContentType.JSON).opType(opType),
-            record
-        );
+        IndexRequest req =
+                new IndexRequest(index).source(payload, XContentType.JSON).opType(opType);
+        if (config.useAutogeneratedIds()) {
+          return req;
+        }
+        return maybeAddExternalVersioning(req.id(id), record);
       default:
         return null; // shouldn't happen
     }

src/main/java/io/confluent/connect/elasticsearch/ElasticsearchSinkConnectorConfig.java

@@ -232,6 +232,9 @@ public class ElasticsearchSinkConnectorConfig extends AbstractConfig {
   private static final String DROP_INVALID_MESSAGE_DISPLAY = "Drop invalid messages";
   private static final boolean DROP_INVALID_MESSAGE_DEFAULT = false;
 
+  private static final String USE_AUTOGENERATED_IDS_DISPLAY = "Elasticsearch Generated IDs";
+  private static final boolean USE_AUTOGENERATED_IDS_DEFAULT = false;
+
   public static final String BEHAVIOR_ON_NULL_VALUES_CONFIG = "behavior.on.null.values";
   private static final String BEHAVIOR_ON_NULL_VALUES_DOC = "How to handle records with a "
       + "non-null key and a null value (i.e. Kafka tombstone records). Valid options are "
@@ -358,7 +361,17 @@ public class ElasticsearchSinkConnectorConfig extends AbstractConfig {
           + DATA_STREAM_NAMESPACE_CONFIG + "``}.";
   private static final String DATA_STREAM_DATASET_DISPLAY = "Data Stream Dataset";
   private static final String DATA_STREAM_DATASET_DEFAULT = "";
-  
+  public static final String USE_AUTO_GENERATED_IDS_CONFIG = "use.autogenerated.ids";
+  private static final String USE_AUTO_GENERATED_IDS_DOC = String.format(
+          "Whether to use auto-generated Elasticsearch document IDs for insertion requests. "
+                  + "Note that this setting removes exactly once guarantees and message "
+                  + "delivery will be at least once. Only applies if %s is set to %s."
+                  + "When this is set to ``true``, ``%s`` option will also be ignored when "
+                  + "sending data to Elasticsearch",
+          WRITE_METHOD_CONFIG,
+          WriteMethod.INSERT,
+          IGNORE_KEY_CONFIG
+  );
   public static final String DATA_STREAM_TYPE_CONFIG = "data.stream.type";
   private static final String DATA_STREAM_TYPE_DOC = String.format(
       "Generic type describing the data to be written to data stream. "
@@ -736,6 +749,16 @@ private static void addConversionConfigs(ConfigDef configDef) {
             Width.SHORT,
             WRITE_METHOD_DISPLAY,
             new EnumRecommender<>(WriteMethod.class)
+        ).define(
+            USE_AUTO_GENERATED_IDS_CONFIG,
+            Type.BOOLEAN,
+            USE_AUTOGENERATED_IDS_DEFAULT,
+            Importance.MEDIUM,
+            USE_AUTO_GENERATED_IDS_DOC,
+            CONNECTOR_GROUP,
+            ++order,
+            Width.SHORT,
+            USE_AUTOGENERATED_IDS_DISPLAY
     );
   }
 
@@ -923,7 +946,8 @@ public boolean shouldDisableHostnameVerification() {
   }
 
   public boolean shouldIgnoreKey(String topic) {
-    return ignoreKey() || ignoreKeyTopics().contains(topic);
+    return ignoreKey() || ignoreKeyTopics().contains(topic)
+            || (writeMethod().equals(WriteMethod.INSERT) && useAutogeneratedIds());
   }
 
   public boolean shouldIgnoreSchema(String topic) {
@@ -1010,6 +1034,10 @@ public Set<String> ignoreSchemaTopics() {
     return new HashSet<>(getList(IGNORE_SCHEMA_TOPICS_CONFIG));
   }
 
+  public boolean useAutogeneratedIds() {
+    return getBoolean(USE_AUTO_GENERATED_IDS_CONFIG);
+  }
+
   public String kerberosUserPrincipal() {
     return getString(KERBEROS_PRINCIPAL_CONFIG);
   }

src/test/java/io/confluent/connect/elasticsearch/DataConverterTest.java

@@ -20,6 +20,7 @@
 import org.apache.kafka.connect.data.*;
 import org.apache.kafka.connect.errors.DataException;
 import org.apache.kafka.connect.sink.SinkRecord;
+import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.index.VersionType;
@@ -39,6 +40,7 @@
 import static io.confluent.connect.elasticsearch.DataConverter.TIMESTAMP_FIELD;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.fail;
@@ -435,6 +437,30 @@ public void failOnNullValue() {
     }
   }
 
+  @Test
+  public void useAutogeneratedIds() {
+    props.put(ElasticsearchSinkConnectorConfig.USE_AUTO_GENERATED_IDS_CONFIG, "true");
+    props.put(ElasticsearchSinkConnectorConfig.WRITE_METHOD_CONFIG, "INSERT");
+
+    converter = new DataConverter(new ElasticsearchSinkConnectorConfig(props));
+
+    SinkRecord sinkRecord = createSinkRecordWithValue(new Struct(schema).put("string","test"));
+    DocWriteRequest<?> req = converter.convertRecord(sinkRecord, index);
+    assertNull(req.id());
+  }
+
+  @Test
+  public void ignoreAutogeneratedIdsForUpsert() {
+    props.put(ElasticsearchSinkConnectorConfig.USE_AUTO_GENERATED_IDS_CONFIG, "true");
+    props.put(ElasticsearchSinkConnectorConfig.WRITE_METHOD_CONFIG, "UPSERT");
+
+    converter = new DataConverter(new ElasticsearchSinkConnectorConfig(props));
+
+    SinkRecord sinkRecord = createSinkRecordWithValue(new Struct(schema).put("string","test"));
+    DocWriteRequest<?> req = converter.convertRecord(sinkRecord, index);
+    assertNotNull(req.id());
+  }
+
   public SinkRecord createSinkRecordWithValue(Object value) {
     return new SinkRecord(
          topic,