Merge 15.0.x into pr_merge_from_14_1_x_to_15_0_x - resolved automatically

Author: jjain1259

pom.xml

@@ -5,11 +5,11 @@
     <parent>
         <groupId>io.confluent</groupId>
         <artifactId>common</artifactId>
-        <version>7.0.12</version>
+        <version>[7.9.0, 7.9.1)</version>
     </parent>
 
     <artifactId>kafka-connect-elasticsearch</artifactId>
-    <version>14.1.4-SNAPSHOT</version>
+    <version>15.0.2-SNAPSHOT</version>
     <packaging>jar</packaging>
     <name>kafka-connect-elasticsearch</name>
     <organization>
@@ -33,13 +33,12 @@
         <connection>scm:git:git://github.com/confluentinc/kafka-connect-elasticsearch.git</connection>
         <developerConnection>scm:git:[email protected]:confluentinc/kafka-connect-elasticsearch.git</developerConnection>
         <url>https://github.com/confluentinc/kafka-connect-elasticsearch</url>
-        <tag>14.1.x</tag>
+        <tag>15.0.x</tag>
     </scm>
 
     <properties>
         <es.version>7.17.24</es.version>
         <hamcrest.version>1.3</hamcrest.version>
-        <mockito.version>2.28.2</mockito.version>
         <gson.version>2.9.0</gson.version>
         <test.containers.version>1.16.3</test.containers.version>
         <kafka.connect.maven.plugin.version>0.11.1</kafka.connect.maven.plugin.version>
@@ -54,7 +53,7 @@
         <dependency.check.version>6.1.6</dependency.check.version>
         <confluent.maven.repo>http://packages.confluent.io/maven/</confluent.maven.repo>
         <commons.codec.version>1.15</commons.codec.version>
-        <confluent.version>${io.confluent.common.version}</confluent.version>
+        <confluent.version>[7.9.0,7.9.1)</confluent.version>
         <jackson.version>2.16.0</jackson.version>
         <dependency.check.skip>true</dependency.check.skip>
     </properties>
@@ -182,7 +181,15 @@
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-clients</artifactId>
-            <version>${kafka.version}</version>
+            <version>7.8.0-ccs</version>
+            <classifier>test</classifier>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-server-common</artifactId>
+            <version>7.8.0-ccs</version>
             <classifier>test</classifier>
             <type>test-jar</type>
             <scope>test</scope>
@@ -312,6 +319,19 @@
         </dependencies>
     </dependencyManagement>
 
+  <distributionManagement>
+    <repository>
+      <id>aws-release</id>
+      <name>AWS Release Repository</name>
+      <url>${confluent.release.repo}</url>
+    </repository>
+    <snapshotRepository>
+      <id>aws-snapshot</id>
+      <name>AWS Snapshot Repository</name>
+      <url>${confluent.snapshot.repo}</url>
+    </snapshotRepository>
+  </distributionManagement>
+
     <build>
         <plugins>
             <plugin>

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

@@ -54,7 +54,7 @@
 import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
 import org.apache.http.nio.reactor.ConnectingIOReactor;
 import org.apache.http.nio.reactor.IOReactorException;
-import org.apache.kafka.common.network.Mode;
+import org.apache.kafka.common.network.ConnectionMode;
 import org.apache.kafka.common.security.ssl.SslFactory;
 import org.apache.kafka.connect.errors.ConnectException;
 import org.elasticsearch.client.RestClientBuilder.HttpClientConfigCallback;
@@ -258,7 +258,7 @@ private void configureSslContext(HttpAsyncClientBuilder builder) {
    * Gets the SslContext for the client.
    */
   private SSLContext sslContext() {
-    SslFactory sslFactory = new SslFactory(Mode.CLIENT, null, false);
+    SslFactory sslFactory = new SslFactory(ConnectionMode.CLIENT, null, false);
     sslFactory.configure(config.sslConfigs());
 
     try {

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

@@ -121,7 +121,7 @@ private String convertKey(Schema keySchema, Object key) {
     }
   }
 
-  public DocWriteRequest<?> convertRecord(SinkRecord record, String index) {
+  public DocWriteRequest<?> convertRecord(SinkRecord record, String resourceName) {
     if (record.value() == null) {
       switch (config.behaviorOnNullValues()) {
         case IGNORE:
@@ -166,7 +166,7 @@ public DocWriteRequest<?> convertRecord(SinkRecord record, String index) {
 
     // delete
     if (record.value() == null) {
-      return maybeAddExternalVersioning(new DeleteRequest(index).id(id), record);
+      return maybeAddExternalVersioning(new DeleteRequest(resourceName).id(id), record);
     }
 
     String payload = getPayload(record);
@@ -175,14 +175,14 @@ public DocWriteRequest<?> convertRecord(SinkRecord record, String index) {
     // index
     switch (config.writeMethod()) {
       case UPSERT:
-        return new UpdateRequest(index, id)
+        return new UpdateRequest(resourceName, id)
             .doc(payload, XContentType.JSON)
             .upsert(payload, XContentType.JSON)
             .retryOnConflict(Math.min(config.maxInFlightRequests(), 5));
       case INSERT:
         OpType opType = config.isDataStream() ? OpType.CREATE : OpType.INDEX;
         IndexRequest req =
-                new IndexRequest(index).source(payload, XContentType.JSON).opType(opType);
+                new IndexRequest(resourceName).source(payload, XContentType.JSON).opType(opType);
         if (config.useAutogeneratedIds()) {
           return req;
         }

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

@@ -280,13 +280,14 @@ public boolean createIndexOrDataStream(String name) {
   /**
    * Creates a mapping for the given index and schema.
    *
-   * @param index the index to create the mapping for
+   * @param resourceName the resource to create the mapping for
    * @param schema the schema to map
    */
-  public void createMapping(String index, Schema schema) {
-    PutMappingRequest request = new PutMappingRequest(index).source(Mapping.buildMapping(schema));
+  public void createMapping(String resourceName, Schema schema) {
+    PutMappingRequest request = new PutMappingRequest(resourceName)
+            .source(Mapping.buildMapping(schema));
     callWithRetries(
-        String.format("create mapping for index %s with schema %s", index, schema),
+        String.format("create mapping for resource %s with schema %s", resourceName, schema),
         () -> client.indices().putMapping(request, RequestOptions.DEFAULT)
     );
   }
@@ -318,11 +319,11 @@ public void waitForInFlightRequests() {
 
   /**
    * Checks whether the index already has a mapping or not.
-   * @param index the index to check
+   * @param resourceName the resource to check
    * @return true if a mapping exists, false if it does not
    */
-  public boolean hasMapping(String index) {
-    MappingMetadata mapping = mapping(index);
+  public boolean hasMapping(String resourceName) {
+    MappingMetadata mapping = mapping(resourceName);
     return mapping != null && mapping.sourceAsMap() != null && !mapping.sourceAsMap().isEmpty();
   }