Repository: kroxylicious/kroxylicious
Pull request #178 (2023-03-08 13:57:29)

Added kroxylicious juni5 extension dependency and removed duplicates
### Type of change

- Refactoring

### Description

- Adding the kroxylicious junit5 extension dependency to kroxylicious project. 
- Removed the duplicates
- Adapted the integration test cases needed

### Additional Context

This PR would replace this one from @tombentley https://github.com/kroxylicious/kroxylicious/pull/93. My changes are based on that PR but added a few more changes on new test cases.



Patch:
diff --git a/.gitignore b/.gitignore
index bbfa3001..1e4b9ab5 100644
--- a/.gitignore
+++ b/.gitignore
@@ -47,7 +47,8 @@ bin/
  # since they will be recreated, and may cause churn.  Uncomment if using
  # auto-import.
  # .idea/artifacts
- # .idea/compiler.xml
+ .idea/compiler.xml
+ .idea/encodings.xml
  # .idea/jarRepositories.xml
  # .idea/modules.xml
  # .idea/*.iml
diff --git a/pom.xml b/pom.xml
index 950ab199..d16f19d8 100644
--- a/pom.xml
+++ b/pom.xml
@@ -37,7 +37,6 @@
         <kafka.version>3.4.0</kafka.version>
         <log4j.version>2.17.2</log4j.version>
         <netty.version>4.1.89.Final</netty.version>
-        <testcontainers.version>1.17.6</testcontainers.version>
         <netty.io_uring.version>0.0.14.Final</netty.io_uring.version>
         <netty.epoll.classifier>linux-x86_64</netty.epoll.classifier>
         <netty.io_uring.classifier>linux-x86_64</netty.io_uring.classifier>
@@ -46,8 +45,8 @@
         <assertj-core.version>3.22.0</assertj-core.version>
         <mockito-core.version>4.6.1</mockito-core.version>
         <slf4j-api.version>1.7.36</slf4j-api.version>
-        <zookeeper.version>3.6.3</zookeeper.version>
         <micrometer.version>1.10.2</micrometer.version>
+        <kroxy.extension.version>0.1</kroxy.extension.version>
     </properties>
 
     <name>Kroxylicious Parent</name>
@@ -115,33 +114,6 @@
                 <artifactId>kafka-clients</artifactId>
                 <version>${kafka.version}</version>
             </dependency>
-            <dependency>
-                <groupId>org.apache.kafka</groupId>
-                <artifactId>kafka_2.13</artifactId>
-                <version>${kafka.version}</version>
-                <scope>test</scope>
-            </dependency>
-            <dependency>
-                <groupId>org.apache.kafka</groupId>
-                <artifactId>kafka-server-common</artifactId>
-                <version>${kafka.version}</version>
-                <scope>test</scope>
-            </dependency>
-            <dependency>
-                <groupId>org.apache.zookeeper</groupId>
-                <artifactId>zookeeper</artifactId>
-                <version>${zookeeper.version}</version>
-                <exclusions>
-                    <exclusion>
-                        <groupId>org.slf4j</groupId>
-                        <artifactId>slf4j-log4j12</artifactId>
-                    </exclusion>
-                    <exclusion>
-                        <groupId>log4j</groupId>
-                        <artifactId>log4j</artifactId>
-                    </exclusion>
-                </exclusions>
-            </dependency>
             <dependency>
                 <groupId>org.apache.logging.log4j</groupId>
                 <artifactId>log4j-bom</artifactId>
@@ -166,24 +138,6 @@
                 <artifactId>slf4j-api</artifactId>
                 <version>${slf4j-api.version}</version>
             </dependency>
-            <dependency>
-                <groupId>org.testcontainers</groupId>
-                <artifactId>testcontainers</artifactId>
-                <version>${testcontainers.version}</version>
-                <scope>test</scope>
-            </dependency>
-            <dependency>
-                <groupId>org.testcontainers</groupId>
-                <artifactId>kafka</artifactId>
-                <version>${testcontainers.version}</version>
-                <scope>test</scope>
-            </dependency>
-            <dependency>
-                <groupId>org.testcontainers</groupId>
-                <artifactId>junit-jupiter</artifactId>
-                <version>${testcontainers.version}</version>
-                <scope>test</scope>
-            </dependency>
             <dependency>
                 <groupId>org.projectlombok</groupId>
                 <artifactId>lombok</artifactId>
@@ -202,6 +156,18 @@
                 <version>${mockito-core.version}</version>
                 <scope>test</scope>
             </dependency>
+            <dependency>
+                <groupId>io.kroxylicious.testing</groupId>
+                <artifactId>testing-junit5-extension</artifactId>
+                <version>${kroxy.extension.version}</version>
+                <scope>test</scope>
+            </dependency>
+            <dependency>
+                <groupId>io.kroxylicious.testing</groupId>
+                <artifactId>testing-api</artifactId>
+                <version>${kroxy.extension.version}</version>
+                <scope>test</scope>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 
@@ -279,27 +245,27 @@
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-clean-plugin</artifactId>
-                    <version>3.1.0</version>
+                    <version>3.2.0</version>
                 </plugin>
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-dependency-plugin</artifactId>
-                    <version>3.3.0</version>
+                    <version>3.5.0</version>
                 </plugin>
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-deploy-plugin</artifactId>
-                    <version>3.0.0-M1</version>
+                    <version>3.1.0</version>
                 </plugin>
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-enforcer-plugin</artifactId>
-                    <version>3.0.0-M3</version>
+                    <version>3.2.1</version>
                 </plugin>
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-failsafe-plugin</artifactId>
-                    <version>3.0.0-M7</version>
+                    <version>3.0.0-M9</version>
                     <configuration>
                         <skipTests>${skipITs}</skipTests>
                     </configuration>
@@ -307,12 +273,12 @@
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-install-plugin</artifactId>
-                    <version>3.0.0-M1</version>
+                    <version>3.1.0</version>
                 </plugin>
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-jar-plugin</artifactId>
-                    <version>3.2.0</version>
+                    <version>3.3.0</version>
                 </plugin>
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
@@ -322,17 +288,17 @@
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-resources-plugin</artifactId>
-                    <version>3.2.0</version>
+                    <version>3.3.0</version>
                 </plugin>
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-site-plugin</artifactId>
-                    <version>3.9.1</version>
+                    <version>3.12.1</version>
                 </plugin>
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-surefire-plugin</artifactId>
-                    <version>3.0.0-M5</version>
+                    <version>3.0.0-M9</version>
                     <configuration>
                         <skipTests>${skipUTs}</skipTests>
                         <runOrder>random</runOrder>


Test patch:
diff --git a/integrationtests/pom.xml b/integrationtests/pom.xml
index 57577d5e..0b660351 100644
--- a/integrationtests/pom.xml
+++ b/integrationtests/pom.xml
@@ -19,6 +19,10 @@
     <artifactId>integrationtests</artifactId>
     <description>Integration tests where Kroxylicious is tested end-to-end with a real Kafka cluster.</description>
 
+    <properties>
+        <java.version>19</java.version>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>io.kroxylicious</groupId>
@@ -33,58 +37,24 @@
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.apache.logging.log4j</groupId>
-            <artifactId>log4j-slf4j-impl</artifactId>
+            <groupId>io.kroxylicious.testing</groupId>
+            <artifactId>testing-junit5-extension</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka-clients</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka_2.13</artifactId>
-            <scope>compile</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka-server-common</artifactId>
-            <scope>compile</scope>
+            <groupId>io.kroxylicious.testing</groupId>
+            <artifactId>testing-api</artifactId>
+            <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.apache.zookeeper</groupId>
-            <artifactId>zookeeper</artifactId>
-            <exclusions>
-                <!-- exclude so log4j2 can be used instead -->
-                <exclusion>
-                    <groupId>org.slf4j</groupId>
-                    <artifactId>slf4j-log4j12</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>log4j</groupId>
-                    <artifactId>log4j</artifactId>
-                </exclusion>
-            </exclusions>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-slf4j-impl</artifactId>
+            <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.testcontainers</groupId>
-            <artifactId>testcontainers</artifactId>
-            <scope>compile</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.testcontainers</groupId>
-            <artifactId>kafka</artifactId>
-            <scope>compile</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.testcontainers</groupId>
-            <artifactId>junit-jupiter</artifactId>
-            <scope>compile</scope>
-        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-params</artifactId>
@@ -95,10 +65,6 @@
             <artifactId>assertj-core</artifactId>
             <scope>compile</scope>
         </dependency>
-        <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
-        </dependency>
     </dependencies>
 
     <build>
diff --git a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/ContainerBasedKafkaCluster.java b/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/ContainerBasedKafkaCluster.java
deleted file mode 100644
index e8ee75e8..00000000
--- a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/ContainerBasedKafkaCluster.java
+++ /dev/null
@@ -1,285 +0,0 @@
-/*
- * Copyright Kroxylicious Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-package io.kroxylicious.proxy.testkafkacluster;
-
-import java.io.ByteArrayOutputStream;
-import java.io.File;
-import java.io.FileWriter;
-import java.io.IOException;
-import java.io.UncheckedIOException;
-import java.nio.file.Path;
-import java.time.Duration;
-import java.time.OffsetDateTime;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.function.Supplier;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-import org.junit.jupiter.api.TestInfo;
-import org.rnorth.ducttape.unreliables.Unreliables;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.Network;
-import org.testcontainers.containers.output.OutputFrame;
-import org.testcontainers.images.builder.Transferable;
-import org.testcontainers.lifecycle.Startable;
-import org.testcontainers.lifecycle.Startables;
-import org.testcontainers.utility.DockerImageName;
-
-import com.github.dockerjava.api.command.InspectContainerResponse;
-
-import io.kroxylicious.proxy.testkafkacluster.KafkaClusterConfig.KafkaEndpoints;
-import io.kroxylicious.proxy.testkafkacluster.KafkaClusterConfig.KafkaEndpoints.Endpoint;
-
-import lombok.SneakyThrows;
-
-/**
- * Provides an easy way to launch a Kafka cluster with multiple brokers in a container
- */
-public class ContainerBasedKafkaCluster implements Startable, KafkaCluster {
-
-    private static final Logger LOGGER = LoggerFactory.getLogger(ContainerBasedKafkaCluster.class);
-    public static final int KAFKA_PORT = 9093;
-    public static final int ZOOKEEPER_PORT = 2181;
-
-    // FIXME: uses container image built from https://github.com/ozangunalp/kafka-native/pull/5
-    private static final DockerImageName DEFAULT_KAFKA_IMAGE = DockerImageName.parse("quay.io/k_wall/kafka-native:1.0.0-SNAPSHOT");
-
-    // FIXME: uses container image built from https://github.com/k-wall/zookeeper-native, move the repo to a permanent location.
-    private static final DockerImageName DEFAULT_ZOOKEEPER_IMAGE = DockerImageName.parse("quay.io/k_wall/zookeeper-native:1.0.0-SNAPSHOT");
-    private static final int READY_TIMEOUT_SECONDS = 120;
-    private static final String KAFKA_CLUSTER_READY_FLAG = "/tmp/kafka_cluster_ready";
-    private static final String ZOOKEEPER_READY_FLAG = "/tmp/kafka_zookeeper_ready";
-    private final DockerImageName kafkaImage;
-    private final DockerImageName zookeeperImage;
-    private final KafkaClusterConfig clusterConfig;
-    private final Network network = Network.newNetwork();
-    private final ZookeeperContainer zookeeper;
-    private final Collection<KafkaContainer> brokers;
-
-    static {
-        if (!System.getenv().containsKey("TESTCONTAINERS_RYUK_DISABLED")) {
-            LOGGER.warn("As per https://github.com/containers/podman/issues/7927#issuecomment-731525556 if using podman, set env var TESTCONTAINERS_RYUK_DISABLED=true");
-        }
-    }
-
-    public ContainerBasedKafkaCluster(KafkaClusterConfig clusterConfig) {
-        this(null, null, clusterConfig);
-    }
-
-    public ContainerBasedKafkaCluster(DockerImageName kafkaImage, DockerImageName zookeeperImage, KafkaClusterConfig clusterConfig) {
-        this.kafkaImage = Optional.ofNullable(kafkaImage).orElse(DEFAULT_KAFKA_IMAGE);
-        this.zookeeperImage = Optional.ofNullable(zookeeperImage).orElse(DEFAULT_ZOOKEEPER_IMAGE);
-        this.clusterConfig = clusterConfig;
-
-        var name = Optional.ofNullable(clusterConfig.getTestInfo())
-                .map(TestInfo::getDisplayName)
-                .map(s -> s.replaceFirst("\\(\\)$", ""))
-                .map(s -> String.format("%s.%s", s, OffsetDateTime.now()))
-                .orElse(null);
-
-        if (this.clusterConfig.isKraftMode()) {
-            this.zookeeper = null;
-        }
-        else {
-            this.zookeeper = new ZookeeperContainer(this.zookeeperImage)
-                    .withName(name)
-                    .withNetwork(network)
-                    .withEnv("SERVER_ZOOKEEPER_READY_FLAG_FILE", ZOOKEEPER_READY_FLAG)
-                    // .withEnv("QUARKUS_LOG_LEVEL", "DEBUG") // Enables org.apache.zookeeper logging too
-                    .withNetworkAliases("zookeeper");
-        }
-
-        Supplier<KafkaEndpoints> endPointConfigSupplier = () -> new KafkaEndpoints() {
-            final List<Integer> ports = Utils.preAllocateListeningPorts(clusterConfig.getBrokersNum()).collect(Collectors.toList());
-
-            @Override
-            public EndpointPair getClientEndpoint(int brokerId) {
-                return EndpointPair.builder().bind(new Endpoint("0.0.0.0", 9093)).connect(new Endpoint("localhost", ports.get(brokerId))).build();
-            }
-
-            @Override
-            public EndpointPair getInterBrokerEndpoint(int brokerId) {
-                return EndpointPair.builder().bind(new Endpoint("0.0.0.0", 9092)).connect(new Endpoint(String.format("broker-%d", brokerId), 9092)).build();
-            }
-
-            @Override
-            public EndpointPair getControllerEndpoint(int brokerId) {
-                return EndpointPair.builder().bind(new Endpoint("0.0.0.0", 9091)).connect(new Endpoint(String.format("broker-%d", brokerId), 9091)).build();
-            }
-        };
-        Supplier<Endpoint> zookeeperEndpointSupplier = () -> new Endpoint("zookeeper", ContainerBasedKafkaCluster.ZOOKEEPER_PORT);
-        this.brokers = clusterConfig.getBrokerConfigs(endPointConfigSupplier, zookeeperEndpointSupplier).map(holder -> {
-            String netAlias = "broker-" + holder.getBrokerNum();
-            KafkaContainer kafkaContainer = new KafkaContainer(this.kafkaImage)
-                    .withName(name)
-                    .withNetwork(this.network)
-                    .withNetworkAliases(netAlias)
-                    // .withEnv("QUARKUS_LOG_LEVEL", "DEBUG") // Enables org.apache.kafka logging too
-                    .withEnv("SERVER_PROPERTIES_FILE", "/cnf/server.properties")
-                    .withEnv("SERVER_CLUSTER_ID", holder.getKafkaKraftClusterId())
-                    .withEnv("SERVER_CLUSTER_READY_FLAG_FILE", KAFKA_CLUSTER_READY_FLAG)
-                    .withEnv("SERVER_CLUSTER_READY_NUM_BROKERS", clusterConfig.getBrokersNum().toString())
-                    .withCopyToContainer(Transferable.of(propertiesToBytes(holder.getProperties()), 0644), "/cnf/server.properties")
-                    .withStartupTimeout(Duration.ofMinutes(2));
-
-            kafkaContainer.addFixedExposedPort(holder.getExternalPort(), KAFKA_PORT);
-
-            if (!this.clusterConfig.isKraftMode()) {
-                kafkaContainer.dependsOn(this.zookeeper);
-            }
-            return kafkaContainer;
-        }).collect(Collectors.toList());
-    }
-
-    private byte[] propertiesToBytes(Properties properties) {
-        try (var byteArrayOutputStream = new ByteArrayOutputStream()) {
-            properties.store(byteArrayOutputStream, "server.properties");
-            return byteArrayOutputStream.toByteArray();
-        }
-        catch (IOException e) {
-            throw new UncheckedIOException(e);
-        }
-    }
-
-    @Override
-    public String getBootstrapServers() {
-        return brokers.stream()
-                .map(b -> String.format("localhost:%d", b.getMappedPort(KAFKA_PORT)))
-                .collect(Collectors.joining(","));
-    }
-
-    private Stream<GenericContainer<?>> allContainers() {
-        return Stream.concat(
-                this.brokers.stream(),
-                Stream.ofNullable(this.zookeeper));
-    }
-
-    @Override
-    @SneakyThrows
-    public void start() {
-        try {
-            if (zookeeper != null) {
-                zookeeper.start();
-                awaitContainerReadyFlagFile(zookeeper, ZOOKEEPER_READY_FLAG);
-            }
-            Startables.deepStart(brokers.stream()).get(READY_TIMEOUT_SECONDS, TimeUnit.SECONDS);
-        }
-        catch (InterruptedException | ExecutionException | TimeoutException e) {
-            if (e instanceof InterruptedException) {
-                Thread.currentThread().interrupt();
-            }
-            stop();
-            throw new RuntimeException("startup failed or timed out", e);
-        }
-
-        awaitContainerReadyFlagFile(this.brokers.iterator().next(), KAFKA_CLUSTER_READY_FLAG);
-    }
-
-    @Override
-    public void close() {
-        this.stop();
-    }
-
-    private void awaitContainerReadyFlagFile(GenericContainer<?> container, String kafkaClusterReadyFlag) {
-        Unreliables.retryUntilTrue(READY_TIMEOUT_SECONDS, TimeUnit.SECONDS, () -> {
-            container.execInContainer(
-                    "sh", "-c",
-                    String.format("while [ ! -f %s ]; do sleep .1; done", kafkaClusterReadyFlag));
-            LOGGER.info("Container {} ready", container.getDockerImageName());
-            return true;
-        });
-    }
-
-    @Override
-    public void stop() {
-        allContainers().parallel().forEach(GenericContainer::stop);
-    }
-
-    @Override
-    public Map<String, Object> getKafkaClientConfiguration() {
-        return clusterConfig.getConnectConfigForCluster(getBootstrapServers());
-    }
-
-    // In kraft mode, currently "Advertised listeners cannot be altered when using a Raft-based metadata quorum", so we
-    // need to know the listening port before we start the kafka container. For this reason, we need this override
-    // to expose addFixedExposedPort to for use.
-    public static class KafkaContainer extends LoggingGenericContainer<KafkaContainer> {
-
-        public KafkaContainer(final DockerImageName dockerImageName) {
-            super(dockerImageName);
-        }
-
-        protected void addFixedExposedPort(int hostPort, int containerPort) {
-            super.addFixedExposedPort(hostPort, containerPort);
-        }
-
-    }
-
-    public static class ZookeeperContainer extends LoggingGenericContainer<ZookeeperContainer> {
-        public ZookeeperContainer(DockerImageName zookeeperImage) {
-            super(zookeeperImage);
-        }
-    }
-
-    public static class LoggingGenericContainer<C extends GenericContainer<C>>
-            extends GenericContainer<C> {
-        private static final String CONTAINER_LOGS_DIR = "container.logs.dir";
-        private String name;
-
-        public LoggingGenericContainer(DockerImageName dockerImageName) {
-            super(dockerImageName);
-        }
-
-        @Override
-        protected void containerIsStarting(InspectContainerResponse containerInfo) {
-            super.containerIsStarting(containerInfo);
-
-            Optional.ofNullable(System.getProperty(CONTAINER_LOGS_DIR)).ifPresent(logDir -> {
-                var target = Path.of(logDir);
-                if (name != null) {
-                    target = target.resolve(name);
-                }
-                target = target.resolve(String.format("%s.%s.%s", getContainerName().replaceFirst(File.separator, ""), getContainerId(), "log"));
-                target.getParent().toFile().mkdirs();
-                try {
-                    var writer = new FileWriter(target.toFile());
-                    super.followOutput(outputFrame -> {
-                        try {
-                            if (outputFrame.equals(OutputFrame.END)) {
-                                writer.close();
-                            }
-                            else {
-                                writer.write(outputFrame.getUtf8String());
-                            }
-                        }
-                        catch (IOException e) {
-                            // ignore
-                        }
-                    });
-                }
-                catch (IOException e) {
-                    logger().warn("Failed to create container log file: {}", target);
-                }
-            });
-
-        }
-
-        public LoggingGenericContainer<C> withName(String name) {
-            this.name = name;
-            return this;
-        }
-    }
-
-}
diff --git a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/InVMKafkaCluster.java b/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/InVMKafkaCluster.java
deleted file mode 100644
index dbe5fc2a..00000000
--- a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/InVMKafkaCluster.java
+++ /dev/null
@@ -1,202 +0,0 @@
-/*
- * Copyright Kroxylicious Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-
-package io.kroxylicious.proxy.testkafkacluster;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.UncheckedIOException;
-import java.net.InetSocketAddress;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.ArrayList;
-import java.util.Comparator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.function.Supplier;
-import java.util.stream.Collectors;
-
-import org.apache.kafka.common.utils.Time;
-import org.apache.zookeeper.server.ServerCnxnFactory;
-import org.apache.zookeeper.server.ZooKeeperServer;
-import org.jetbrains.annotations.NotNull;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import io.kroxylicious.proxy.testkafkacluster.KafkaClusterConfig.ConfigHolder;
-import io.kroxylicious.proxy.testkafkacluster.KafkaClusterConfig.KafkaEndpoints;
-
-import kafka.server.KafkaConfig;
-import kafka.server.KafkaRaftServer;
-import kafka.server.KafkaServer;
-import kafka.server.Server;
-import kafka.tools.StorageTool;
-import scala.Option;
-
-import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_BOOTSTRAP_VERSION;
-
-public class InVMKafkaCluster implements KafkaCluster {
-    private static final Logger LOGGER = LoggerFactory.getLogger(InVMKafkaCluster.class);
-
-    private final KafkaClusterConfig clusterConfig;
-    private final Path tempDirectory;
-    private final ServerCnxnFactory zooFactory;
-    private final ZooKeeperServer zooServer;
-    private final List<Server> servers;
-    private final List<String> bootstraps = new ArrayList<>();
-
-    public InVMKafkaCluster(KafkaClusterConfig clusterConfig) {
-        this.clusterConfig = clusterConfig;
-        try {
-            tempDirectory = Files.createTempDirectory("kafka");
-            tempDirectory.toFile().deleteOnExit();
-
-            // kraft mode: per-broker: 1 external port + 1 inter-broker port + 1 controller port
-            // zk mode: per-cluster: 1 zk port; per-broker: 1 external port + 1 inter-broker port
-            var numPorts = clusterConfig.getBrokersNum() * (clusterConfig.isKraftMode() ? 3 : 2) + (clusterConfig.isKraftMode() ? 0 : 1);
-            LinkedList<Integer> ports = Utils.preAllocateListeningPorts(numPorts).collect(Collectors.toCollection(LinkedList::new));
-
-            final Supplier<KafkaEndpoints.Endpoint> zookeeperEndpointSupplier;
-            if (!clusterConfig.isKraftMode()) {
-                var zookeeperPort = ports.pop();
-
-                zooFactory = ServerCnxnFactory.createFactory(new InetSocketAddress("localhost", zookeeperPort), 1024);
-
-                var zoo = tempDirectory.resolve("zoo");
-                var snapshotDir = zoo.resolve("snapshot");
-                var logDir = zoo.resolve("log");
-                snapshotDir.toFile().mkdirs();
-                logDir.toFile().mkdirs();
-
-                zooServer = new ZooKeeperServer(snapshotDir.toFile(), logDir.toFile(), 500);
-                zookeeperEndpointSupplier = () -> new KafkaEndpoints.Endpoint("localhost", zookeeperPort);
-            }
-            else {
-                zooFactory = null;
-                zooServer = null;
-                zookeeperEndpointSupplier = null;
-            }
-
-            Supplier<KafkaEndpoints> kafkaEndpointsSupplier = () -> new KafkaEndpoints() {
-                final List<Integer> clientPorts = ports.subList(0, clusterConfig.getBrokersNum());
-                final List<Integer> interBrokerPorts = ports.subList(clusterConfig.getBrokersNum(), 2 * clusterConfig.getBrokersNum());
-                final List<Integer> controllerPorts = ports.subList(clusterConfig.getBrokersNum() * 2, ports.size());
-
-                @Override
-                public EndpointPair getClientEndpoint(int brokerId) {
-                    var port = clientPorts.get(brokerId);
-                    return EndpointPair.builder().bind(new Endpoint("0.0.0.0", port)).connect(new Endpoint("localhost", port)).build();
-                }
-
-                @Override
-                public EndpointPair getInterBrokerEndpoint(int brokerId) {
-                    var port = interBrokerPorts.get(brokerId);
-                    return EndpointPair.builder().bind(new Endpoint("0.0.0.0", port)).connect(new Endpoint("localhost", port)).build();
-                }
-
-                @Override
-                public EndpointPair getControllerEndpoint(int brokerId) {
-                    if (!clusterConfig.isKraftMode()) {
-                        throw new IllegalStateException();
-                    }
-                    var port = controllerPorts.get(brokerId);
-                    return EndpointPair.builder().bind(new Endpoint("0.0.0.0", port)).connect(new Endpoint("localhost", port)).build();
-                }
-            };
-
-            servers = clusterConfig.getBrokerConfigs(kafkaEndpointsSupplier, zookeeperEndpointSupplier).map(this::buildKafkaServer).collect(Collectors.toList());
-
-        }
-        catch (IOException e) {
-            throw new UncheckedIOException(e);
-        }
-    }
-
-    @NotNull
-    private Server buildKafkaServer(ConfigHolder c) {
-        bootstraps.add(c.getEndpoint());
-        KafkaConfig config = buildBrokerConfig(c, tempDirectory);
-        Option<String> threadNamePrefix = Option.apply(null);
-
-        boolean kraftMode = clusterConfig.isKraftMode();
-        if (kraftMode) {
-            var directories = StorageTool.configToLogDirectories(config);
-            var clusterId = c.getKafkaKraftClusterId();
-            var metaProperties = StorageTool.buildMetadataProperties(clusterId, config);
-            StorageTool.formatCommand(System.out, directories, metaProperties, MINIMUM_BOOTSTRAP_VERSION, true);
-            return new KafkaRaftServer(config, Time.SYSTEM, threadNamePrefix);
-        }
-        else {
-            return new KafkaServer(config, Time.SYSTEM, threadNamePrefix, false);
-
-        }
-    }
-
-    @NotNull
-    private KafkaConfig buildBrokerConfig(ConfigHolder c, Path tempDirectory) {
-        Properties properties = new Properties();
-        properties.putAll(c.getProperties());
-        Path logsDir = tempDirectory.resolve(String.format("broker-%d", c.getBrokerNum()));
-        properties.setProperty(KafkaConfig.LogDirProp(), logsDir.toAbsolutePath().toString());
-
-        return new KafkaConfig(properties);
-    }
-
-    @Override
-    public void start() {
-        if (zooFactory != null) {
-            try {
-                zooFactory.startup(zooServer);
-            }
-            catch (IOException e) {
-                throw new UncheckedIOException(e);
-            }
-            catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-                throw new RuntimeException(e);
-            }
-        }
-
-        servers.stream().parallel().forEach(Server::startup);
-
-    }
-
-    @Override
-    public String getBootstrapServers() {
-        return String.join(",", bootstraps);
-    }
-
-    @Override
-    public Map<String, Object> getKafkaClientConfiguration() {
-        return clusterConfig.getConnectConfigForCluster(getBootstrapServers());
-    }
-
-    @Override
-    public void close() throws Exception {
-        try {
-            try {
-                servers.stream().parallel().forEach(Server::shutdown);
-                bootstraps.clear();
-            }
-            finally {
-                if (zooServer != null) {
-                    zooServer.shutdown(true);
-                }
-            }
-        }
-        finally {
-            if (tempDirectory.toFile().exists()) {
-                try (var s = Files.walk(tempDirectory)
-                        .sorted(Comparator.reverseOrder())
-                        .map(Path::toFile)) {
-                    s.forEach(File::delete);
-                }
-            }
-        }
-    }
-}
diff --git a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/KafkaCluster.java b/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/KafkaCluster.java
deleted file mode 100644
index e48d120f..00000000
--- a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/KafkaCluster.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Copyright Kroxylicious Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-
-package io.kroxylicious.proxy.testkafkacluster;
-
-import java.util.Map;
-
-public interface KafkaCluster extends AutoCloseable {
-    /**
-     * starts the cluster.
-     */
-    void start();
-
-    /**
-     * stops the cluster.
-     */
-    @Override
-    void close() throws Exception;
-
-    /**
-     * Gets the bootstrap servers for this cluster
-     * @return bootstrap servers
-     */
-    String getBootstrapServers();
-
-    /**
-     * Gets the kafka configuration for making connections to this cluster as required by the
-     * {@link org.apache.kafka.clients.admin.AdminClient}, {@link org.apache.kafka.clients.producer.Producer} etc.
-     * Details such the bootstrap and SASL configuration are provided automatically.
-     * The returned map is guaranteed to be mutable and is unique to the caller.
-     *
-     * @return mutable configuration map
-     */
-    Map<String, Object> getKafkaClientConfiguration();
-}
diff --git a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/KafkaClusterConfig.java b/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/KafkaClusterConfig.java
deleted file mode 100644
index 1523d2cd..00000000
--- a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/KafkaClusterConfig.java
+++ /dev/null
@@ -1,263 +0,0 @@
-/*
- * Copyright Kroxylicious Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-package io.kroxylicious.proxy.testkafkacluster;
-
-import java.io.IOException;
-import java.security.GeneralSecurityException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
-import java.util.TreeMap;
-import java.util.function.Supplier;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-import java.util.stream.Stream;
-
-import org.apache.kafka.clients.CommonClientConfigs;
-import org.apache.kafka.common.Uuid;
-import org.apache.kafka.common.config.SaslConfigs;
-import org.apache.kafka.common.security.auth.SecurityProtocol;
-import org.junit.jupiter.api.TestInfo;
-
-import io.kroxylicious.proxy.testkafkacluster.KafkaClusterConfig.KafkaEndpoints.Endpoint;
-
-import lombok.Builder;
-import lombok.Getter;
-import lombok.Singular;
-import lombok.ToString;
-
-@Builder(toBuilder = true)
-@Getter
-@ToString
-public class KafkaClusterConfig {
-
-    private TestInfo testInfo;
-    private KeytoolCertificateGenerator keytoolCertificateGenerator;
-
-    /**
-     * specifies the cluster execution mode.
-     */
-    private final KafkaClusterExecutionMode execMode;
-
-    /**
-     * if true, the cluster will be brought up in Kraft-mode
-     */
-    private final Boolean kraftMode;
-
-    /**
-     * name of SASL mechanism to be configured on kafka for the external listener, if null, anonymous communication
-     * will be used.
-     */
-    private final String saslMechanism;
-    private final String securityProtocol;
-    @Builder.Default
-    private Integer brokersNum = 1;
-
-    @Builder.Default
-    private Integer kraftControllers = 1;
-
-    private final String kafkaKraftClusterId = Uuid.randomUuid().toString();
-    /**
-     * The users and passwords to be configured into the server's JAAS configuration used for the external listener.
-     */
-    @Singular
-    private final Map<String, String> users;
-
-    public Stream<ConfigHolder> getBrokerConfigs(Supplier<KafkaEndpoints> endPointConfigSupplier, Supplier<Endpoint> zookeeperEndpointSupplier) {
-        List<ConfigHolder> properties = new ArrayList<>();
-        KafkaEndpoints kafkaEndpoints = endPointConfigSupplier.get();
-        for (int brokerNum = 0; brokerNum < brokersNum; brokerNum++) {
-            Properties server = new Properties();
-
-            server.put("broker.id", Integer.toString(brokerNum));
-
-            var interBrokerEndpoint = kafkaEndpoints.getInterBrokerEndpoint(brokerNum);
-            var clientEndpoint = kafkaEndpoints.getClientEndpoint(brokerNum);
-
-            // - EXTERNAL: used for communications to/from consumers/producers
-            // - INTERNAL: used for inter-broker communications (always no auth)
-            // - CONTROLLER: used for inter-broker controller communications (kraft - always no auth)
-
-            var externalListenerTransport = securityProtocol == null ? SecurityProtocol.PLAINTEXT.name() : securityProtocol;
-
-            var protocolMap = new TreeMap<String, String>();
-            var listeners = new TreeMap<String, String>();
-            var advertisedListeners = new TreeMap<String, String>();
-            protocolMap.put("EXTERNAL", externalListenerTransport);
-            listeners.put("EXTERNAL", clientEndpoint.getBind().toString());
-            advertisedListeners.put("EXTERNAL", clientEndpoint.getConnect().toString());
-
-            protocolMap.put("INTERNAL", SecurityProtocol.PLAINTEXT.name());
-            listeners.put("INTERNAL", interBrokerEndpoint.getBind().toString());
-            advertisedListeners.put("INTERNAL", interBrokerEndpoint.getConnect().toString());
-            server.put("inter.broker.listener.name", "INTERNAL");
-
-            if (isKraftMode()) {
-                server.put("node.id", Integer.toString(brokerNum)); // Required by Kafka 3.3 onwards.
-
-                var controllerEndpoint = kafkaEndpoints.getControllerEndpoint(brokerNum);
-                var quorumVoters = IntStream.range(0, kraftControllers)
-                        .mapToObj(b -> String.format("%d@%s", b, kafkaEndpoints.getControllerEndpoint(b).getConnect().toString())).collect(Collectors.joining(","));
-                server.put("controller.quorum.voters", quorumVoters);
-                server.put("controller.listener.names", "CONTROLLER");
-                protocolMap.put("CONTROLLER", SecurityProtocol.PLAINTEXT.name());
-
-                if (brokerNum == 0) {
-                    server.put("process.roles", "broker,controller");
-
-                    listeners.put("CONTROLLER", controllerEndpoint.getBind().toString());
-                }
-                else {
-                    server.put("process.roles", "broker");
-                }
-            }
-            else {
-                server.put("zookeeper.connect", String.format("%s:%d", zookeeperEndpointSupplier.get().getHost(), zookeeperEndpointSupplier.get().getPort()));
-                server.put("zookeeper.sasl.enabled", "false");
-                server.put("zookeeper.connection.timeout.ms", Long.toString(60000));
-            }
-
-            server.put("listener.security.protocol.map", protocolMap.entrySet().stream().map(e -> e.getKey() + ":" + e.getValue()).collect(Collectors.joining(",")));
-            server.put("listeners", listeners.entrySet().stream().map(e -> e.getKey() + ":" + e.getValue()).collect(Collectors.joining(",")));
-            server.put("advertised.listeners", advertisedListeners.entrySet().stream().map(e -> e.getKey() + ":" + e.getValue()).collect(Collectors.joining(",")));
-            server.put("early.start.listeners", advertisedListeners.keySet().stream().map(Object::toString).collect(Collectors.joining(",")));
-
-            if (saslMechanism != null) {
-                server.put("sasl.enabled.mechanisms", saslMechanism);
-
-                var saslPairs = new StringBuilder();
-
-                Optional.of(users).orElse(Map.of()).forEach((key, value) -> {
-                    saslPairs.append(String.format("user_%s", key));
-                    saslPairs.append("=");
-                    saslPairs.append(value);
-                    saslPairs.append(" ");
-                });
-
-                // TODO support other than PLAIN
-                String plainModuleConfig = String.format("org.apache.kafka.common.security.plain.PlainLoginModule required %s;", saslPairs);
-                server.put(String.format("listener.name.%s.plain.sasl.jaas.config", "EXTERNAL".toLowerCase()), plainModuleConfig);
-            }
-
-            if (securityProtocol != null && securityProtocol.contains("SSL")) {
-                if (keytoolCertificateGenerator == null) {
-                    throw new RuntimeException("keytoolCertificateGenerator needs to be initialized when calling KafkaClusterConfig");
-                }
-                try {
-                    keytoolCertificateGenerator.generateSelfSignedCertificateEntry("test@redhat.com", clientEndpoint.getConnect().getHost(), "KI", "RedHat", null, null,
-                            "US");
-                }
-                catch (GeneralSecurityException | IOException e) {
-                    throw new RuntimeException(e);
-                }
-                server.put("ssl.client.auth", "required");
-                server.put("ssl.truststore.location", keytoolCertificateGenerator.getCertLocation());
-                server.put("ssl.truststore.password", keytoolCertificateGenerator.getPassword());
-                server.put("ssl.keystore.location", keytoolCertificateGenerator.getCertLocation());
-                server.put("ssl.keystore.password", keytoolCertificateGenerator.getPassword());
-                server.put("ssl.key.password", keytoolCertificateGenerator.getPassword());
-            }
-
-            server.put("offsets.topic.replication.factor", Integer.toString(1));
-            // 1 partition for the __consumer_offsets_ topic should be enough
-            server.put("offsets.topic.num.partitions", Integer.toString(1));
-            // Disable delay during every re-balance
-            server.put("group.initial.rebalance.delay.ms", Integer.toString(0));
-
-            properties.add(new ConfigHolder(server, clientEndpoint.getConnect().getPort(),
-                    String.format("%s:%d", clientEndpoint.getConnect().getHost(), clientEndpoint.getConnect().getPort()), brokerNum, kafkaKraftClusterId));
-        }
-
-        return properties.stream();
-    }
-
-    protected Map<String, Object> getConnectConfigForCluster(String bootstrapServers) {
-        Map<String, Object> kafkaConfig = new HashMap<>();
-        String saslMechanism = getSaslMechanism();
-        String securityProtocol = getSecurityProtocol();
-
-        if (securityProtocol != null) {
-            kafkaConfig.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol);
-
-            if (securityProtocol.contains("SSL")) {
-                kafkaConfig.put("ssl.truststore.location", keytoolCertificateGenerator.getCertLocation());
-                kafkaConfig.put("ssl.truststore.password", keytoolCertificateGenerator.getPassword());
-                if (securityProtocol.equals(SecurityProtocol.SSL.name())) {
-                    kafkaConfig.put("ssl.keystore.location", keytoolCertificateGenerator.getCertLocation());
-                    kafkaConfig.put("ssl.keystore.password", keytoolCertificateGenerator.getPassword());
-                    kafkaConfig.put("ssl.key.password", keytoolCertificateGenerator.getPassword());
-                }
-            }
-        }
-
-        if (saslMechanism != null) {
-            if (securityProtocol == null) {
-                kafkaConfig.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SASL_PLAINTEXT.name());
-            }
-            kafkaConfig.put(SaslConfigs.SASL_MECHANISM, saslMechanism);
-
-            Map<String, String> users = getUsers();
-            if ("PLAIN".equals(saslMechanism) && !users.isEmpty()) {
-                var first = users.entrySet().iterator().next();
-                kafkaConfig.put(SaslConfigs.SASL_JAAS_CONFIG,
-                        String.format("org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s\" password=\"%s\";",
-                                first.getKey(), first.getValue()));
-            }
-            else {
-                throw new IllegalStateException(String.format("unsupported SASL mechanism %s", saslMechanism));
-            }
-        }
-
-        kafkaConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
-
-        return kafkaConfig;
-    }
-
-    public boolean isKraftMode() {
-        return this.getKraftMode() == null || this.getKraftMode();
-    }
-
-    @Builder
-    @Getter
-    public static class ConfigHolder {
-        private final Properties properties;
-        private final Integer externalPort;
-        private final String endpoint;
-        private final int brokerNum;
-        private final String kafkaKraftClusterId;
-    }
-
-    protected interface KafkaEndpoints {
-
-        @Builder
-        @Getter
-        class EndpointPair {
-            private final Endpoint bind;
-            private final Endpoint connect;
-        }
-
-        @Builder
-        @Getter
-        class Endpoint {
-            private final String host;
-            private final int port;
-
-            @Override
-            public String toString() {
-                return String.format("//%s:%d", host, port);
-            }
-        }
-
-        EndpointPair getInterBrokerEndpoint(int brokerId);
-
-        EndpointPair getControllerEndpoint(int brokerId);
-
-        EndpointPair getClientEndpoint(int brokerId);
-    }
-}
diff --git a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/KafkaClusterExecutionMode.java b/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/KafkaClusterExecutionMode.java
deleted file mode 100644
index 3477dfc3..00000000
--- a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/KafkaClusterExecutionMode.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * Copyright Kroxylicious Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-
-package io.kroxylicious.proxy.testkafkacluster;
-
-public enum KafkaClusterExecutionMode {
-    /**
-     * Kafka/Zookeeper will be run within the same JVM as the caller
-     */
-    IN_VM,
-    /**
-     * Kafka/Zookeeper will be run in containers
-     */
-    CONTAINER;
-
-    public static KafkaClusterExecutionMode convertClusterExecutionMode(String mode, KafkaClusterExecutionMode defaultMode) {
-        try {
-            if (mode == null) {
-                return defaultMode;
-            }
-            return valueOf(KafkaClusterExecutionMode.class, mode.toUpperCase());
-        }
-        catch (IllegalArgumentException e) {
-            return defaultMode;
-        }
-    }
-}
diff --git a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/KafkaClusterFactory.java b/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/KafkaClusterFactory.java
deleted file mode 100644
index fc101ee6..00000000
--- a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/KafkaClusterFactory.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * Copyright Kroxylicious Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-package io.kroxylicious.proxy.testkafkacluster;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class KafkaClusterFactory {
-    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaClusterFactory.class);
-
-    /**
-     * environment variable specifying execution mode, IN_VM or CONTAINER.
-     */
-    public static final String TEST_CLUSTER_EXECUTION_MODE = "TEST_CLUSTER_EXECUTION_MODE";
-
-    /**
-     * environment variable specifying kraft mode, true or false.
-     */
-    public static final String TEST_CLUSTER_KRAFT_MODE = "TEST_CLUSTER_KRAFT_MODE";
-
-    public static KafkaCluster create(KafkaClusterConfig clusterConfig) {
-        if (clusterConfig == null) {
-            throw new NullPointerException();
-        }
-
-        var clusterMode = KafkaClusterExecutionMode.convertClusterExecutionMode(System.getenv().get(TEST_CLUSTER_EXECUTION_MODE), KafkaClusterExecutionMode.IN_VM);
-        var kraftMode = convertClusterKraftMode(System.getenv().get(TEST_CLUSTER_KRAFT_MODE), true);
-
-        var builder = clusterConfig.toBuilder();
-
-        if (clusterConfig.getExecMode() == null) {
-            builder.execMode(clusterMode);
-        }
-
-        if (clusterConfig.getKraftMode() == null) {
-            builder.kraftMode(kraftMode);
-        }
-
-        var actual = builder.build();
-        LOGGER.info("Test cluster : {}", actual);
-
-        if (actual.getExecMode() == KafkaClusterExecutionMode.IN_VM) {
-            return new InVMKafkaCluster(actual);
-        }
-        else {
-            return new ContainerBasedKafkaCluster(actual);
-        }
-    }
-
-    private static boolean convertClusterKraftMode(String mode, boolean defaultMode) {
-        if (mode == null) {
-            return defaultMode;
-        }
-        return Boolean.parseBoolean(mode);
-    }
-
-}
diff --git a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/KeytoolCertificateGenerator.java b/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/KeytoolCertificateGenerator.java
deleted file mode 100644
index 90c2bdf6..00000000
--- a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/KeytoolCertificateGenerator.java
+++ /dev/null
@@ -1,123 +0,0 @@
-/*
- * Copyright Kroxylicious Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-package io.kroxylicious.proxy.testkafkacluster;
-
-import java.io.BufferedReader;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.security.GeneralSecurityException;
-import java.security.KeyStore;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.UUID;
-import java.util.stream.Collectors;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class KeytoolCertificateGenerator {
-    private String password;
-    private final Path certFilePath;
-    private final Logger log = LoggerFactory.getLogger(KeytoolCertificateGenerator.class.getName());
-
-    public KeytoolCertificateGenerator() throws IOException {
-        Path certsDirectory = Files.createTempDirectory("kproxy");
-        certsDirectory.toFile().deleteOnExit();
-        certFilePath = Paths.get(certsDirectory.toAbsolutePath().toString(), "kafka.jks");
-    }
-
-    public String getCertLocation() {
-        return certFilePath.toAbsolutePath().toString();
-    }
-
-    public String getPassword() {
-        if (password == null) {
-            password = UUID.randomUUID().toString().replace("-", "");
-        }
-        return password;
-    }
-
-    public boolean canGenerateWildcardSAN() {
-        return Runtime.version().feature() >= 17;
-    }
-
-    public void generateSelfSignedCertificateEntry(String email, String domain, String organizationUnit,
-                                                   String organization, String city, String state,
-                                                   String country)
-            throws GeneralSecurityException, IOException {
-
-        KeyStore keyStore = KeyStore.getInstance("JKS");
-        if (certFilePath.toFile().exists()) {
-            keyStore.load(new FileInputStream(certFilePath.toFile()), getPassword().toCharArray());
-
-            if (keyStore.containsAlias(domain)) {
-                keyStore.deleteEntry(domain);
-                keyStore.store(new FileOutputStream(certFilePath.toFile()), getPassword().toCharArray());
-            }
-        }
-
-        final List<String> commandParameters = new ArrayList<>(List.of("keytool", "-genkey"));
-        commandParameters.addAll(List.of("-alias", domain));
-        commandParameters.addAll(List.of("-keyalg", "RSA"));
-        commandParameters.addAll(List.of("-keysize", "2048"));
-        commandParameters.addAll(List.of("-sigalg", "SHA256withRSA"));
-        commandParameters.addAll(List.of("-storetype", "JKS"));
-        commandParameters.addAll(List.of("-keystore", getCertLocation()));
-        commandParameters.addAll(List.of("-storepass", getPassword()));
-        commandParameters.addAll(List.of("-keypass", getPassword()));
-        commandParameters.addAll(
-                List.of("-dname", getDomainName(email, domain, organizationUnit, organization, city, state, country)));
-        commandParameters.addAll(List.of("-validity", "365"));
-        commandParameters.addAll(List.of("-deststoretype", "pkcs12"));
-        commandParameters.addAll(List.of("-storetype", "JKS"));
-        if (canGenerateWildcardSAN() && !isWildcardDomain(domain)) {
-            commandParameters.addAll(getSAN(domain));
-        }
-        ProcessBuilder keytool = new ProcessBuilder().command(commandParameters);
-
-        final Process process = keytool.start();
-        try {
-            process.waitFor();
-        }
-        catch (InterruptedException e) {
-            throw new IOException("Keytool execution error");
-        }
-
-        log.info("Generating certificate using `keytool` using command: " + process.info() + ", parameters: " +
-                commandParameters);
-
-        if (process.exitValue() > 0) {
-            final String processError = (new BufferedReader(new InputStreamReader(process.getErrorStream()))).lines()
-                    .collect(Collectors.joining(" \\ "));
-            final String processOutput = (new BufferedReader(new InputStreamReader(process.getInputStream()))).lines()
-                    .collect(Collectors.joining(" \\ "));
-            log.warn("Error generating certificate, error output: " + processError + ", normal output: " +
-                    processOutput + ", commandline parameters: " + commandParameters);
-            throw new IOException(
-                    "Keytool execution error: '" + processError + "', output: '" + processOutput + "'" + ", commandline parameters: " + commandParameters);
-        }
-    }
-
-    private boolean isWildcardDomain(String domain) {
-        return domain.startsWith("*.");
-    }
-
-    private String getDomainName(String email, String domain, String organizationUnit, String organization, String city,
-                                 String state, String country) {
-        // keytool doesn't allow for a domain with wildcard in SAN extension, so it has to go directly to CN
-        return "CN=" + domain + ", OU=" + organizationUnit + ", O=" + organization + ", L=" + city + ", ST=" + state +
-                ", C=" + country + ", EMAILADDRESS=" + email;
-    }
-
-    private List<String> getSAN(String domain) {
-        return List.of("-ext", "SAN=dns:" + domain);
-    }
-}
diff --git a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/Utils.java b/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/Utils.java
deleted file mode 100644
index 1232de59..00000000
--- a/integrationtests/src/main/java/io/kroxylicious/proxy/testkafkacluster/Utils.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * Copyright Kroxylicious Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-
-package io.kroxylicious.proxy.testkafkacluster;
-
-import java.io.IOException;
-import java.io.UncheckedIOException;
-import java.net.ServerSocket;
-import java.util.stream.Stream;
-
-public class Utils {
-    /**
-     * Pre-allocate 1 or more ephemeral ports which are available for use.
-     *
-     * @param num number of ports to pre-allocate
-     * @return list of ephemeral ports
-     */
-    public static Stream<Integer> preAllocateListeningPorts(int num) {
-        // Uses recursive algorithm to avoid the risk of returning a duplicate ephemeral port.
-        if (num < 1) {
-            return Stream.of();
-        }
-        try (var serverSocket = new ServerSocket(0)) {
-            serverSocket.setReuseAddress(true);
-            int localPort = serverSocket.getLocalPort();
-            return Stream.concat(Stream.of(localPort), preAllocateListeningPorts(num - 1));
-        }
-        catch (IOException e) {
-            throw new UncheckedIOException(e);
-        }
-    }
-}
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/KafkaClusterIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/KafkaClusterIT.java
deleted file mode 100644
index 3f06bd83..00000000
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/KafkaClusterIT.java
+++ /dev/null
@@ -1,248 +0,0 @@
-/*
- * Copyright Kroxylicious Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-package io.kroxylicious.proxy;
-
-import java.io.IOException;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.time.Duration;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.kafka.clients.admin.AdminClient;
-import org.apache.kafka.clients.admin.DescribeClusterResult;
-import org.apache.kafka.clients.admin.KafkaAdminClient;
-import org.apache.kafka.clients.admin.NewTopic;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.serialization.StringDeserializer;
-import org.apache.kafka.common.serialization.StringSerializer;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInfo;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import io.kroxylicious.proxy.testkafkacluster.ContainerBasedKafkaCluster;
-import io.kroxylicious.proxy.testkafkacluster.KafkaCluster;
-import io.kroxylicious.proxy.testkafkacluster.KafkaClusterConfig;
-import io.kroxylicious.proxy.testkafkacluster.KafkaClusterFactory;
-import io.kroxylicious.proxy.testkafkacluster.KeytoolCertificateGenerator;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assumptions.assumeTrue;
-
-/**
- * Test case that simply exercises the ability to control the kafka cluster from the test.
- * It intentional that this test does not involve the proxy.
- */
-public class KafkaClusterIT {
-
-    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaClusterIT.class);
-    private TestInfo testInfo;
-    private KeytoolCertificateGenerator keytoolCertificateGenerator;
-
-    @Test
-    public void kafkaClusterKraftMode() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder()
-                .testInfo(testInfo)
-                .kraftMode(true)
-                .build())) {
-            cluster.start();
-            verifyRecordRoundTrip(1, cluster);
-        }
-    }
-
-    @Test
-    public void kafkaClusterZookeeperMode() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder()
-                .testInfo(testInfo)
-                .kraftMode(false)
-                .build())) {
-            cluster.start();
-            verifyRecordRoundTrip(1, cluster);
-        }
-    }
-
-    @Test
-    public void kafkaTwoNodeClusterKraftMode() throws Exception {
-        int brokersNum = 2;
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder()
-                .testInfo(testInfo)
-                .brokersNum(brokersNum)
-                .kraftMode(true)
-                .build())) {
-            assumeTrue(cluster instanceof ContainerBasedKafkaCluster, "KAFKA-14287: kraft timing out on shutdown in multinode case");
-            cluster.start();
-            verifyRecordRoundTrip(brokersNum, cluster);
-        }
-    }
-
-    @Test
-    public void kafkaTwoNodeClusterZookeeperMode() throws Exception {
-        int brokersNum = 2;
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder()
-                .testInfo(testInfo)
-                .brokersNum(brokersNum)
-                .kraftMode(false)
-                .build())) {
-            cluster.start();
-            verifyRecordRoundTrip(brokersNum, cluster);
-        }
-    }
-
-    @Test
-    public void kafkaClusterKraftModeWithAuth() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder()
-                .kraftMode(true)
-                .testInfo(testInfo)
-                .securityProtocol("SASL_PLAINTEXT")
-                .saslMechanism("PLAIN")
-                .user("guest", "guest")
-                .build())) {
-            cluster.start();
-            verifyRecordRoundTrip(1, cluster);
-        }
-    }
-
-    @Test
-    public void kafkaClusterZookeeperModeWithAuth() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder()
-                .testInfo(testInfo)
-                .kraftMode(false)
-                .securityProtocol("SASL_PLAINTEXT")
-                .saslMechanism("PLAIN")
-                .user("guest", "guest")
-                .build())) {
-            cluster.start();
-            verifyRecordRoundTrip(1, cluster);
-        }
-    }
-
-    @Test
-    public void kafkaClusterKraftModeSASL_SSL() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder()
-                .testInfo(testInfo)
-                .keytoolCertificateGenerator(keytoolCertificateGenerator)
-                .kraftMode(true)
-                .securityProtocol("SASL_SSL")
-                .saslMechanism("PLAIN")
-                .user("guest", "guest")
-                .build())) {
-            cluster.start();
-            verifyRecordRoundTrip(1, cluster);
-        }
-    }
-
-    @Test
-    public void kafkaClusterKraftModeSSL() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder()
-                .testInfo(testInfo)
-                .keytoolCertificateGenerator(keytoolCertificateGenerator)
-                .kraftMode(true)
-                .securityProtocol("SSL")
-                .build())) {
-            cluster.start();
-            verifyRecordRoundTrip(1, cluster);
-        }
-    }
-
-    @Test
-    public void kafkaClusterZookeeperModeSASL_SSL() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder()
-                .testInfo(testInfo)
-                .keytoolCertificateGenerator(keytoolCertificateGenerator)
-                .kraftMode(false)
-                .securityProtocol("SASL_SSL")
-                .saslMechanism("PLAIN")
-                .user("guest", "guest")
-                .build())) {
-            cluster.start();
-            verifyRecordRoundTrip(1, cluster);
-        }
-    }
-
-    @Test
-    public void kafkaClusterZookeeperModeSSL() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder()
-                .testInfo(testInfo)
-                .keytoolCertificateGenerator(keytoolCertificateGenerator)
-                .kraftMode(false)
-                .securityProtocol("SSL")
-                .build())) {
-            cluster.start();
-            verifyRecordRoundTrip(1, cluster);
-        }
-    }
-
-    private void verifyRecordRoundTrip(int expected, KafkaCluster cluster) throws Exception {
-        var topic = "TOPIC_1";
-        var message = "Hello, world!";
-
-        try (var admin = KafkaAdminClient.create(cluster.getKafkaClientConfiguration())) {
-            assertEquals(expected, getActualNumberOfBrokers(admin));
-            var rf = (short) Math.min(1, Math.max(expected, 3));
-            createTopic(admin, topic, rf);
-        }
-
-        produce(cluster, topic, message);
-        consume(cluster, topic, "Hello, world!");
-    }
-
-    private void produce(KafkaCluster cluster, String topic, String message) throws Exception {
-        Map<String, Object> config = cluster.getKafkaClientConfiguration();
-        config.putAll(Map.<String, Object> of(
-                ProducerConfig.CLIENT_ID_CONFIG, "myclient",
-                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
-                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
-                ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 3_600_000));
-        try (var producer = new KafkaProducer<String, String>(config)) {
-            producer.send(new ProducerRecord<>(topic, "my-key", message)).get();
-        }
-    }
-
-    private void consume(KafkaCluster cluster, String topic, String message) throws Exception {
-        Map<String, Object> config = cluster.getKafkaClientConfiguration();
-        config.putAll(Map.of(
-                ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
-                ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
-                ConsumerConfig.GROUP_ID_CONFIG, "my-group-id",
-                ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"));
-        try (var consumer = new KafkaConsumer<String, String>(config)) {
-            consumer.subscribe(Set.of(topic));
-            var records = consumer.poll(Duration.ofSeconds(10));
-            assertEquals(1, records.count());
-            assertEquals(message, records.iterator().next().value());
-            consumer.unsubscribe();
-        }
-    }
-
-    private int getActualNumberOfBrokers(AdminClient admin) throws Exception {
-        DescribeClusterResult describeClusterResult = admin.describeCluster();
-        return describeClusterResult.nodes().get().size();
-    }
-
-    private void createTopic(AdminClient admin1, String topic, short replicationFactor) throws Exception {
-        admin1.createTopics(List.of(new NewTopic(topic, 1, replicationFactor))).all().get();
-    }
-
-    @BeforeEach
-    void before(TestInfo testInfo) throws IOException {
-        this.testInfo = testInfo;
-        this.keytoolCertificateGenerator = new KeytoolCertificateGenerator();
-    }
-
-    @AfterEach
-    void after() {
-        Path filePath = Paths.get(keytoolCertificateGenerator.getCertLocation());
-        filePath.toFile().deleteOnExit();
-    }
-}
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyConfigBuilder.java b/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyConfigBuilder.java
index 5ed61932..6b7c64b5 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyConfigBuilder.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyConfigBuilder.java
@@ -41,7 +41,7 @@ public class KroxyConfigBuilder {
     public record MicrometerConfig(@JsonInclude(NON_EMPTY) List<String> binders,
                                    @JsonInclude(NON_NULL) String configurationHookClass,
                                    @JsonInclude(NON_EMPTY) Map<String, String> commonTags) {
-        MicrometerConfig withTag(String key, String value){
+        public MicrometerConfig withTag(String key, String value) {
             Map<String, String> tags = commonTags != null ? new HashMap<>(commonTags) : new HashMap<>();
             tags.put(key, value);
             return new MicrometerConfig(this.binders, this.configurationHookClass, tags);
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
index 58aa59b7..c23e685b 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
@@ -7,6 +7,7 @@ package io.kroxylicious.proxy;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
@@ -29,22 +30,20 @@ import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
 
-import io.kroxylicious.proxy.config.ConfigParser;
-import io.kroxylicious.proxy.config.Configuration;
 import io.kroxylicious.proxy.internal.filter.ByteBufferTransformation;
-import io.kroxylicious.proxy.testkafkacluster.KafkaClusterConfig;
-import io.kroxylicious.proxy.testkafkacluster.KafkaClusterFactory;
-import io.kroxylicious.proxy.testkafkacluster.KeytoolCertificateGenerator;
+import io.kroxylicious.testing.kafka.api.KafkaCluster;
+import io.kroxylicious.testing.kafka.common.KeytoolCertificateGenerator;
+import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
 
+import static io.kroxylicious.proxy.Utils.startProxy;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertArrayEquals;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.*;
 
+@ExtendWith(KafkaClusterExtension.class)
 public class KrpcFilterIT {
 
     private static final String TOPIC_1 = "my-test-topic";
@@ -54,13 +53,8 @@ public class KrpcFilterIT {
             (byte) 0x64, (byte) 0x67, (byte) 0x61, (byte) 0x59, (byte) 0x16 };
     private static final byte[] TOPIC_2_CIPHERTEXT = { (byte) 0xffffffa7, (byte) 0xffffffc4, (byte) 0xffffffcb, (byte) 0xffffffcb, (byte) 0xffffffce, (byte) 0xffffff8b,
             (byte) 0x7f, (byte) 0xffffffd6, (byte) 0xffffffce, (byte) 0xffffffd1, (byte) 0xffffffcb, (byte) 0xffffffc3, (byte) 0xffffff80 };
-
-    private TestInfo testInfo;
-
-    @BeforeEach
-    public void beforeEach(TestInfo testInfo) {
-        this.testInfo = testInfo;
-    }
+    @TempDir
+    private Path certsDirectory;
 
     @BeforeAll
     public static void checkReversibleEncryption() {
@@ -113,20 +107,23 @@ public class KrpcFilterIT {
     }
 
     @Test
-    public void shouldPassThroughRecordUnchanged() throws Exception {
+    public void shouldPassThroughRecordUnchanged(KafkaCluster cluster, Admin admin) throws Exception {
         String proxyAddress = "localhost:9192";
 
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
-            cluster.start();
+        admin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1))).all().get();
 
-            String bootstrapServers = cluster.getBootstrapServers();
-            try (var admin = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers))) {
-                admin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1))).all().get();
-            }
-            String config = baseConfigBuilder(proxyAddress, bootstrapServers).build();
-
-            var proxy = startProxy(config);
+        String config = """
+                proxy:
+                  address: %s
+                clusters:
+                  demo:
+                    bootstrap_servers: %s
+                filters:
+                - type: ApiVersions
+                - type: BrokerAddress
+                """.formatted(proxyAddress, cluster.getBootstrapServers());
 
+        try (var proxy = startProxy(config)) {
             try (var producer = new KafkaProducer<String, String>(Map.of(
                     ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress,
                     ProducerConfig.CLIENT_ID_CONFIG, "shouldPassThroughRecordUnchanged",
@@ -148,182 +145,157 @@ public class KrpcFilterIT {
                 assertEquals(1, records.count());
                 assertEquals("Hello, world!", records.iterator().next().value());
             }
-
-            // shutdown the proxy
-            proxy.shutdown();
-
         }
-
     }
 
     @Test
-    public void shouldModifyProduceMessage() throws Exception {
+    public void shouldModifyProduceMessage(KafkaCluster cluster, Admin admin) throws Exception {
         String proxyAddress = "localhost:9192";
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
-            cluster.start();
-            try (var admin = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.getBootstrapServers()))) {
-                admin.createTopics(List.of(
-                        new NewTopic(TOPIC_1, 1, (short) 1),
-                        new NewTopic(TOPIC_2, 1, (short) 1))).all().get();
-            }
-
-            String config = baseConfigBuilder(proxyAddress, cluster.getBootstrapServers())
-                    .addFilter("ProduceRequestTransformation", "transformation", TestEncoder.class.getName())
-                    .build();
-
-            var proxy = startProxy(config);
-            try {
-                try (var producer = new KafkaProducer<String, String>(Map.of(
-                        ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress,
-                        ProducerConfig.CLIENT_ID_CONFIG, "shouldModifyProduceMessage",
-                        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
-                        ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
-                        ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 3_600_000))) {
-                    producer.send(new ProducerRecord<>(TOPIC_1, "my-key", PLAINTEXT)).get();
-                    producer.send(new ProducerRecord<>(TOPIC_2, "my-key", PLAINTEXT)).get();
-                    producer.flush();
-                }
-
-                ConsumerRecords<String, byte[]> records1;
-                ConsumerRecords<String, byte[]> records2;
-                try (var consumer = new KafkaConsumer<String, byte[]>(Map.of(
-                        ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress,
-                        ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
-                        ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class,
-                        ConsumerConfig.GROUP_ID_CONFIG, "my-group-id",
-                        ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"))) {
-                    consumer.subscribe(Set.of(TOPIC_1));
-                    records1 = consumer.poll(Duration.ofSeconds(10));
-                    consumer.subscribe(Set.of(TOPIC_2));
-                    records2 = consumer.poll(Duration.ofSeconds(10));
-                }
-
-                assertEquals(1, records1.count());
-                assertArrayEquals(TOPIC_1_CIPHERTEXT, records1.iterator().next().value());
-                assertEquals(1, records2.count());
-                assertArrayEquals(TOPIC_2_CIPHERTEXT, records2.iterator().next().value());
 
+        admin.createTopics(List.of(
+                new NewTopic(TOPIC_1, 1, (short) 1),
+                new NewTopic(TOPIC_2, 1, (short) 1))).all().get();
+
+        String config = """
+                proxy:
+                  address: %s
+                clusters:
+                  demo:
+                    bootstrap_servers: %s
+                filters:
+                - type: ApiVersions
+                - type: BrokerAddress
+                - type: ProduceRequestTransformation
+                  config:
+                    transformation: %s
+                """.formatted(proxyAddress, cluster.getBootstrapServers(), TestEncoder.class.getName());
+
+        try (var proxy = startProxy(config)) {
+            try (var producer = new KafkaProducer<String, String>(Map.of(
+                    ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress,
+                    ProducerConfig.CLIENT_ID_CONFIG, "shouldModifyProduceMessage",
+                    ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
+                    ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
+                    ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 3_600_000))) {
+                producer.send(new ProducerRecord<>(TOPIC_1, "my-key", PLAINTEXT)).get();
+                producer.send(new ProducerRecord<>(TOPIC_2, "my-key", PLAINTEXT)).get();
+                producer.flush();
             }
-            finally {
-                // shutdown the proxy
-                proxy.shutdown();
+
+            ConsumerRecords<String, byte[]> records1;
+            ConsumerRecords<String, byte[]> records2;
+            try (var consumer = new KafkaConsumer<String, byte[]>(Map.of(
+                    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress,
+                    ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+                    ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class,
+                    ConsumerConfig.GROUP_ID_CONFIG, "my-group-id",
+                    ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"))) {
+                consumer.subscribe(Set.of(TOPIC_1));
+                records1 = consumer.poll(Duration.ofSeconds(10));
+                consumer.subscribe(Set.of(TOPIC_2));
+                records2 = consumer.poll(Duration.ofSeconds(10));
             }
 
+            assertEquals(1, records1.count());
+            assertArrayEquals(TOPIC_1_CIPHERTEXT, records1.iterator().next().value());
+            assertEquals(1, records2.count());
+            assertArrayEquals(TOPIC_2_CIPHERTEXT, records2.iterator().next().value());
         }
     }
 
     @Test
-    public void shouldModifyFetchMessage() throws Exception {
+    public void shouldModifyFetchMessage(KafkaCluster cluster, Admin admin) throws Exception {
         String proxyAddress = "localhost:9192";
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
-            cluster.start();
-            try (var admin = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.getBootstrapServers()))) {
-                admin.createTopics(List.of(
-                        new NewTopic(TOPIC_1, 1, (short) 1),
-                        new NewTopic(TOPIC_2, 1, (short) 1))).all().get();
-            }
-
-            String config = baseConfigBuilder(proxyAddress, cluster.getBootstrapServers())
-                    .addFilter("FetchResponseTransformation", "transformation", TestDecoder.class.getName())
-                    .build();
-
-            var proxy = startProxy(config);
 
-            try {
-
-                try (var producer = new KafkaProducer<String, byte[]>(Map.of(
-                        ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress,
-                        ProducerConfig.CLIENT_ID_CONFIG, "shouldModifyFetchMessage",
-                        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
-                        ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class,
-                        ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 3_600_000))) {
-
-                    producer.send(new ProducerRecord<>(TOPIC_1, "my-key", TOPIC_1_CIPHERTEXT)).get();
-                    producer.send(new ProducerRecord<>(TOPIC_2, "my-key", TOPIC_2_CIPHERTEXT)).get();
-                }
+        admin.createTopics(List.of(
+                new NewTopic(TOPIC_1, 1, (short) 1),
+                new NewTopic(TOPIC_2, 1, (short) 1))).all().get();
+
+        String config = """
+                proxy:
+                  address: %s
+                clusters:
+                  demo:
+                    bootstrap_servers: %s
+                filters:
+                - type: ApiVersions
+                - type: BrokerAddress
+                - type: FetchResponseTransformation
+                  config:
+                    transformation: %s
+                """.formatted(proxyAddress, cluster.getBootstrapServers(), TestDecoder.class.getName());
+
+        try (var proxy = startProxy(config)) {
+            try (var producer = new KafkaProducer<String, byte[]>(Map.of(
+                    ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress,
+                    ProducerConfig.CLIENT_ID_CONFIG, "shouldModifyFetchMessage",
+                    ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
+                    ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class,
+                    ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 3_600_000))) {
 
-                ConsumerRecords<String, String> records1;
-                ConsumerRecords<String, String> records2;
-                try (var consumer = new KafkaConsumer<String, String>(Map.of(
-                        ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress,
-                        ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
-                        ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
-                        ConsumerConfig.GROUP_ID_CONFIG, "my-group-id",
-                        ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"))) {
-                    consumer.subscribe(Set.of(TOPIC_1));
+                producer.send(new ProducerRecord<>(TOPIC_1, "my-key", TOPIC_1_CIPHERTEXT)).get();
+                producer.send(new ProducerRecord<>(TOPIC_2, "my-key", TOPIC_2_CIPHERTEXT)).get();
+            }
 
-                    records1 = consumer.poll(Duration.ofSeconds(100));
+            ConsumerRecords<String, String> records1;
+            ConsumerRecords<String, String> records2;
+            try (var consumer = new KafkaConsumer<String, String>(Map.of(
+                    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress,
+                    ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+                    ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+                    ConsumerConfig.GROUP_ID_CONFIG, "my-group-id",
+                    ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"))) {
+                consumer.subscribe(Set.of(TOPIC_1));
 
-                    consumer.subscribe(Set.of(TOPIC_2));
+                records1 = consumer.poll(Duration.ofSeconds(100));
 
-                    records2 = consumer.poll(Duration.ofSeconds(100));
-                }
-                assertEquals(1, records1.count());
-                assertEquals(1, records2.count());
-                assertEquals(List.of(PLAINTEXT, PLAINTEXT),
-                        List.of(records1.iterator().next().value(),
-                                records2.iterator().next().value()));
+                consumer.subscribe(Set.of(TOPIC_2));
 
+                records2 = consumer.poll(Duration.ofSeconds(100));
             }
-            finally {
-                // shutdown the proxy
-                proxy.shutdown();
-            }
-
+            assertEquals(1, records1.count());
+            assertEquals(1, records2.count());
+            assertEquals(List.of(PLAINTEXT, PLAINTEXT),
+                    List.of(records1.iterator().next().value(),
+                            records2.iterator().next().value()));
         }
-
     }
 
     @Test
-    public void proxySslClusterPlain() throws Exception {
+    public void proxySslClusterPlain(KafkaCluster cluster) throws Exception {
         String proxyAddress = "localhost:9192";
 
-        var certificateGenerator = new KeytoolCertificateGenerator();
-        certificateGenerator.generateSelfSignedCertificateEntry("test@redhat.com", "localhost", "KI", "RedHat", null, null, "US");
-
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
-            cluster.start();
-
-            String config = baseConfigBuilder(proxyAddress, cluster.getBootstrapServers())
-                    .withKeyStoreConfig(certificateGenerator.getCertLocation(), certificateGenerator.getPassword())
-                    .build();
-
-            var proxy = startProxy(config);
-            try {
-                try (var admin = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress,
-                        CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name,
-                        SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, certificateGenerator.getCertLocation(),
-                        SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, certificateGenerator.getPassword()))) {
-                    // do some work to ensure connection is opened
-                    admin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1))).all().get();
-                    var connectionsMetric = admin.metrics().entrySet().stream().filter(metricNameEntry -> "connections".equals(metricNameEntry.getKey().name()))
-                            .findFirst();
-                    assertTrue(connectionsMetric.isPresent());
-                    var protocol = connectionsMetric.get().getKey().tags().get("protocol");
-                    assertThat(protocol).startsWith("TLS");
-                }
-            }
-            finally {
-                // shutdown the proxy
-                proxy.shutdown();
+        var brokerCertificateGenerator = new KeytoolCertificateGenerator();
+        brokerCertificateGenerator.generateSelfSignedCertificateEntry("test@redhat.com", "localhost", "KI", "RedHat", null, null, "US");
+        Path clientTrustStore = certsDirectory.resolve("kafka.truststore.jks");
+        brokerCertificateGenerator.generateTrustStore(brokerCertificateGenerator.getCertFilePath(), "client",
+                clientTrustStore.toAbsolutePath().toString());
+        String config = """
+                proxy:
+                  address: %s
+                  keyStoreFile: %s
+                  keyPassword: %s
+                clusters:
+                  demo:
+                    bootstrap_servers: %s
+                filters:
+                - type: ApiVersions
+                - type: BrokerAddress
+                """.formatted(proxyAddress, brokerCertificateGenerator.getKeyStoreLocation(), brokerCertificateGenerator.getPassword(), cluster.getBootstrapServers());
+
+        try (var proxy = startProxy(config)) {
+            try (var admin = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress,
+                    CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name,
+                    SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, clientTrustStore.toAbsolutePath().toString(),
+                    SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, brokerCertificateGenerator.getPassword()))) {
+                // do some work to ensure connection is opened
+                admin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1))).all().get();
+                var connectionsMetric = admin.metrics().entrySet().stream().filter(metricNameEntry -> "connections".equals(metricNameEntry.getKey().name()))
+                        .findFirst();
+                assertTrue(connectionsMetric.isPresent());
+                var protocol = connectionsMetric.get().getKey().tags().get("protocol");
+                assertThat(protocol).startsWith("TLS");
             }
         }
     }
-
-    private static KroxyConfigBuilder baseConfigBuilder(String proxyAddress, String bootstrapServers) {
-        return new KroxyConfigBuilder(proxyAddress)
-                .withDefaultCluster(bootstrapServers)
-                .addFilter("ApiVersions")
-                .addFilter("BrokerAddress");
-    }
-
-    private KafkaProxy startProxy(String config) throws InterruptedException {
-        Configuration proxyConfig = new ConfigParser().parseConfiguration(config);
-
-        KafkaProxy kafkaProxy = new KafkaProxy(proxyConfig);
-        kafkaProxy.startup();
-
-        return kafkaProxy;
-    }
-
 }
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java
index 8f7f79ed..49477a83 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java
@@ -13,50 +13,34 @@ import java.util.Arrays;
 import java.util.Random;
 import java.util.regex.Pattern;
 
-import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.extension.ExtendWith;
 
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.Metrics;
 import io.micrometer.core.instrument.binder.jvm.JvmThreadMetrics;
 import io.netty.handler.codec.http.HttpResponseStatus;
 
-import io.kroxylicious.proxy.config.ConfigParser;
-import io.kroxylicious.proxy.config.Configuration;
 import io.kroxylicious.proxy.config.micrometer.MicrometerConfigurationHook;
-import io.kroxylicious.proxy.testkafkacluster.KafkaClusterConfig;
-import io.kroxylicious.proxy.testkafkacluster.KafkaClusterFactory;
+import io.kroxylicious.testing.kafka.api.KafkaCluster;
+import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
 
+import static io.kroxylicious.proxy.Utils.startProxy;
 import static java.net.http.HttpResponse.BodyHandlers.ofString;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+@ExtendWith(KafkaClusterExtension.class)
 public class MetricsIT {
 
     public static final String PROXY_ADDRESS = "localhost:9192";
-    private TestInfo testInfo;
-    private KafkaProxy proxy;
 
     @BeforeEach
-    public void beforeEach(TestInfo testInfo) {
-        this.testInfo = testInfo;
+    public void beforeEach() {
         Metrics.globalRegistry.clear();
     }
 
-    @AfterEach
-    public void teardown() {
-        if (proxy != null) {
-            try {
-                proxy.shutdown();
-            }
-            catch (InterruptedException e) {
-                throw new RuntimeException(e);
-            }
-        }
-    }
-
     public static class ConfigHook implements MicrometerConfigurationHook {
 
         @Override
@@ -66,15 +50,11 @@ public class MetricsIT {
     }
 
     @Test
-    public void shouldOfferPrometheusMetricsScrapeEndpoint() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
-            cluster.start();
-
-            String config = baseConfigBuilder(PROXY_ADDRESS, cluster.getBootstrapServers())
-                    .withPrometheusEndpoint().build();
-
-            startProxy(config);
+    public void shouldOfferPrometheusMetricsScrapeEndpoint(KafkaCluster cluster) throws Exception {
+        String config = baseConfigBuilder(PROXY_ADDRESS, cluster.getBootstrapServers())
+                .withPrometheusEndpoint().build();
 
+        try (var proxy = startProxy(config)) {
             String counter_name = "test_metric_" + Math.abs(new Random().nextLong()) + "_total";
             Metrics.counter(counter_name).increment();
             HttpRequest request = HttpRequest.newBuilder(URI.create("http://localhost:9193/metrics")).GET().build();
@@ -87,33 +67,25 @@ public class MetricsIT {
     }
 
     @Test
-    public void shouldOfferPrometheusMetricsWithNamedBinder() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
-            cluster.start();
-
-            String config = baseConfigBuilder(PROXY_ADDRESS, cluster.getBootstrapServers())
-                    .withMicrometerBinder("JvmGcMetrics")
-                    .withPrometheusEndpoint().build();
+    public void shouldOfferPrometheusMetricsWithNamedBinder(KafkaCluster cluster) throws Exception {
+        String config = baseConfigBuilder(PROXY_ADDRESS, cluster.getBootstrapServers())
+                .withMicrometerBinder("JvmGcMetrics")
+                .withPrometheusEndpoint().build();
 
-            startProxy(config);
+        try (var proxy = startProxy(config)) {
             HttpRequest request = HttpRequest.newBuilder(URI.create("http://localhost:9193/metrics")).GET().build();
             HttpResponse<String> response = HttpClient.newHttpClient().send(request, ofString());
             assertResponseBodyContainsMeter(response, "jvm_gc_memory_allocated_bytes_total");
         }
-
     }
 
     @Test
-    public void shouldOfferPrometheusMetricsWithCommonTags() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
-            cluster.start();
-
-            String config = baseConfigBuilder(PROXY_ADDRESS, cluster.getBootstrapServers())
-                    .withMicrometerCommonTag("a", "b")
-                    .withPrometheusEndpoint().build();
-
-            startProxy(config);
+    public void shouldOfferPrometheusMetricsWithCommonTags(KafkaCluster cluster) throws Exception {
+        String config = baseConfigBuilder(PROXY_ADDRESS, cluster.getBootstrapServers())
+                .withMicrometerCommonTag("a", "b")
+                .withPrometheusEndpoint().build();
 
+        try (var proxy = startProxy(config)) {
             String counter_name = "test_metric_" + Math.abs(new Random().nextLong()) + "_total";
             Metrics.counter(counter_name).increment();
             HttpRequest request = HttpRequest.newBuilder(URI.create("http://localhost:9193/metrics")).GET().build();
@@ -123,15 +95,12 @@ public class MetricsIT {
     }
 
     @Test
-    public void shouldOfferPrometheusMetricsWithConfigHook() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
-            cluster.start();
-
-            String config = baseConfigBuilder(PROXY_ADDRESS, cluster.getBootstrapServers())
-                    .withMicrometerConfigHook(ConfigHook.class.getTypeName())
-                    .withPrometheusEndpoint().build();
+    public void shouldOfferPrometheusMetricsWithConfigHook(KafkaCluster cluster) throws Exception {
+        String config = baseConfigBuilder(PROXY_ADDRESS, cluster.getBootstrapServers())
+                .withMicrometerConfigHook(ConfigHook.class.getTypeName())
+                .withPrometheusEndpoint().build();
 
-            startProxy(config);
+        try (var proxy = startProxy(config)) {
             HttpRequest request = HttpRequest.newBuilder(URI.create("http://localhost:9193/metrics")).GET().build();
             HttpResponse<String> response = HttpClient.newHttpClient().send(request, ofString());
             assertResponseBodyContainsMeter(response, "jvm_threads_live_threads");
@@ -139,15 +108,12 @@ public class MetricsIT {
     }
 
     @Test
-    public void shouldOfferPrometheusMetricsWithFullyQualifiedBinder() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
-            cluster.start();
+    public void shouldOfferPrometheusMetricsWithFullyQualifiedBinder(KafkaCluster cluster) throws Exception {
+        String config = baseConfigBuilder(PROXY_ADDRESS, cluster.getBootstrapServers())
+                .withMicrometerBinder("io.micrometer.core.instrument.binder.system.UptimeMetrics")
+                .withPrometheusEndpoint().build();
 
-            String config = baseConfigBuilder(PROXY_ADDRESS, cluster.getBootstrapServers())
-                    .withMicrometerBinder("io.micrometer.core.instrument.binder.system.UptimeMetrics")
-                    .withPrometheusEndpoint().build();
-
-            startProxy(config);
+        try (var proxy = startProxy(config)) {
             HttpRequest request = HttpRequest.newBuilder(URI.create("http://localhost:9193/metrics")).GET().build();
             HttpResponse<String> response = HttpClient.newHttpClient().send(request, ofString());
             assertResponseBodyContainsMeter(response, "process_uptime_seconds");
@@ -176,14 +142,4 @@ public class MetricsIT {
                 .addFilter("ApiVersions")
                 .addFilter("BrokerAddress");
     }
-
-    private void startProxy(String config) throws InterruptedException {
-        Configuration proxyConfig = new ConfigParser().parseConfiguration(config);
-
-        KafkaProxy kafkaProxy = new KafkaProxy(proxyConfig);
-        kafkaProxy.startup();
-
-        proxy = kafkaProxy;
-    }
-
 }
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/Utils.java b/integrationtests/src/test/java/io/kroxylicious/proxy/Utils.java
new file mode 100644
index 00000000..5dda63f0
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/Utils.java
@@ -0,0 +1,20 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.proxy;
+
+import io.kroxylicious.proxy.config.ConfigParser;
+import io.kroxylicious.proxy.config.Configuration;
+
+public class Utils {
+    public static KafkaProxy startProxy(String config) throws InterruptedException {
+        Configuration proxyConfig = new ConfigParser().parseConfiguration(config);
+
+        KafkaProxy kafkaProxy = new KafkaProxy(proxyConfig);
+        kafkaProxy.startup();
+
+        return kafkaProxy;
+    }
+}
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
index 5962bf76..fe42382b 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
@@ -5,6 +5,7 @@
  */
 package io.kroxylicious.proxy.multitenant;
 
+import java.nio.file.Path;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.Deque;
@@ -46,23 +47,21 @@ import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.kroxylicious.proxy.KafkaProxy;
-import io.kroxylicious.proxy.config.ConfigParser;
-import io.kroxylicious.proxy.config.Configuration;
-import io.kroxylicious.proxy.testkafkacluster.KafkaCluster;
-import io.kroxylicious.proxy.testkafkacluster.KafkaClusterConfig;
-import io.kroxylicious.proxy.testkafkacluster.KafkaClusterFactory;
-import io.kroxylicious.proxy.testkafkacluster.KeytoolCertificateGenerator;
+import io.kroxylicious.testing.kafka.api.KafkaCluster;
+import io.kroxylicious.testing.kafka.common.KeytoolCertificateGenerator;
+import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
 
+import static io.kroxylicious.proxy.Utils.startProxy;
 import static org.assertj.core.api.Assertions.allOf;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.*;
 
+@ExtendWith(KafkaClusterExtension.class)
 public class MultiTenantIT {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(MultiTenantIT.class);
@@ -82,130 +81,109 @@ public class MultiTenantIT {
     private static final String MY_VALUE = "my-value";
     private TestInfo testInfo;
     private KeytoolCertificateGenerator certificateGenerator;
+    private Path clientTrustStore;
+    @TempDir
+    private Path certsDirectory;
 
     @BeforeEach
     public void beforeEach(TestInfo testInfo) throws Exception {
         this.testInfo = testInfo;
         this.certificateGenerator = new KeytoolCertificateGenerator();
         this.certificateGenerator.generateSelfSignedCertificateEntry("test@redhat.com", "*.multitenant.kafka", "KI", "RedHat", null, null, "US");
+        this.clientTrustStore = certsDirectory.resolve("kafka.truststore.jks");
+        this.certificateGenerator.generateTrustStore(this.certificateGenerator.getCertFilePath(), "client",
+                clientTrustStore.toAbsolutePath().toString());
     }
 
     @Test
-    public void createAndDeleteTopic() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
-            cluster.start();
-
-            String config = getConfig(PROXY_ADDRESS, cluster);
-
-            try (var proxy = startProxy(config)) {
-                try (var admin = Admin.create(commonConfig(TENANT1_PROXY_ADDRESS, Map.of()))) {
-                    var created = createTopics(admin, List.of(NEW_TOPIC_1, NEW_TOPIC_2));
-
-                    ListTopicsResult listTopicsResult = admin.listTopics();
-                    var topicMap = listTopicsResult.namesToListings().get();
-                    assertEquals(2, topicMap.size());
-                    assertThat(topicMap).hasSize(2);
-                    assertThat(topicMap).hasEntrySatisfying(TOPIC_1,
-                            allOf(matches(TopicListing::name, TOPIC_1),
-                                    matches(TopicListing::topicId, created.topicId(TOPIC_1).get())));
-                    assertThat(topicMap).hasEntrySatisfying(TOPIC_1,
-                            allOf(matches(TopicListing::name, TOPIC_1),
-                                    matches(TopicListing::topicId, created.topicId(TOPIC_1).get())));
-                    assertThat(topicMap).hasEntrySatisfying(TOPIC_2, matches(TopicListing::name, TOPIC_2));
-
-                    // Delete by name
-                    var topics1 = TopicNameCollection.ofTopicNames(List.of(TOPIC_1));
-                    var deleted = deleteTopics(admin, topics1);
-                    assertThat(deleted.topicNameValues().keySet()).containsAll(topics1.topicNames());
-
-                    // Delete by id
-                    var topics2 = TopicCollection.ofTopicIds(List.of(created.topicId(TOPIC_2).get()));
-                    deleted = deleteTopics(admin, topics2);
-                    assertThat(deleted.topicIdValues().keySet()).containsAll(topics2.topicIds());
-                }
+    public void createAndDeleteTopic(KafkaCluster cluster) throws Exception {
+        String config = getConfig(PROXY_ADDRESS, cluster);
+
+        try (var proxy = startProxy(config)) {
+            try (var admin = Admin.create(commonConfig(TENANT1_PROXY_ADDRESS, Map.of()))) {
+                var created = createTopics(admin, List.of(NEW_TOPIC_1, NEW_TOPIC_2));
+
+                ListTopicsResult listTopicsResult = admin.listTopics();
+                var topicMap = listTopicsResult.namesToListings().get();
+                assertEquals(2, topicMap.size());
+                assertThat(topicMap).hasSize(2);
+                assertThat(topicMap).hasEntrySatisfying(TOPIC_1,
+                        allOf(matches(TopicListing::name, TOPIC_1),
+                                matches(TopicListing::topicId, created.topicId(TOPIC_1).get())));
+                assertThat(topicMap).hasEntrySatisfying(TOPIC_1,
+                        allOf(matches(TopicListing::name, TOPIC_1),
+                                matches(TopicListing::topicId, created.topicId(TOPIC_1).get())));
+                assertThat(topicMap).hasEntrySatisfying(TOPIC_2, matches(TopicListing::name, TOPIC_2));
+
+                // Delete by name
+                var topics1 = TopicNameCollection.ofTopicNames(List.of(TOPIC_1));
+                var deleted = deleteTopics(admin, topics1);
+                assertThat(deleted.topicNameValues().keySet()).containsAll(topics1.topicNames());
+
+                // Delete by id
+                var topics2 = TopicCollection.ofTopicIds(List.of(created.topicId(TOPIC_2).get()));
+                deleted = deleteTopics(admin, topics2);
+                assertThat(deleted.topicIdValues().keySet()).containsAll(topics2.topicIds());
             }
         }
     }
 
     @Test
-    public void describeTopic() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
-            cluster.start();
-
-            String config = getConfig(PROXY_ADDRESS, cluster);
-
-            try (var proxy = startProxy(config)) {
-                try (var admin = Admin.create(commonConfig(TENANT1_PROXY_ADDRESS, Map.of()))) {
-                    var created = createTopics(admin, List.of(NEW_TOPIC_1));
-
-                    var describeTopicsResult = admin.describeTopics(TopicNameCollection.ofTopicNames(List.of(TOPIC_1)));
-                    var topicMap = describeTopicsResult.allTopicNames().get();
-                    assertThat(topicMap).hasEntrySatisfying(TOPIC_1,
-                            allOf(matches(TopicDescription::name, TOPIC_1),
-                                    matches(TopicDescription::topicId, created.topicId(TOPIC_1).get())));
-                }
+    public void describeTopic(KafkaCluster cluster) throws Exception {
+        String config = getConfig(PROXY_ADDRESS, cluster);
+        try (var proxy = startProxy(config)) {
+            try (var admin = Admin.create(commonConfig(TENANT1_PROXY_ADDRESS, Map.of()))) {
+                var created = createTopics(admin, List.of(NEW_TOPIC_1));
+
+                var describeTopicsResult = admin.describeTopics(TopicNameCollection.ofTopicNames(List.of(TOPIC_1)));
+                var topicMap = describeTopicsResult.allTopicNames().get();
+                assertThat(topicMap).hasEntrySatisfying(TOPIC_1,
+                        allOf(matches(TopicDescription::name, TOPIC_1),
+                                matches(TopicDescription::topicId, created.topicId(TOPIC_1).get())));
             }
         }
     }
 
     @Test
-    public void publishOne() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
-            cluster.start();
-
-            String config = getConfig(PROXY_ADDRESS, cluster);
-
-            try (var proxy = startProxy(config)) {
-                createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
-                produceAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE);
-            }
+    public void publishOne(KafkaCluster cluster) throws Exception {
+        String config = getConfig(PROXY_ADDRESS, cluster);
+        try (var proxy = startProxy(config)) {
+            createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
+            produceAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE);
         }
     }
 
     @Test
-    public void consumeOne() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
-            cluster.start();
-
-            String config = getConfig(PROXY_ADDRESS, cluster);
-
-            try (var proxy = startProxy(config)) {
-                createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
-                produceAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE);
-                consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE, false);
-            }
+    public void consumeOne(KafkaCluster cluster) throws Exception {
+        String config = getConfig(PROXY_ADDRESS, cluster);
+        try (var proxy = startProxy(config)) {
+            createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
+            produceAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE);
+            consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE, false);
         }
     }
 
     @Test
-    public void consumeOneAndOffsetCommit() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
-            cluster.start();
-            String config = getConfig(PROXY_ADDRESS, cluster);
-
-            try (var proxy = startProxy(config)) {
-                createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
-                produceAndVerify(TENANT1_PROXY_ADDRESS,
-                        Stream.of(new ProducerRecord<>(TOPIC_1, MY_KEY, MY_VALUE + "1"), new ProducerRecord<>(TOPIC_1, MY_KEY, MY_VALUE + "2")));
-                consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE + "1", true);
-                consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE + "2", true);
-            }
+    public void consumeOneAndOffsetCommit(KafkaCluster cluster) throws Exception {
+        String config = getConfig(PROXY_ADDRESS, cluster);
+        try (var proxy = startProxy(config)) {
+            createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
+            produceAndVerify(TENANT1_PROXY_ADDRESS,
+                    Stream.of(new ProducerRecord<>(TOPIC_1, MY_KEY, MY_VALUE + "1"), new ProducerRecord<>(TOPIC_1, MY_KEY, MY_VALUE + "2")));
+            consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE + "1", true);
+            consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE + "2", true);
         }
     }
 
     @Test
-    public void tenantIsolation() throws Exception {
-        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
-            cluster.start();
-            String config = getConfig(PROXY_ADDRESS, cluster);
-
-            try (var proxy = startProxy(config)) {
-                createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
-                createTopics(TENANT2_PROXY_ADDRESS, List.of(NEW_TOPIC_2, NEW_TOPIC_3));
-
-                verifyTenant(TENANT1_PROXY_ADDRESS, TOPIC_1);
-                verifyTenant(TENANT2_PROXY_ADDRESS, TOPIC_2, TOPIC_3);
-            }
+    public void tenantIsolation(KafkaCluster cluster) throws Exception {
+        String config = getConfig(PROXY_ADDRESS, cluster);
+        try (var proxy = startProxy(config)) {
+            createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
+            createTopics(TENANT2_PROXY_ADDRESS, List.of(NEW_TOPIC_2, NEW_TOPIC_3));
+
+            verifyTenant(TENANT1_PROXY_ADDRESS, TOPIC_1);
+            verifyTenant(TENANT2_PROXY_ADDRESS, TOPIC_2, TOPIC_3);
         }
     }
 
@@ -257,7 +235,6 @@ public class MultiTenantIT {
             if (offsetCommit) {
                 consumer.commitSync(Duration.ofSeconds(5));
             }
-
         }
     }
 
@@ -287,9 +264,9 @@ public class MultiTenantIT {
         }
     }
 
-    private CreateTopicsResult createTopics(String address, List<NewTopic> topics) throws Exception {
+    private void createTopics(String address, List<NewTopic> topics) throws Exception {
         try (var admin = Admin.create(commonConfig(address, Map.of()))) {
-            return createTopics(admin, topics);
+            createTopics(admin, topics);
         }
     }
 
@@ -320,7 +297,7 @@ public class MultiTenantIT {
         config.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, address);
         config.put(CommonClientConfigs.CLIENT_ID_CONFIG, testInfo.getDisplayName());
         config.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name);
-        config.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, certificateGenerator.getCertLocation());
+        config.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, this.clientTrustStore.toAbsolutePath().toString());
         config.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, certificateGenerator.getPassword());
         return config;
     }
@@ -340,19 +317,10 @@ public class MultiTenantIT {
                   config:
                     addressMapperClazz: io.kroxylicious.proxy.internal.filter.SniAddressMapping
                 - type: MultiTenant
-                """.formatted(proxyAddress, certificateGenerator.getCertLocation(), certificateGenerator.getPassword(), cluster.getBootstrapServers());
+                """.formatted(proxyAddress, certificateGenerator.getKeyStoreLocation(), certificateGenerator.getPassword(), cluster.getBootstrapServers());
         return config;
     }
 
-    private KafkaProxy startProxy(String config) throws InterruptedException {
-        Configuration proxyConfig = new ConfigParser().parseConfiguration(config);
-
-        KafkaProxy kafkaProxy = new KafkaProxy(proxyConfig);
-        kafkaProxy.startup();
-
-        return kafkaProxy;
-    }
-
     @NotNull
     private <T, V> Condition<T> matches(Function<T, V> extractor, V expectedValue) {
         return new Condition<>(item -> Objects.equals(extractor.apply(item), expectedValue), "unexpected entry");