From b422fa2a47e6c4d3bfb0858ef3352758f31aa908 Mon Sep 17 00:00:00 2001
From: sp-gupta <spgupta@confluent.io>
Date: Tue, 7 Jan 2025 17:20:09 +0530
Subject: [PATCH] Fixes

---
 pom.xml                                          | 16 ++++++++++++----
 .../elasticsearch/ConfigCallbackHandler.java     |  4 ++--
 .../elasticsearch/ElasticsearchClientTest.java   |  2 +-
 .../integration/BaseConnectorIT.java             |  6 +++++-
 4 files changed, 20 insertions(+), 8 deletions(-)

diff --git a/pom.xml b/pom.xml
index d4040c430..4bc563f2b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -5,7 +5,7 @@
     <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>
@@ -39,7 +39,6 @@
     <properties>
         <es.version>7.17.23</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,8 @@
         <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>
+        <kafka.test.version>7.8.0-ccs</kafka.test.version>
         <jackson.version>2.16.0</jackson.version>
         <dependency.check.skip>true</dependency.check.skip>
     </properties>
@@ -182,7 +182,15 @@
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-clients</artifactId>
-            <version>${kafka.version}</version>
+            <version>${kafka.test.version}</version>
+            <classifier>test</classifier>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-server-common</artifactId>
+            <version>${kafka.test.version}</version>
             <classifier>test</classifier>
             <type>test-jar</type>
             <scope>test</scope>
diff --git a/src/main/java/io/confluent/connect/elasticsearch/ConfigCallbackHandler.java b/src/main/java/io/confluent/connect/elasticsearch/ConfigCallbackHandler.java
index 1c2676604..bd7a90317 100644
--- a/src/main/java/io/confluent/connect/elasticsearch/ConfigCallbackHandler.java
+++ b/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 {
diff --git a/src/test/java/io/confluent/connect/elasticsearch/ElasticsearchClientTest.java b/src/test/java/io/confluent/connect/elasticsearch/ElasticsearchClientTest.java
index e3b44ff98..b11e3275d 100644
--- a/src/test/java/io/confluent/connect/elasticsearch/ElasticsearchClientTest.java
+++ b/src/test/java/io/confluent/connect/elasticsearch/ElasticsearchClientTest.java
@@ -780,7 +780,7 @@ public void testSsl() throws Exception {
     helperClient = null;
 
     container.close();
-    container = ElasticsearchContainer.fromSystemProperties();
+    container = ElasticsearchContainer.fromSystemProperties().withSslEnabled(false);
     container.start();
   }
 
diff --git a/src/test/java/io/confluent/connect/elasticsearch/integration/BaseConnectorIT.java b/src/test/java/io/confluent/connect/elasticsearch/integration/BaseConnectorIT.java
index cb2b7ef34..19fa882bf 100644
--- a/src/test/java/io/confluent/connect/elasticsearch/integration/BaseConnectorIT.java
+++ b/src/test/java/io/confluent/connect/elasticsearch/integration/BaseConnectorIT.java
@@ -15,6 +15,8 @@
 
 package io.confluent.connect.elasticsearch.integration;
 
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 
@@ -41,8 +43,10 @@ public abstract class BaseConnectorIT {
   protected RestApp restApp;
 
   protected void startConnect() {
+    Map<String, String> workerProps = new HashMap<>();
+    workerProps.put("plugin.discovery","hybrid_warn");
     connect = new EmbeddedConnectCluster.Builder()
-        .name("elasticsearch-it-connect-cluster")
+        .name("elasticsearch-it-connect-cluster").workerProps(workerProps)
         .build();
 
     // start the clusters