feat: migrate from HLRC to elasticsearch-java 8.x client

Author: yeikel

checkstyle/suppressions.xml

@@ -8,7 +8,7 @@
 
     <!-- switch statements on types exceed maximum complexity -->
     <suppress
-      checks="(CyclomaticComplexity)"
+      checks="(CyclomaticComplexity|NPathComplexity)"
       files="(Mapping|DataConverter).java"
     />
 

pom.xml

@@ -37,7 +37,7 @@
     </scm>
 
     <properties>
-        <es.version>7.17.24</es.version>
+        <elasticsearch.java.version>8.19.15</elasticsearch.java.version>
         <lz4.version>1.10.1</lz4.version>
         <hamcrest.version>1.3</hamcrest.version>
         <gson.version>2.9.0</gson.version>
@@ -84,23 +84,8 @@
         </dependency>
         <dependency>
             <groupId>org.elasticsearch.client</groupId>
-            <artifactId>elasticsearch-rest-high-level-client</artifactId>
-            <version>${es.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>org.elasticsearch</groupId>
-            <artifactId>elasticsearch</artifactId>
-            <version>${es.version}</version>
-            <exclusions>
-                <exclusion>
-                    <groupId>org.yaml</groupId>
-                    <artifactId>snakeyaml</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>org.lz4</groupId>
-                    <artifactId>lz4-java</artifactId>
-                </exclusion>
-            </exclusions>
+            <artifactId>elasticsearch-rest-client</artifactId>
+            <version>${elasticsearch.java.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.logging.log4j</groupId>
@@ -115,7 +100,7 @@
         <dependency>
             <groupId>co.elastic.clients</groupId>
             <artifactId>elasticsearch-java</artifactId>
-            <version>8.2.2</version>
+            <version>${elasticsearch.java.version}</version>
         </dependency>
         <dependency>
             <groupId>org.eclipse.parsson</groupId>
@@ -508,7 +493,7 @@
                             </tags>
 
                             <requirements>
-                                <requirement>Elasticsearch 7.x</requirement>
+                                <requirement>Elasticsearch 8.x / 9.x</requirement>
                             </requirements>
 
                             <deliveryGuarantee>

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

@@ -37,13 +37,10 @@
 import org.apache.kafka.connect.storage.Converter;
 import org.apache.kafka.connect.storage.HeaderConverter;
 import org.apache.kafka.connect.storage.SimpleHeaderConverter;
-import org.elasticsearch.action.DocWriteRequest;
-import org.elasticsearch.action.DocWriteRequest.OpType;
-import org.elasticsearch.action.delete.DeleteRequest;
-import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.action.update.UpdateRequest;
-import org.elasticsearch.index.VersionType;
-import org.elasticsearch.xcontent.XContentType;
+import co.elastic.clients.elasticsearch._types.VersionType;
+import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
+import co.elastic.clients.json.JsonData;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -121,7 +118,7 @@ private String convertKey(Schema keySchema, Object key) {
     }
   }
 
-  public DocWriteRequest<?> convertRecord(SinkRecord record, String resourceName) {
+  public BulkOperation convertRecord(SinkRecord record, String resourceName) {
     if (record.value() == null) {
       switch (config.behaviorOnNullValues()) {
         case IGNORE:
@@ -166,32 +163,91 @@ public DocWriteRequest<?> convertRecord(SinkRecord record, String resourceName)
 
     // delete
     if (record.value() == null) {
-      return maybeAddExternalVersioning(new DeleteRequest(resourceName).id(id), record);
+      if (!config.isDataStream() && !config.shouldIgnoreKey(record.topic())) {
+        final long externalVersion = resolveExternalVersion(record);
+        return BulkOperation.of(b -> b.delete(d -> d
+            .index(resourceName).id(id)
+            .versionType(VersionType.External).version(externalVersion)));
+      }
+      return BulkOperation.of(b -> b.delete(d -> d.index(resourceName).id(id)));
     }
 
     String payload = getPayload(record);
     payload = maybeAddTimestamp(payload, record.timestamp());
+    final String finalPayload = payload;
 
     // index
     switch (config.writeMethod()) {
       case UPSERT:
-        return new UpdateRequest(resourceName, id)
-            .doc(payload, XContentType.JSON)
-            .upsert(payload, XContentType.JSON)
-            .retryOnConflict(Math.min(config.maxInFlightRequests(), 5));
+        return BulkOperation.of(b -> b.update(u -> u
+            .index(resourceName)
+            .id(id)
+            .retryOnConflict(Math.min(config.maxInFlightRequests(), 5))
+            .action(a -> a
+                .doc(JsonData.fromJson(finalPayload))
+                .upsert(JsonData.fromJson(finalPayload))
+            )
+        ));
       case INSERT:
-        OpType opType = config.isDataStream() ? OpType.CREATE : OpType.INDEX;
-        IndexRequest req =
-                new IndexRequest(resourceName).source(payload, XContentType.JSON).opType(opType);
+        if (config.isDataStream()) {
+          if (config.useAutogeneratedIds()) {
+            return BulkOperation.of(b -> b.create(c -> c
+                .index(resourceName)
+                .document(JsonData.fromJson(finalPayload))
+            ));
+          }
+          return BulkOperation.of(b -> b.create(c -> c
+              .index(resourceName).id(id)
+              .document(JsonData.fromJson(finalPayload))
+          ));
+        }
         if (config.useAutogeneratedIds()) {
-          return req;
+          return BulkOperation.of(b -> b.index(i -> i
+              .index(resourceName)
+              .document(JsonData.fromJson(finalPayload))
+          ));
         }
-        return maybeAddExternalVersioning(req.id(id), record);
+        if (!config.shouldIgnoreKey(record.topic())) {
+          final long externalVersion = resolveExternalVersion(record);
+          return BulkOperation.of(b -> b.index(i -> i
+              .index(resourceName).id(id)
+              .document(JsonData.fromJson(finalPayload))
+              .versionType(VersionType.External).version(externalVersion)));
+        }
+        return BulkOperation.of(b -> b.index(i -> i
+            .index(resourceName).id(id)
+            .document(JsonData.fromJson(finalPayload))
+        ));
       default:
         return null; // shouldn't happen
     }
   }
 
+  /**
+   * In many cases, we explicitly set the record version using the topic's offset.
+   * This version will, in turn, be checked by Elasticsearch and will throw a versioning
+   * error if the request represents an equivalent or older version of the record.
+   *
+   * @param record the record to be processed
+   * @return the version to use for the Elasticsearch request
+   */
+  private long resolveExternalVersion(SinkRecord record) {
+    if (config.hasExternalVersionHeader()) {
+      final Header versionHeader = record.headers().lastWithName(config.externalVersionHeader());
+      final byte[] versionValue = HEADER_CONVERTER.fromConnectHeader(
+          record.topic(), versionHeader.key(), versionHeader.schema(), versionHeader.value()
+      );
+      try {
+        //fromConnectHeader byte output is UTF_8
+        return Long.parseLong(new String(versionValue, StandardCharsets.UTF_8));
+      } catch (NumberFormatException e) {
+        throw new ConnectException("Error converting to long: "
+            + new String(versionValue, StandardCharsets.UTF_8), e);
+      }
+    }
+    return record.kafkaOffset();
+  }
+
   private String getPayload(SinkRecord record) {
     if (record.value() == null) {
       return null;
@@ -241,44 +297,6 @@ private String maybeAddTimestamp(String payload, Long timestamp) {
     return payload;
   }
 
-  /**
-   * In many cases, we explicitly set the record version using the topic's offset.
-   * This version will, in turn, be checked by Elasticsearch and will throw a versioning
-   * error if the request represents an equivalent or older version of the record.
-   *
-   * @param request the request currently being constructed for `record`
-   * @param record the record to be processed
-   * @return the (possibly modified) request which was passed in
-   */
-  private DocWriteRequest<?> maybeAddExternalVersioning(
-      DocWriteRequest<?> request,
-      SinkRecord record
-  ) {
-    if (!config.isDataStream() && !config.shouldIgnoreKey(record.topic())) {
-      request.versionType(VersionType.EXTERNAL);
-      if (config.hasExternalVersionHeader()) {
-        final Header versionHeader = record.headers().lastWithName(config.externalVersionHeader());
-        final byte[] versionValue = HEADER_CONVERTER.fromConnectHeader(
-                record.topic(),
-                versionHeader.key(),
-                versionHeader.schema(),
-                versionHeader.value()
-        );
-        try {
-          //fromConnectHeader byte output is UTF_8
-          request.version(Long.parseLong(new String(versionValue, StandardCharsets.UTF_8)));
-        } catch (NumberFormatException e) {
-          throw new ConnectException("Error converting to long: "
-                  + new String(versionValue, StandardCharsets.UTF_8), e);
-        }
-      } else {
-        request.version(record.kafkaOffset());
-      }
-    }
-
-    return request;
-  }
-
   // We need to pre process the Kafka Connect schema before converting to JSON as Elasticsearch
   // expects a different JSON format from the current JSON converter provides. Rather than
   // completely rewrite a converter for Elasticsearch, we will refactor the JSON converter to