feat: add ignore version configuration

fix: add logs

fix: logs

Author: Mahanmmi

pom.xml

@@ -327,7 +327,7 @@
 
                                 Mapping is the process of defining how a document, and the fields it contains, are stored and indexed. Users can explicitly define mappings for types in indices. When a mapping is not explicitly defined, Elasticsearch can determine field names and types from data, however, some types such as timestamp and decimal, may not be correctly inferred. To ensure that the types are correctly inferred, the connector provides a feature to infer a mapping from the schemas of Kafka messages.
                             </description>
-                            <logo>logos/elasticsearch.jpg</logo>
+                            <logo>logos/elasticlogo.png</logo>
 
                             <supportProviderName>Confluent, Inc.</supportProviderName>
                             <supportSummary><![CDATA[This connector is <a href="https://www.confluent.io/subscription/">supported by Confluent</a> as part of a <a href="https://www.confluent.io/product/confluent-platform/">Confluent Platform</a> subscription.]]></supportSummary>

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

@@ -247,9 +247,15 @@ private DocWriteRequest<?> maybeAddExternalVersioning(
       DocWriteRequest<?> request,
       SinkRecord record
   ) {
-    if (!config.isDataStream() && !config.shouldIgnoreKey(record.topic())) {
+    if (!config.shouldIgnoreVersion()
+        && !config.isDataStream()
+        && !config.shouldIgnoreKey(record.topic())) {
+      log.debug("Adding external versioning to {} with offset {}, the ignore setting is {}",
+          recordString(record), record.kafkaOffset(), config.shouldIgnoreVersion());
       request.versionType(VersionType.EXTERNAL);
       request.version(record.kafkaOffset());
+    } else {
+      log.trace("Not adding external versioning to {}", recordString(record));
     }
 
     return request;

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

@@ -278,6 +278,15 @@ public class ElasticsearchSinkConnectorConfig extends AbstractConfig {
           + "is acceptable and is necessary for troubleshooting.";
   private static final Boolean LOG_SENSITIVE_DATA_DEFAULT = Boolean.FALSE;
 
+  public static final String IGNORE_VERSION_CONFIG = "version.ignore";
+  private static final String IGNORE_VERSION_DOC =
+      "Whether to use kafka record offset as Elasticsearch document EXTERNAL version."
+          + " When this is set to ``true``, kafka records with offset lower than previously"
+          + " indexed will overwrite the document. When set to ``false``, the kafka record with"
+          + " offset lower than one previously indexed will be ignored.";
+  private static final String IGNORE_VERSION_DISPLAY = "Ignore Version mode";
+  private static final boolean IGNORE_VERSION_DEFAULT = Boolean.FALSE;
+
   // Proxy group
   public static final String PROXY_HOST_CONFIG = "proxy.host";
   private static final String PROXY_HOST_DISPLAY = "Proxy Host";
@@ -702,6 +711,16 @@ private static void addConversionConfigs(ConfigDef configDef) {
             Width.SHORT,
             WRITE_METHOD_DISPLAY,
             new EnumRecommender<>(WriteMethod.class)
+        ).define(
+            IGNORE_VERSION_CONFIG,
+            Type.BOOLEAN,
+            IGNORE_VERSION_DEFAULT,
+            Importance.LOW,
+            IGNORE_VERSION_DOC,
+            DATA_CONVERSION_GROUP,
+            ++order,
+            Width.SHORT,
+            IGNORE_VERSION_DISPLAY
     );
   }
 
@@ -1053,6 +1072,10 @@ public WriteMethod writeMethod() {
     return WriteMethod.valueOf(getString(WRITE_METHOD_CONFIG).toUpperCase());
   }
 
+  public boolean shouldIgnoreVersion() {
+    return getBoolean(IGNORE_VERSION_CONFIG);
+  }
+
   private static class DataStreamDatasetValidator implements Validator {
 
     @Override

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

@@ -32,6 +32,7 @@
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.SECURITY_PROTOCOL_CONFIG;
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.SSL_CONFIG_PREFIX;
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.WRITE_METHOD_CONFIG;
+import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.IGNORE_VERSION_CONFIG;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThrows;
@@ -599,6 +600,42 @@ private List<SinkRecord> causeExternalVersionConflictError(ElasticsearchClient c
 
     return conflict_list;
   }
+  
+  /**
+   * VERSION_IGNORE_CONFIG is set to true, then Version.EXTERNAL is not used
+   * even if the IGNORE_KEY_CONFIG is said to false.
+   *
+   * In this case, even if records
+   * are written with a kafka offset, less than the previous indexed record and no
+   * exception should be thrown.
+   * 
+   * @throws Exception will be thrown if the test fails
+   */
+  @Test
+  public void testInternalVersionUsedForVersionIgnore() throws Exception {
+    props.put(IGNORE_VERSION_CONFIG, "true");
+    props.put(IGNORE_KEY_CONFIG, "false");
+    config = new ElasticsearchSinkConnectorConfig(props);
+    converter = new DataConverter(config);
+
+    ErrantRecordReporter reporter = mock(ErrantRecordReporter.class);
+    ElasticsearchClient client = new ElasticsearchClient(config, reporter, () -> offsetTracker.updateOffsets()) {
+      protected boolean handleResponse(BulkItemResponse response, DocWriteRequest<?> request,
+          long executionId) {
+        // Check if request version type is INTERNAL
+        assertEquals(request.versionType(), VersionType.INTERNAL);
+        return super.handleResponse(response, request, executionId);
+      }
+    };
+
+    List<SinkRecord> duplicate_records = causeExternalVersionConflictError(client);
+
+    // Make sure that no error was reported for any record(s)
+    for (SinkRecord duplicated_record : duplicate_records) {
+      verify(reporter, never()).report(eq(duplicated_record), any(Throwable.class));
+    }
+    client.close();
+  }
 
   /**
    * If the record version is set to VersionType.EXTERNAL (normal case for non-streaming),