Repository: LangStream/langstream (language: python)

Issue:
Improve deployer jobs timing
After the deploy is completed, the pod still hanging for half a minute due to the k8s client not being closed


Discussion / hints:

Fix patch:
diff --git a/api/src/main/java/com/datastax/oss/sga/api/runtime/ClusterRuntimeRegistry.java b/api/src/main/java/com/datastax/oss/sga/api/runtime/ClusterRuntimeRegistry.java
index ea50a8001..9939962ad 100644
--- a/api/src/main/java/com/datastax/oss/sga/api/runtime/ClusterRuntimeRegistry.java
+++ b/api/src/main/java/com/datastax/oss/sga/api/runtime/ClusterRuntimeRegistry.java
@@ -13,9 +13,11 @@
  * The runtime registry is a singleton that holds all the runtime information about the
  * possible implementations of the SGA API.
  */
-public class ClusterRuntimeRegistry {
+public class ClusterRuntimeRegistry implements AutoCloseable {
 
     private final Map<String, Map<String, Object>> computeClusterConfigurations;
+    protected final Map<String, ComputeClusterRuntime> computeClusterImplementations = new ConcurrentHashMap<>();
+    protected final Map<String, StreamingClusterRuntime> streamingClusterImplementations = new ConcurrentHashMap<>();
 
     public ClusterRuntimeRegistry() {
         this(Collections.emptyMap());
@@ -24,9 +26,6 @@ public ClusterRuntimeRegistry(Map<String, Map<String, Object>> computeClusterCon
         this.computeClusterConfigurations = Collections.unmodifiableMap(computeClusterConfigurations);
     }
 
-    protected final Map<String, ComputeClusterRuntime> computeClusterImplementations = new ConcurrentHashMap<>();
-    protected final Map<String, StreamingClusterRuntime> streamingClusterImplementations = new ConcurrentHashMap<>();
-
     public ComputeClusterRuntime getClusterRuntime(ComputeCluster computeCluster) {
         Objects.requireNonNull(computeCluster, "computeCluster cannot be null");
         Objects.requireNonNull(computeCluster.type(), "computeCluster type cannot be null");
@@ -73,4 +72,13 @@ private static StreamingClusterRuntime loadStreamingClusterRuntime(String stream
         return clusterRuntimeProviderProvider.get().getImplementation();
     }
 
+    @Override
+    public void close() {
+        for (ComputeClusterRuntime computeClusterRuntime : computeClusterImplementations.values()) {
+            computeClusterRuntime.close();
+        }
+        for (StreamingClusterRuntime streamingClusterRuntime : streamingClusterImplementations.values()) {
+            streamingClusterRuntime.close();
+        }
+    }
 }
diff --git a/api/src/main/java/com/datastax/oss/sga/api/runtime/ComputeClusterRuntime.java b/api/src/main/java/com/datastax/oss/sga/api/runtime/ComputeClusterRuntime.java
index 2cb33143d..ab8c39777 100644
--- a/api/src/main/java/com/datastax/oss/sga/api/runtime/ComputeClusterRuntime.java
+++ b/api/src/main/java/com/datastax/oss/sga/api/runtime/ComputeClusterRuntime.java
@@ -24,7 +24,7 @@
  * This is the interface that the SGA framework uses to interact with the cluster. It is used to
  * model a physical cluster runtime (Pulsar, Kafka....)
  */
-public interface ComputeClusterRuntime {
+public interface ComputeClusterRuntime extends AutoCloseable {
 
     String getClusterType();
 
@@ -57,4 +57,7 @@ void delete(String tenant, ExecutionPlan applicationInstance, StreamingClusterRu
     default AgentNodeMetadata computeAgentMetadata(AgentConfiguration agentConfiguration, ExecutionPlan physicalApplicationInstance, StreamingClusterRuntime streamingClusterRuntime) {
         return null;
     }
+
+    default void close() {
+    }
 }
diff --git a/api/src/main/java/com/datastax/oss/sga/api/runtime/StreamingClusterRuntime.java b/api/src/main/java/com/datastax/oss/sga/api/runtime/StreamingClusterRuntime.java
index fdd6f3c70..01191cadd 100644
--- a/api/src/main/java/com/datastax/oss/sga/api/runtime/StreamingClusterRuntime.java
+++ b/api/src/main/java/com/datastax/oss/sga/api/runtime/StreamingClusterRuntime.java
@@ -24,7 +24,7 @@
  * This is the interface that the SGA framework uses to interact with the StreamingCluster. It is used to
  * model a physical cluster runtime with Brokers (Pulsar, Kafka....)
  */
-public interface StreamingClusterRuntime {
+public interface StreamingClusterRuntime extends AutoCloseable {
 
     /**
      * Deploy the topics on the StreamingCluster
@@ -67,4 +67,8 @@ default Map<String, Object> createProducerConfiguration(AgentNode agentImplement
         return Map.of();
     }
 
+    default void close() {
+
+    }
+
 }
diff --git a/core/src/main/java/com/datastax/oss/sga/impl/deploy/ApplicationDeployer.java b/core/src/main/java/com/datastax/oss/sga/impl/deploy/ApplicationDeployer.java
index e772f618f..a9bb8c85a 100644
--- a/core/src/main/java/com/datastax/oss/sga/impl/deploy/ApplicationDeployer.java
+++ b/core/src/main/java/com/datastax/oss/sga/impl/deploy/ApplicationDeployer.java
@@ -10,7 +10,7 @@
 import lombok.Builder;
 
 @Builder
-public final class ApplicationDeployer {
+public final class ApplicationDeployer implements AutoCloseable {
 
       private ClusterRuntimeRegistry registry;
       private PluginsRegistry pluginsRegistry;
@@ -53,4 +53,9 @@ public void delete(String tenant, ExecutionPlan physicalApplicationInstance, Str
       StreamingClusterRuntime streamingClusterRuntime = registry.getStreamingClusterRuntime(applicationInstance.getInstance().streamingCluster());
       clusterRuntime.delete(tenant, physicalApplicationInstance, streamingClusterRuntime, codeStorageArchiveId);
   }
+
+    @Override
+    public void close() {
+        registry.close();
+    }
 }
diff --git a/dev/start-standalone.sh b/dev/start-standalone.sh
index e90786953..691223345 100755
--- a/dev/start-standalone.sh
+++ b/dev/start-standalone.sh
@@ -2,7 +2,7 @@
 
 cd $(dirname $0)/..
 
-set -x
+set -ex
 # This script bootstraps a local K8S based stack with SGA from the current code
 
 eval $(minikube docker-env)
@@ -12,15 +12,16 @@ eval $(minikube docker-env)
 kubectl apply -f helm/examples/minio-dev.yaml
 
 # Start SGA
-helm install sga helm/sga --values helm/examples/local.yaml
+helm install sga helm/sga --values helm/examples/local.yaml || helm upgrade sga helm/sga --values helm/examples/local.yaml
 kubectl wait deployment/sga-control-plane --for condition=available --timeout=60s
 
 
 # Start Kafka
-kubectl create namespace kafka
-kubectl create -f 'https://strimzi.io/install/latest?namespace=kafka' -n kafka
+kubectl create namespace kafka || true
+kubectl apply -f 'https://strimzi.io/install/latest?namespace=kafka' -n kafka
 kubectl apply -f https://strimzi.io/examples/latest/kafka/kafka-persistent-single.yaml -n kafka
 kubectl wait kafka/my-cluster --for=condition=Ready --timeout=300s -n kafka  
 
-# Port forward SGA control plane
+# Port forward SGA control plane
+pkill -f "kubectl port-forward svc/sga-control-plane 8090:8090" || true
 kubectl port-forward svc/sga-control-plane 8090:8090 &
diff --git a/k8s-common/src/main/java/com/datastax/oss/sga/impl/k8s/KubernetesClientFactory.java b/k8s-common/src/main/java/com/datastax/oss/sga/impl/k8s/KubernetesClientFactory.java
index 3f9e19bc9..d766ea079 100644
--- a/k8s-common/src/main/java/com/datastax/oss/sga/impl/k8s/KubernetesClientFactory.java
+++ b/k8s-common/src/main/java/com/datastax/oss/sga/impl/k8s/KubernetesClientFactory.java
@@ -13,37 +13,38 @@ public class KubernetesClientFactory {
 
     private static final Map<String, KubernetesClient> clients = new ConcurrentHashMap<>();
 
-    public static KubernetesClient get(String context) {
-        context = contextKey(context);
-        return clients.computeIfAbsent(context, c -> {
-            final Config config = Config.autoConfigure(c.equals("__null__") ? null : c);
-            try {
-                log.info("Creating kubernetes client for server {}", config.getMasterUrl());
-                final KubernetesClient newClient = new KubernetesClientBuilder()
-                        .withConfig(config)
-                        .build();
-                // https://kubernetes.io/docs/reference/using-api/health-checks/
-                final String livez = newClient.raw("/livez");
-                if (livez == null) {
-                    // 404 means the health check is not implemented
-                    log.warn(
-                            "/livez endpoint not available on server {}, cannot determine if the server is up and running",
-                            config.getMasterUrl());
-                }
-                log.info("Created kubernetes client for server {}", config.getMasterUrl());
-                return newClient;
-            } catch (KubernetesClientException ex) {
-                log.error("Cannot create kubernetes client for server {}, is kubernetes up and running?",
-                        config.getMasterUrl(), ex);
-                throw new RuntimeException("Cannot create kubernetes client for server "
-                        + config.getMasterUrl() + ", is kubernetes up and running?");
-            } catch (Throwable ex) {
-                log.error("Cannot create kubernetes client for server {}, is kubernetes up and running?",
-                        config.getMasterUrl(), ex);
-                throw new RuntimeException(ex);
+    public static KubernetesClient create(String context) {
+        final Config config = Config.autoConfigure(context);
+        try {
+            log.info("Creating kubernetes client for server {}", config.getMasterUrl());
+            final KubernetesClient newClient = new KubernetesClientBuilder()
+                    .withConfig(config)
+                    .build();
+            // https://kubernetes.io/docs/reference/using-api/health-checks/
+            final String livez = newClient.raw("/livez");
+            if (livez == null) {
+                // 404 means the health check is not implemented
+                log.warn(
+                        "/livez endpoint not available on server {}, cannot determine if the server is up and running",
+                        config.getMasterUrl());
             }
+            log.info("Created kubernetes client for server {}", config.getMasterUrl());
+            return newClient;
+        } catch (KubernetesClientException ex) {
+            log.error("Cannot create kubernetes client for server {}, is kubernetes up and running?",
+                    config.getMasterUrl(), ex);
+            throw new RuntimeException("Cannot create kubernetes client for server "
+                    + config.getMasterUrl() + ", is kubernetes up and running?");
+        } catch (Throwable ex) {
+            log.error("Cannot create kubernetes client for server {}, is kubernetes up and running?",
+                    config.getMasterUrl(), ex);
+            throw new RuntimeException(ex);
+        }
+    }
 
-        });
+    public static KubernetesClient get(String context) {
+        context = contextKey(context);
+        return clients.computeIfAbsent(context, KubernetesClientFactory::create);
     }
 
     private static String contextKey(String context) {
@@ -53,28 +54,6 @@ private static String contextKey(String context) {
         return context;
     }
 
-    /**
-     * Visible for testing
-     * @param context
-     * @return
-     */
-    public static void set(String context, KubernetesClient client) {
-        context = contextKey(context);
-        final KubernetesClient old = clients.put(context, client);
-        if (old != null) {
-            old.close();
-        }
-    }
-
-    /**
-     * Visible for testing
-     * @return
-     */
-    public static void clear() {
-        clients.values().forEach(KubernetesClient::close);
-        clients.clear();
-    }
-
     private KubernetesClientFactory() {
     }
 }
diff --git a/k8s-runtime/k8s-runtime-core/src/main/java/com/datastax/oss/sga/runtime/impl/k8s/KubernetesClusterRuntime.java b/k8s-runtime/k8s-runtime-core/src/main/java/com/datastax/oss/sga/runtime/impl/k8s/KubernetesClusterRuntime.java
index bd24ed563..5ac001892 100644
--- a/k8s-runtime/k8s-runtime-core/src/main/java/com/datastax/oss/sga/runtime/impl/k8s/KubernetesClusterRuntime.java
+++ b/k8s-runtime/k8s-runtime-core/src/main/java/com/datastax/oss/sga/runtime/impl/k8s/KubernetesClusterRuntime.java
@@ -37,7 +37,7 @@ public String getClusterType() {
     @SneakyThrows
     public void initialize(Map<String, Object> configuration) {
         this.configuration = mapper.convertValue(configuration, KubernetesClusterRuntimeConfiguration.class);
-        this.client = KubernetesClientFactory.get(null);
+        this.client = KubernetesClientFactory.create(null);
     }
 
     @Override
@@ -152,4 +152,11 @@ private String computeNamespace(String tenant) {
         final String namespace = configuration.getNamespacePrefix() + tenant;
         return namespace;
     }
+
+    @Override
+    public void close() {
+        if (client != null) {
+            client.close();
+        }
+    }
 }
diff --git a/pom.xml b/pom.xml
index 3278d335d..607360074 100644
--- a/pom.xml
+++ b/pom.xml
@@ -159,6 +159,11 @@
         <artifactId>mockito-core</artifactId>
         <version>${mockito-core.version}</version>
       </dependency>
+      <dependency>
+        <groupId>org.mockito</groupId>
+        <artifactId>mockito-inline</artifactId>
+        <version>${mockito-core.version}</version>
+      </dependency>
       <dependency>
         <groupId>org.awaitility</groupId>
         <artifactId>awaitility</artifactId>
diff --git a/pulsar/src/main/java/com/datastax/oss/sga/pulsar/PulsarStreamingClusterRuntime.java b/pulsar/src/main/java/com/datastax/oss/sga/pulsar/PulsarStreamingClusterRuntime.java
index daecd91a7..117a64518 100644
--- a/pulsar/src/main/java/com/datastax/oss/sga/pulsar/PulsarStreamingClusterRuntime.java
+++ b/pulsar/src/main/java/com/datastax/oss/sga/pulsar/PulsarStreamingClusterRuntime.java
@@ -33,8 +33,6 @@
 @Slf4j
 public class PulsarStreamingClusterRuntime implements StreamingClusterRuntime {
 
-    static final ObjectMapper mapper = new ObjectMapper();
-
     @Override
     @SneakyThrows
     public void deploy(ExecutionPlan applicationInstance) {
diff --git a/runtime/runtime-impl/src/main/assemble/entrypoint.sh b/runtime/runtime-impl/src/main/assemble/entrypoint.sh
index e9bb17c7e..b8d43f6f9 100644
--- a/runtime/runtime-impl/src/main/assemble/entrypoint.sh
+++ b/runtime/runtime-impl/src/main/assemble/entrypoint.sh
@@ -1,2 +1,2 @@
 #!/bin/bash
-java ${JAVA_OPTS} -cp "/app/lib/*" "com.datastax.oss.sga.runtime.Main" "$@"
\ No newline at end of file
+exec java ${JAVA_OPTS} -cp "/app/lib/*" "com.datastax.oss.sga.runtime.Main" "$@"
\ No newline at end of file
diff --git a/runtime/runtime-impl/src/main/java/com/datastax/oss/sga/runtime/deployer/RuntimeDeployer.java b/runtime/runtime-impl/src/main/java/com/datastax/oss/sga/runtime/deployer/RuntimeDeployer.java
index 6bc99c6e9..823b8247d 100644
--- a/runtime/runtime-impl/src/main/java/com/datastax/oss/sga/runtime/deployer/RuntimeDeployer.java
+++ b/runtime/runtime-impl/src/main/java/com/datastax/oss/sga/runtime/deployer/RuntimeDeployer.java
@@ -90,15 +90,16 @@ private static void deploy(Map<String, Map<String, Object>> clusterRuntimeConfig
                 MAPPER.readValue(applicationConfig, Application.class);
         appInstance.setSecrets(secrets);
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        try (ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry(clusterRuntimeConfiguration))
                 .pluginsRegistry(new PluginsRegistry())
-                .build();
+                .build()) {
 
-        final ExecutionPlan implementation = deployer.createImplementation(applicationId, appInstance);
-        deployer.deploy(configuration.getTenant(), implementation,  configuration.getCodeStorageArchiveId());
-        log.info("Application {} deployed", applicationId);
+            final ExecutionPlan implementation = deployer.createImplementation(applicationId, appInstance);
+            deployer.deploy(configuration.getTenant(), implementation, configuration.getCodeStorageArchiveId());
+            log.info("Application {} deployed", applicationId);
+        }
     }
 
     private static void delete(Map<String, Map<String, Object>> clusterRuntimeConfiguration,
@@ -113,16 +114,17 @@ private static void delete(Map<String, Map<String, Object>> clusterRuntimeConfig
                 MAPPER.readValue(applicationConfig, Application.class);
         appInstance.setSecrets(secrets);
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        try (ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry(clusterRuntimeConfiguration))
                 .pluginsRegistry(new PluginsRegistry())
-                .build();
+                .build();) {
 
-        log.info("Deleting application {}", applicationId);
-        final ExecutionPlan implementation = deployer.createImplementation(applicationId, appInstance);
-        deployer.delete(configuration.getTenant(), implementation, configuration.getCodeStorageArchiveId());
-        log.info("Application {} deleted", applicationId);
+            log.info("Deleting application {}", applicationId);
+            final ExecutionPlan implementation = deployer.createImplementation(applicationId, appInstance);
+            deployer.delete(configuration.getTenant(), implementation, configuration.getCodeStorageArchiveId());
+            log.info("Application {} deleted", applicationId);
+        }
     }
 }
 


Test patch:
diff --git a/core/src/test/java/com/datastax/oss/sga/impl/deploy/ApplicationDeployerTest.java b/core/src/test/java/com/datastax/oss/sga/impl/deploy/ApplicationDeployerTest.java
index 0fd1ad5c3..2300274d7 100644
--- a/core/src/test/java/com/datastax/oss/sga/impl/deploy/ApplicationDeployerTest.java
+++ b/core/src/test/java/com/datastax/oss/sga/impl/deploy/ApplicationDeployerTest.java
@@ -12,6 +12,7 @@
 import com.datastax.oss.sga.impl.parser.ModelBuilder;
 
 import java.util.Map;
+import lombok.Cleanup;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -55,7 +56,7 @@ void testDeploy() throws Exception {
         registry.addClusterRuntime("mock", mockRuntime);
         registry.addStreamingClusterRuntime("mock", mockStreamingRuntime);
 
-        final ApplicationDeployer deployer = ApplicationDeployer
+        final @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .pluginsRegistry(new PluginsRegistry())
                 .registry(registry)
diff --git a/k8s-common/src/test/java/com/datastax/oss/sga/impl/k8s/tests/KubeK3sServer.java b/k8s-common/src/test/java/com/datastax/oss/sga/impl/k8s/tests/KubeK3sServer.java
index 8e13e26c4..f9452be2e 100644
--- a/k8s-common/src/test/java/com/datastax/oss/sga/impl/k8s/tests/KubeK3sServer.java
+++ b/k8s-common/src/test/java/com/datastax/oss/sga/impl/k8s/tests/KubeK3sServer.java
@@ -1,5 +1,6 @@
 package com.datastax.oss.sga.impl.k8s.tests;
 
+import static org.mockito.ArgumentMatchers.isNull;
 import com.dajudge.kindcontainer.K3sContainer;
 import com.dajudge.kindcontainer.K3sContainerVersion;
 import com.dajudge.kindcontainer.KubernetesImageSpec;
@@ -21,6 +22,8 @@
 import org.junit.jupiter.api.extension.AfterAllCallback;
 import org.junit.jupiter.api.extension.BeforeAllCallback;
 import org.junit.jupiter.api.extension.ExtensionContext;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
 import org.testcontainers.containers.output.OutputFrame;
 import org.testcontainers.utility.DockerImageName;
 
@@ -30,6 +33,7 @@ public class KubeK3sServer implements AutoCloseable, BeforeAllCallback, AfterAll
     private K3sContainer k3s;
     @Getter
     private KubernetesClient client;
+    private MockedStatic mocked;
 
     private final boolean installCRDs;
 
@@ -46,7 +50,12 @@ public void close() throws Exception {
         if (k3s != null) {
             k3s.stop();
         }
-        KubernetesClientFactory.clear();
+        if (client != null) {
+            client.close();
+        }
+        if (mocked != null) {
+            mocked.close();
+        }
     }
 
     @Override
@@ -72,7 +81,9 @@ public void beforeAll(ExtensionContext extensionContext) throws Exception {
                 .build();
 
 
-        KubernetesClientFactory.set(null, client);
+        mocked = Mockito.mockStatic(KubernetesClientFactory.class);
+        mocked.when(() -> KubernetesClientFactory.get(isNull())).thenReturn(client);
+        mocked.when(() -> KubernetesClientFactory.create(isNull())).thenReturn(client);
 
         if (installCRDs) {
 
diff --git a/k8s-common/src/test/java/com/datastax/oss/sga/impl/k8s/tests/KubeTestServer.java b/k8s-common/src/test/java/com/datastax/oss/sga/impl/k8s/tests/KubeTestServer.java
index bde8cd043..625d757d7 100644
--- a/k8s-common/src/test/java/com/datastax/oss/sga/impl/k8s/tests/KubeTestServer.java
+++ b/k8s-common/src/test/java/com/datastax/oss/sga/impl/k8s/tests/KubeTestServer.java
@@ -1,9 +1,12 @@
 package com.datastax.oss.sga.impl.k8s.tests;
 
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.isNull;
 import com.datastax.oss.sga.deployer.k8s.api.crds.agents.AgentCustomResource;
 import com.datastax.oss.sga.impl.k8s.KubernetesClientFactory;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
+import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
 import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -18,23 +21,40 @@
 import org.junit.jupiter.api.extension.BeforeAllCallback;
 import org.junit.jupiter.api.extension.BeforeEachCallback;
 import org.junit.jupiter.api.extension.ExtensionContext;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import org.mockito.stubbing.OngoingStubbing;
 
 @Slf4j
 public class KubeTestServer implements AutoCloseable, BeforeEachCallback, BeforeAllCallback, AfterAllCallback {
 
     public static class Server extends KubernetesMockServer {
 
+        MockedStatic<KubernetesClientFactory> mocked;
+        NamespacedKubernetesClient client;
+
         @Override
         @SneakyThrows
         public void init() {
             super.init();
-            KubernetesClientFactory.set(null, createClient());
+
+            mocked = Mockito.mockStatic(KubernetesClientFactory.class);
+            client = createClient();
+
+            mocked.when(() -> KubernetesClientFactory.get(isNull())).thenReturn(client);
+            mocked.when(() -> KubernetesClientFactory.create(isNull())).thenReturn(client);
         }
 
         @Override
         public void destroy() {
             super.destroy();
-            KubernetesClientFactory.clear();
+            System.out.println("stop mock");
+            if (mocked != null) {
+                mocked.close();
+            }
+            if (client != null) {
+                client.close();
+            }
         }
     }
 
diff --git a/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/GenAIAgentsTest.java b/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/GenAIAgentsTest.java
index 79756613f..42af2ba1a 100644
--- a/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/GenAIAgentsTest.java
+++ b/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/GenAIAgentsTest.java
@@ -12,6 +12,7 @@
 import com.datastax.oss.sga.impl.deploy.ApplicationDeployer;
 import com.datastax.oss.sga.impl.noop.NoOpStreamingClusterRuntimeProvider;
 import com.datastax.oss.sga.impl.parser.ModelBuilder;
+import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
 import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.Test;
@@ -76,40 +77,43 @@ public void testOpenAIComputeEmbeddingFunction() throws Exception {
                                       text: "{{% value.name }} {{% value.description }}"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        try (ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
-                .build();
-
-        Module module = applicationInstance.getModule("module-1");
-
-        ExecutionPlan implementation = deployer.createImplementation("app", applicationInstance);
-        assertEquals(1, implementation.getAgents().size());
-        assertTrue(implementation.getConnectionImplementation(module,
-                new Connection(TopicDefinition.fromName("input-topic"))) instanceof NoOpStreamingClusterRuntimeProvider.SimpleTopic);
-        assertTrue(implementation.getConnectionImplementation(module,
-                new Connection(TopicDefinition.fromName("output-topic"))) instanceof NoOpStreamingClusterRuntimeProvider.SimpleTopic);
-
-        AgentNode agentImplementation = implementation.getAgentImplementation(module, "step1");
-        assertNotNull(agentImplementation);
-        DefaultAgentNode step =
-                (DefaultAgentNode) agentImplementation;
-        Map<String, Object> configuration = step.getConfiguration();
-        log.info("Configuration: {}", configuration);
-        Map<String, Object> openAIConfiguration = (Map<String, Object>) configuration.get("openai");
-        log.info("openAIConfiguration: {}", openAIConfiguration);
-        assertEquals("http://something", openAIConfiguration.get("url"));
-        assertEquals("xxcxcxc", openAIConfiguration.get("access-key"));
-        assertEquals("azure", openAIConfiguration.get("provider"));
+                .build();) {
+
+            Module module = applicationInstance.getModule("module-1");
+
+            ExecutionPlan implementation = deployer.createImplementation("app", applicationInstance);
+            assertEquals(1, implementation.getAgents().size());
+            assertTrue(implementation.getConnectionImplementation(module,
+                    new Connection(TopicDefinition.fromName(
+                            "input-topic"))) instanceof NoOpStreamingClusterRuntimeProvider.SimpleTopic);
+            assertTrue(implementation.getConnectionImplementation(module,
+                    new Connection(TopicDefinition.fromName(
+                            "output-topic"))) instanceof NoOpStreamingClusterRuntimeProvider.SimpleTopic);
+
+            AgentNode agentImplementation = implementation.getAgentImplementation(module, "step1");
+            assertNotNull(agentImplementation);
+            DefaultAgentNode step =
+                    (DefaultAgentNode) agentImplementation;
+            Map<String, Object> configuration = step.getConfiguration();
+            log.info("Configuration: {}", configuration);
+            Map<String, Object> openAIConfiguration = (Map<String, Object>) configuration.get("openai");
+            log.info("openAIConfiguration: {}", openAIConfiguration);
+            assertEquals("http://something", openAIConfiguration.get("url"));
+            assertEquals("xxcxcxc", openAIConfiguration.get("access-key"));
+            assertEquals("azure", openAIConfiguration.get("provider"));
 
 
-        List<Map<String, Object>> steps = (List<Map<String, Object>>) configuration.get("steps");
-        assertEquals(1, steps.size());
-        Map<String, Object> step1 = steps.get(0);
-        assertEquals("text-embedding-ada-002", step1.get("model"));
-        assertEquals("value.embeddings", step1.get("embeddings-field"));
-        assertEquals("{{ value.name }} {{ value.description }}", step1.get("text"));
+            List<Map<String, Object>> steps = (List<Map<String, Object>>) configuration.get("steps");
+            assertEquals(1, steps.size());
+            Map<String, Object> step1 = steps.get(0);
+            assertEquals("text-embedding-ada-002", step1.get("model"));
+            assertEquals("value.embeddings", step1.get("embeddings-field"));
+            assertEquals("{{ value.name }} {{ value.description }}", step1.get("text"));
+        }
 
 
     }
@@ -159,7 +163,7 @@ public void testMergeGenAIToolKitAgents() throws Exception {
                                       part: "value"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
@@ -298,7 +302,7 @@ public void testMapAllGenAIToolKitAgents() throws Exception {
                                       schema-type: "string"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
@@ -401,7 +405,7 @@ public void testMultipleQuerySteps() throws Exception {
                                       schema-type: "string"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
diff --git a/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/KafkaConnectAgentsTest.java b/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/KafkaConnectAgentsTest.java
index ca3ccf6b6..1dc6f38a8 100644
--- a/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/KafkaConnectAgentsTest.java
+++ b/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/KafkaConnectAgentsTest.java
@@ -13,6 +13,7 @@
 import com.datastax.oss.sga.impl.deploy.ApplicationDeployer;
 import com.datastax.oss.sga.impl.noop.NoOpStreamingClusterRuntimeProvider;
 import com.datastax.oss.sga.impl.parser.ModelBuilder;
+import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
 import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.Test;
@@ -59,7 +60,7 @@ public void testConfigureKafkaConnectSink() throws Exception {
                                       file: /tmp/test.sink.txt
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
@@ -105,7 +106,7 @@ public void testConfigureKafkaConnectSources() throws Exception {
                                       file: /tmp/test.txt
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
diff --git a/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/KubernetesClusterRuntimeDockerTest.java b/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/KubernetesClusterRuntimeDockerTest.java
index 04adf41ff..f2b3bd21a 100644
--- a/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/KubernetesClusterRuntimeDockerTest.java
+++ b/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/KubernetesClusterRuntimeDockerTest.java
@@ -15,6 +15,7 @@
 import com.datastax.oss.sga.impl.k8s.tests.KubeTestServer;
 import com.datastax.oss.sga.impl.parser.ModelBuilder;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
@@ -46,7 +47,7 @@ private ApplicationDeployer getDeployer() {
         config.setImagePullPolicy("Never");
         config.setNamespacePrefix("sga-");
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry(
                         Map.of("kubernetes", new ObjectMapper().convertValue(config, Map.class))))
@@ -96,7 +97,7 @@ public void testOpenAIComputeEmbeddingFunction() throws Exception {
                                       text: "{{% value.name }} {{% value.description }}"
                                 """));
 
-        ApplicationDeployer deployer = getDeployer();
+        @Cleanup ApplicationDeployer deployer = getDeployer();
 
         Module module = applicationInstance.getModule("module-1");
 
diff --git a/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/PythonCodeAgentsTest.java b/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/PythonCodeAgentsTest.java
index 02da94ebb..ab9dcdca9 100644
--- a/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/PythonCodeAgentsTest.java
+++ b/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/PythonCodeAgentsTest.java
@@ -13,6 +13,7 @@
 import com.datastax.oss.sga.impl.deploy.ApplicationDeployer;
 import com.datastax.oss.sga.impl.noop.NoOpStreamingClusterRuntimeProvider;
 import com.datastax.oss.sga.impl.parser.ModelBuilder;
+import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
 import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.Test;
@@ -69,7 +70,7 @@ public void testConfigurePythonAgents() throws Exception {
                                       config2: value2
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
diff --git a/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/SimpleClusterRuntimeTest.java b/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/SimpleClusterRuntimeTest.java
index 40eef7856..8da5b45f5 100644
--- a/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/SimpleClusterRuntimeTest.java
+++ b/k8s-runtime/k8s-runtime-core/src/test/java/com/datastax/oss/sga/runtime/impl/k8s/SimpleClusterRuntimeTest.java
@@ -12,6 +12,7 @@
 import com.datastax.oss.sga.impl.noop.NoOpComputeClusterRuntimeProvider;
 import com.datastax.oss.sga.impl.noop.NoOpStreamingClusterRuntimeProvider;
 import com.datastax.oss.sga.impl.parser.ModelBuilder;
+import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
 import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.Test;
@@ -68,7 +69,7 @@ public void testMapGenericPulsarFunctionsChain() throws Exception {
                                       config2: "value2"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
diff --git a/kafka/src/test/java/com/datastax/oss/sga/kafka/KafkaClusterRuntimeDockerTest.java b/kafka/src/test/java/com/datastax/oss/sga/kafka/KafkaClusterRuntimeDockerTest.java
index 38f74a6db..7826e6201 100644
--- a/kafka/src/test/java/com/datastax/oss/sga/kafka/KafkaClusterRuntimeDockerTest.java
+++ b/kafka/src/test/java/com/datastax/oss/sga/kafka/KafkaClusterRuntimeDockerTest.java
@@ -10,6 +10,7 @@
 import com.datastax.oss.sga.api.runtime.PluginsRegistry;
 import com.datastax.oss.sga.impl.deploy.ApplicationDeployer;
 import com.datastax.oss.sga.impl.parser.ModelBuilder;
+import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.TopicDescription;
@@ -51,7 +52,7 @@ public void testMapKafkaTopics() throws Exception {
                                     partitions: 2                                     
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
diff --git a/kafka/src/test/java/com/datastax/oss/sga/kafka/KafkaStreamingClusterRuntimeTest.java b/kafka/src/test/java/com/datastax/oss/sga/kafka/KafkaStreamingClusterRuntimeTest.java
index 8bf19bad8..7e7702f61 100644
--- a/kafka/src/test/java/com/datastax/oss/sga/kafka/KafkaStreamingClusterRuntimeTest.java
+++ b/kafka/src/test/java/com/datastax/oss/sga/kafka/KafkaStreamingClusterRuntimeTest.java
@@ -10,6 +10,7 @@
 import com.datastax.oss.sga.api.runtime.PluginsRegistry;
 import com.datastax.oss.sga.impl.deploy.ApplicationDeployer;
 import com.datastax.oss.sga.impl.parser.ModelBuilder;
+import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.jupiter.api.Test;
 import java.util.Map;
@@ -35,7 +36,7 @@ public void testMapKafkaTopics() throws Exception {
                                       schema: '{"type":"record","namespace":"examples","name":"Product","fields":[{"name":"id","type":"string"},{"name":"name","type":"string"},{"name":"description","type":"string"},{"name":"price","type":"double"},{"name":"category","type":"string"},{"name":"item_vector","type":"bytes"}]}}'                               
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
diff --git a/pulsar/src/test/java/com/datastax/oss/sga/pulsar/PulsarClusterRuntimeDockerTest.java b/pulsar/src/test/java/com/datastax/oss/sga/pulsar/PulsarClusterRuntimeDockerTest.java
index 3e4482c7b..ad6886034 100644
--- a/pulsar/src/test/java/com/datastax/oss/sga/pulsar/PulsarClusterRuntimeDockerTest.java
+++ b/pulsar/src/test/java/com/datastax/oss/sga/pulsar/PulsarClusterRuntimeDockerTest.java
@@ -11,6 +11,7 @@
 import com.datastax.oss.sga.impl.common.DefaultAgentNode;
 import com.datastax.oss.sga.impl.deploy.ApplicationDeployer;
 import com.datastax.oss.sga.impl.parser.ModelBuilder;
+import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminException;
@@ -73,7 +74,7 @@ public void testDeployTopics() throws Exception {
                                 pipeline:
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .<ExecutionPlan>builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
@@ -121,7 +122,7 @@ public void testDeployCassandraSink() throws Exception {
                                       mappings: "id=value.id,name=value.name,description=value.description,item_vector=value.item_vector"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
@@ -185,7 +186,7 @@ public void testDeployDataGeneratorSource() throws Exception {
                                       sourceType: "data-generator"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
@@ -237,7 +238,7 @@ public void testDeployChainOfGenericFunctions() throws Exception {
                                     steps: []
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
@@ -301,7 +302,7 @@ public void testOpenAIComputeEmbeddingFunction() throws Exception {
                                       text: "{{% value.name }} {{% value.description }}"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
diff --git a/pulsar/src/test/java/com/datastax/oss/sga/pulsar/PulsarClusterRuntimeTest.java b/pulsar/src/test/java/com/datastax/oss/sga/pulsar/PulsarClusterRuntimeTest.java
index 8c562a854..7587a5cf4 100644
--- a/pulsar/src/test/java/com/datastax/oss/sga/pulsar/PulsarClusterRuntimeTest.java
+++ b/pulsar/src/test/java/com/datastax/oss/sga/pulsar/PulsarClusterRuntimeTest.java
@@ -12,6 +12,7 @@
 import com.datastax.oss.sga.impl.deploy.ApplicationDeployer;
 import com.datastax.oss.sga.impl.parser.ModelBuilder;
 import com.datastax.oss.sga.pulsar.agents.PulsarAgentNodeMetadata;
+import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
 import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.Test;
@@ -49,7 +50,7 @@ public void testMapCassandraSink() throws Exception {
                                       mappings: "id=value.id,name=value.name,description=value.description,item_vector=value.item_vector"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
@@ -119,7 +120,7 @@ public void testMapGenericPulsarSink() throws Exception {
                                       config2: "value2"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
@@ -165,7 +166,7 @@ public void testMapGenericPulsarSource() throws Exception {
                                       config2: "value2"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
@@ -216,7 +217,7 @@ public void testMapGenericPulsarFunction() throws Exception {
                                       config2: "value2"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
@@ -286,7 +287,7 @@ public void testMapGenericPulsarFunctionsChain() throws Exception {
                                       config2: "value2"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
@@ -382,7 +383,7 @@ public void testOpenAIComputeEmbeddingFunction() throws Exception {
                                       text: "{{% value.name }} {{% value.description }}"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
@@ -460,7 +461,7 @@ public void testSanitizePipelineName() throws Exception {
                                       config2: "value2"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
@@ -538,7 +539,7 @@ public void testMergeGenAIToolKitAgents() throws Exception {
                                       text: "{{% value.name }}"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
diff --git a/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/kafka/GenIAgentsRunnerTest.java b/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/kafka/GenIAgentsRunnerTest.java
index 5c2ac2f0b..e9dcdd0bc 100644
--- a/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/kafka/GenIAgentsRunnerTest.java
+++ b/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/kafka/GenIAgentsRunnerTest.java
@@ -17,6 +17,7 @@
 import com.datastax.oss.sga.runtime.api.agent.RuntimePodConfiguration;
 import com.datastax.oss.sga.runtime.k8s.api.PodAgentConfiguration;
 import java.nio.charset.StandardCharsets;
+import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -79,7 +80,7 @@ public void testRunAITools() throws Exception {
                                         - "description"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
diff --git a/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/kafka/KafkaConnectRunnerTest.java b/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/kafka/KafkaConnectRunnerTest.java
index 601e907ec..e7309e0ee 100644
--- a/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/kafka/KafkaConnectRunnerTest.java
+++ b/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/kafka/KafkaConnectRunnerTest.java
@@ -16,6 +16,7 @@
 import com.datastax.oss.sga.runtime.api.agent.CodeStorageConfig;
 import com.datastax.oss.sga.runtime.api.agent.RuntimePodConfiguration;
 import com.datastax.oss.sga.runtime.k8s.api.PodAgentConfiguration;
+import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.producer.KafkaProducer;
@@ -80,7 +81,7 @@ public void testRunKafkaConnectSink() throws Exception {
                                       file: /tmp/test.sink.txt
                                 """.formatted(DummySinkConnector.class.getName())));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
diff --git a/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/kafka/KafkaRunnerDockerTest.java b/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/kafka/KafkaRunnerDockerTest.java
index 5f0622893..a3646c5c7 100644
--- a/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/kafka/KafkaRunnerDockerTest.java
+++ b/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/kafka/KafkaRunnerDockerTest.java
@@ -14,6 +14,7 @@
 import com.datastax.oss.sga.api.runtime.PluginsRegistry;
 import com.datastax.oss.sga.impl.deploy.ApplicationDeployer;
 import com.datastax.oss.sga.impl.parser.ModelBuilder;
+import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -58,7 +59,7 @@ public void testConnectToTopics() throws Exception {
                                     creation-mode: create-if-not-exists
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
diff --git a/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/pulsar/PulsarRunnerDockerTest.java b/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/pulsar/PulsarRunnerDockerTest.java
index bc5d5666f..ea907f701 100644
--- a/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/pulsar/PulsarRunnerDockerTest.java
+++ b/runtime/runtime-impl/src/test/java/com/datastax/oss/sga/pulsar/PulsarRunnerDockerTest.java
@@ -15,6 +15,7 @@
 import com.datastax.oss.sga.runtime.api.agent.CodeStorageConfig;
 import com.datastax.oss.sga.runtime.api.agent.RuntimePodConfiguration;
 import com.datastax.oss.sga.runtime.k8s.api.PodAgentConfiguration;
+import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminException;
@@ -79,7 +80,7 @@ public void testRunAITools() throws Exception {
                                         - "description"
                                 """));
 
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
diff --git a/webservice/src/test/java/com/datastax/oss/sga/webservice/application/ApplicationServiceValidateUpdateTest.java b/webservice/src/test/java/com/datastax/oss/sga/webservice/application/ApplicationServiceValidateUpdateTest.java
index 119241483..3525c7a0c 100644
--- a/webservice/src/test/java/com/datastax/oss/sga/webservice/application/ApplicationServiceValidateUpdateTest.java
+++ b/webservice/src/test/java/com/datastax/oss/sga/webservice/application/ApplicationServiceValidateUpdateTest.java
@@ -12,6 +12,7 @@
 import com.datastax.oss.sga.impl.parser.ModelBuilder;
 import java.util.List;
 import java.util.Map;
+import lombok.Cleanup;
 import lombok.SneakyThrows;
 import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.Test;
@@ -113,7 +114,7 @@ private static void checkTopics(List<ModelBuilder.TopicDefinitionModel> from,
 
     @SneakyThrows
     private static ExecutionPlan buildPlanWithTopics(List<ModelBuilder.TopicDefinitionModel> topics) {
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())
@@ -283,7 +284,7 @@ private static void checkAgents(List<ModelBuilder.AgentModel> from,
 
     @SneakyThrows
     private static ExecutionPlan buildPlanWithModels(List<ModelBuilder.AgentModel> agents) {
-        ApplicationDeployer deployer = ApplicationDeployer
+        @Cleanup ApplicationDeployer deployer = ApplicationDeployer
                 .builder()
                 .registry(new ClusterRuntimeRegistry())
                 .pluginsRegistry(new PluginsRegistry())