chore: upgrade to Confluent Platform 8.0

.semaphore/semaphore.yml

@@ -24,7 +24,7 @@ global_job_config:
   prologue:
     commands:
       - checkout
-      - sem-version java 8
+      - sem-version java 17
       - . cache-maven restore
 
 blocks:

pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.confluent</groupId>
         <artifactId>common</artifactId>
-        <version>[7.9.0, 7.9.1)</version>
+        <version>[8.0.0, 8.0.1)</version>
     </parent>
 
     <artifactId>kafka-connect-elasticsearch</artifactId>
@@ -53,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>[7.9.0,7.9.1)</confluent.version>
+        <confluent.version>[8.0.0,8.0.1)</confluent.version>
         <jackson.version>2.16.0</jackson.version>
         <dependency.check.skip>true</dependency.check.skip>
     </properties>
@@ -159,6 +159,11 @@
             <artifactId>junit-jupiter-api</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.junit.vintage</groupId>
+            <artifactId>junit-vintage-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>connect-runtime</artifactId>
@@ -208,6 +213,12 @@
             <scope>test</scope>
             <version>${kafka.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-test-common-runtime</artifactId>
+            <version>${kafka.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>elasticsearch</artifactId>
@@ -233,9 +244,9 @@
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>com.github.tomakehurst</groupId>
-            <artifactId>wiremock-jre8</artifactId>
-            <version>2.30.1</version>
+            <groupId>org.wiremock</groupId>
+            <artifactId>wiremock-standalone</artifactId>
+            <version>3.3.1</version>
             <scope>test</scope>
         </dependency>
         <dependency>

src/main/resources/META-INF/services/org.apache.kafka.connect.sink.SinkConnector

@@ -0,0 +1,2 @@
+io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
+

src/test/java/io/confluent/connect/elasticsearch/integration/BlockingTransformer.java

@@ -10,18 +10,27 @@
 
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * Transformer that blocks all incoming requests until {@link #release(int)} is called
  * to fairly unblock a given number of requests.
  */
 public class BlockingTransformer extends ResponseTransformer {
 
+  // Static reference to track the current instance (WireMock creates via reflection)
+  private static final AtomicReference<BlockingTransformer> CURRENT_INSTANCE = new AtomicReference<>();
+
   private final Semaphore s = new Semaphore(0, true);
   private final AtomicInteger requestCount = new AtomicInteger();
 
   public static final String NAME = "blockingTransformer";
 
+  public BlockingTransformer() {
+    // Register this instance when WireMock creates it via reflection
+    CURRENT_INSTANCE.set(this);
+  }
+
   @Override
   public Response transform(Request request, Response response, FileSource files, Parameters parameters) {
     try {
@@ -63,10 +72,17 @@ public boolean applyGlobally() {
     return false;
   }
 
+  /**
+   * Get the BlockingTransformer instance created by WireMock.
+   * WireMock creates the instance via reflection when the rule is initialized.
+   */
   public static BlockingTransformer getInstance(WireMockRule wireMockRule) {
-    return wireMockRule.getOptions()
-            .extensionsOfType(BlockingTransformer.class)
-            .get(NAME);
+    BlockingTransformer instance = CURRENT_INSTANCE.get();
+    if (instance == null) {
+      throw new IllegalStateException(
+              "No BlockingTransformer instance found. Ensure WireMockRule is initialized.");
+    }
+    return instance;
   }
 
 }