diff --git a/pom.xml b/pom.xml
index a72e703e3..0eab167fd 100644
--- a/pom.xml
+++ b/pom.xml
@@ -114,7 +114,7 @@
         <testcontainer.version>2.0.3</testcontainer.version>
         <mockwebserver.version>5.3.2</mockwebserver.version>
         <!-- Dependencies -->
-        <kafka.version>3.9.2</kafka.version>
+        <kafka.version>4.3.0</kafka.version>
         <picocli.version>4.7.7</picocli.version>
         <jackson.version>2.18.6</jackson.version>
         <jinjava.version>2.8.3</jinjava.version>
@@ -175,11 +175,6 @@
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
-            <dependency>
-                <groupId>org.apache.zookeeper</groupId>
-                <artifactId>zookeeper</artifactId>
-                <version>3.8.6</version>
-            </dependency>
             <dependency>
                 <groupId>org.graalvm.sdk</groupId>
                 <artifactId>graal-sdk</artifactId>
diff --git a/providers/jikkou-provider-kafka/src/integration-test/java/io/jikkou/kafka/AbstractKafkaIntegrationTest.java b/providers/jikkou-provider-kafka/src/integration-test/java/io/jikkou/kafka/AbstractKafkaIntegrationTest.java
index 6bb378f36..65253ef79 100644
--- a/providers/jikkou-provider-kafka/src/integration-test/java/io/jikkou/kafka/AbstractKafkaIntegrationTest.java
+++ b/providers/jikkou-provider-kafka/src/integration-test/java/io/jikkou/kafka/AbstractKafkaIntegrationTest.java
@@ -35,7 +35,7 @@ public class AbstractKafkaIntegrationTest {
     private static final Logger LOG = LoggerFactory.getLogger(AbstractKafkaIntegrationTest.class);
     private static final Network KAFKA_NETWORK = Network.newNetwork();
 
-    public static final String APACHE_KAFKA_VERSION = "3.8.0";
+    public static final String APACHE_KAFKA_VERSION = "4.3.0";
     public static final int DEFAULT_NUM_PARTITIONS = 1;
     public static final short DEFAULT_REPLICATION_FACTOR = (short) 1;
 
diff --git a/providers/jikkou-provider-kafka/src/main/java/io/jikkou/kafka/internals/producer/DefaultProducerFactory.java b/providers/jikkou-provider-kafka/src/main/java/io/jikkou/kafka/internals/producer/DefaultProducerFactory.java
index 5806fcd6c..e7e9981a4 100644
--- a/providers/jikkou-provider-kafka/src/main/java/io/jikkou/kafka/internals/producer/DefaultProducerFactory.java
+++ b/providers/jikkou-provider-kafka/src/main/java/io/jikkou/kafka/internals/producer/DefaultProducerFactory.java
@@ -30,6 +30,7 @@
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.ProducerFencedException;
+import org.apache.kafka.common.metrics.KafkaMetric;
 import org.apache.kafka.common.serialization.Serializer;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
@@ -196,22 +197,28 @@ public void beginTransaction() throws ProducerFencedException {
         }
 
         /**
-         * @see KafkaProducer#sendOffsetsToTransaction(Map, String)
+         * @see KafkaProducer#sendOffsetsToTransaction(Map, ConsumerGroupMetadata)
          **/
-        @Deprecated
         @Override
         public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
-                                             String consumerGroupId) throws ProducerFencedException {
-            delegate.getResourceHandle().sendOffsetsToTransaction(offsets, consumerGroupId);
+                                             ConsumerGroupMetadata groupMetadata) throws ProducerFencedException {
+            delegate.getResourceHandle().sendOffsetsToTransaction(offsets, groupMetadata);
         }
 
         /**
-         * @see KafkaProducer#sendOffsetsToTransaction(Map, ConsumerGroupMetadata)
+         * @see KafkaProducer#registerMetricForSubscription(KafkaMetric)
          **/
         @Override
-        public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
-                                             ConsumerGroupMetadata groupMetadata) throws ProducerFencedException {
-            delegate.getResourceHandle().sendOffsetsToTransaction(offsets, groupMetadata);
+        public void registerMetricForSubscription(KafkaMetric metric) {
+            delegate.getResourceHandle().registerMetricForSubscription(metric);
+        }
+
+        /**
+         * @see KafkaProducer#unregisterMetricFromSubscription(KafkaMetric)
+         **/
+        @Override
+        public void unregisterMetricFromSubscription(KafkaMetric metric) {
+            delegate.getResourceHandle().unregisterMetricFromSubscription(metric);
         }
 
         /**