Ensure that local containers used during tests are properly shutdown after the test is completed

tests/src/test/java/org/apache/camel/kafkaconnector/services/cassandra/CassandraLocalContainerService.java

@@ -57,4 +57,10 @@ public CassandraClient getClient() {
     public void initialize() {
         LOG.info("Cassandra server running at address {}", getCQL3Endpoint());
     }
+
+    @Override
+    public void shutdown() {
+        LOG.info("Stopping the Cassandra container");
+        container.stop();
+    }
 }

tests/src/test/java/org/apache/camel/kafkaconnector/services/cassandra/CassandraService.java

@@ -18,13 +18,14 @@
 package org.apache.camel.kafkaconnector.services.cassandra;
 
 import org.apache.camel.kafkaconnector.clients.cassandra.CassandraClient;
+import org.junit.jupiter.api.extension.AfterAllCallback;
 import org.junit.jupiter.api.extension.BeforeAllCallback;
 import org.junit.jupiter.api.extension.ExtensionContext;
 
 /**
  * Represents an endpoint to a Cassandra instnace
  */
-public interface CassandraService extends BeforeAllCallback {
+public interface CassandraService extends BeforeAllCallback, AfterAllCallback {
 
     int getCQL3Port();
 
@@ -39,6 +40,10 @@ default String getCQL3Endpoint() {
      */
     void initialize();
 
+    /**
+     * Shuts down the service after the test has completed
+     */
+    void shutdown();
 
     CassandraClient getClient();
 
@@ -47,4 +52,9 @@ default String getCQL3Endpoint() {
     default void beforeAll(ExtensionContext extensionContext) throws Exception {
         initialize();
     }
+
+    @Override
+    default void afterAll(ExtensionContext extensionContext) throws Exception {
+        shutdown();
+    }
 }

tests/src/test/java/org/apache/camel/kafkaconnector/services/cassandra/RemoteCassandraService.java

@@ -52,4 +52,9 @@ public CassandraClient getClient() {
     public void initialize() {
         // NO-OP
     }
+
+    @Override
+    public void shutdown() {
+        // NO-OP
+    }
 }

tests/src/test/java/org/apache/camel/kafkaconnector/services/elasticsearch/ElasticSearchLocalContainerService.java

@@ -52,6 +52,12 @@ public void initialize() {
         LOG.info("ElasticSearch instance running at {}", getHttpHostAddress());
     }
 
+    @Override
+    public void shutdown() {
+        LOG.info("Stopping the ElasticSearch container");
+        container.stop();
+    }
+
     @Override
     public ElasticSearchClient getClient() {
         return new ElasticSearchClient("localhost", container.getMappedPort(ELASTIC_SEARCH_PORT),

tests/src/test/java/org/apache/camel/kafkaconnector/services/elasticsearch/ElasticSearchService.java

@@ -18,10 +18,11 @@
 package org.apache.camel.kafkaconnector.services.elasticsearch;
 
 import org.apache.camel.kafkaconnector.clients.elasticsearch.ElasticSearchClient;
+import org.junit.jupiter.api.extension.AfterAllCallback;
 import org.junit.jupiter.api.extension.BeforeAllCallback;
 import org.junit.jupiter.api.extension.ExtensionContext;
 
-public interface ElasticSearchService extends BeforeAllCallback {
+public interface ElasticSearchService extends BeforeAllCallback, AfterAllCallback {
 
     String getHttpHostAddress();
 
@@ -30,11 +31,20 @@ public interface ElasticSearchService extends BeforeAllCallback {
      */
     void initialize();
 
+    /**
+     * Shuts down the service after the test has completed
+     */
+    void shutdown();
 
     ElasticSearchClient getClient();
 
     @Override
     default void beforeAll(ExtensionContext extensionContext) throws Exception {
         initialize();
     }
+
+    @Override
+    default void afterAll(ExtensionContext extensionContext) throws Exception {
+        shutdown();
+    }
 }

tests/src/test/java/org/apache/camel/kafkaconnector/services/elasticsearch/RemoteElasticSearchService.java

@@ -47,6 +47,11 @@ public void initialize() {
         // NO-OP
     }
 
+    @Override
+    public void shutdown() {
+        // NO-OP
+    }
+
     @Override
     public ElasticSearchClient getClient() {
         return new ElasticSearchClient(getHost(), getPort(), TestCommon.DEFAULT_ELASTICSEARCH_INDEX);

tests/src/test/java/org/apache/camel/kafkaconnector/services/jms/ContainerLocalService.java

@@ -57,4 +57,10 @@ public String getDefaultEndpoint() {
     public void initialize() {
         LOG.info("JMS broker running at address {}", container.getDefaultEndpoint());
     }
+
+    @Override
+    public void shutdown() {
+        LOG.info("Stopping JMS broker container");
+        container.stop();
+    }
 }

tests/src/test/java/org/apache/camel/kafkaconnector/services/jms/JMSService.java

@@ -20,10 +20,11 @@
 import java.util.Properties;
 
 import org.apache.camel.kafkaconnector.clients.jms.JMSClient;
+import org.junit.jupiter.api.extension.AfterAllCallback;
 import org.junit.jupiter.api.extension.BeforeAllCallback;
 import org.junit.jupiter.api.extension.ExtensionContext;
 
-public interface JMSService extends BeforeAllCallback {
+public interface JMSService extends BeforeAllCallback, AfterAllCallback {
 
     /**
      * Gets the connection properties for accessing the service
@@ -48,9 +49,19 @@ public interface JMSService extends BeforeAllCallback {
      */
     void initialize();
 
+    /**
+     * Shuts down the service after the test has completed
+     */
+    void shutdown();
+
 
     @Override
     default void beforeAll(ExtensionContext extensionContext) throws Exception {
         initialize();
     }
+
+    @Override
+    default void afterAll(ExtensionContext extensionContext) throws Exception {
+        shutdown();
+    }
 }

tests/src/test/java/org/apache/camel/kafkaconnector/services/jms/RemoteJMSService.java

@@ -30,6 +30,11 @@ public void initialize() {
         // NO-OP
     }
 
+    @Override
+    public void shutdown() {
+        // NO-OP
+    }
+
     @Override
     public Properties getConnectionProperties() {
         return PropertyUtils.getProperties();

tests/src/test/java/org/apache/camel/kafkaconnector/sink/cassandra/CamelSinkCassandraITCase.java

@@ -43,15 +43,14 @@
 
 @Testcontainers
 public class CamelSinkCassandraITCase extends AbstractKafkaTest {
-    private static final Logger LOG = LoggerFactory.getLogger(CamelSinkCassandraITCase.class);
-
     @RegisterExtension
-    public CassandraService cassandraService = CassandraServiceFactory.createService();
+    public static CassandraService cassandraService = CassandraServiceFactory.createService();
+
+    private static final Logger LOG = LoggerFactory.getLogger(CamelSinkCassandraITCase.class);
 
     private CassandraClient cassandraClient;
     private TestDataDao testDataDao;
 
-
     private final int expect = 10;
     private int received;
 

tests/src/test/java/org/apache/camel/kafkaconnector/sink/elasticsearch/CamelSinkElasticSearchITCase.java

@@ -47,10 +47,10 @@
 
 @Testcontainers
 public class CamelSinkElasticSearchITCase extends AbstractKafkaTest {
-    private static final Logger LOG = LoggerFactory.getLogger(CamelElasticSearchPropertyFactory.class);
-
     @RegisterExtension
-    public ElasticSearchService elasticSearch = ElasticSearchServiceFactory.createService();
+    public static ElasticSearchService elasticSearch = ElasticSearchServiceFactory.createService();
+
+    private static final Logger LOG = LoggerFactory.getLogger(CamelElasticSearchPropertyFactory.class);
 
     private ElasticSearchClient client;
 

tests/src/test/java/org/apache/camel/kafkaconnector/sink/jms/CamelSinkJMSITCase.java

@@ -51,10 +51,10 @@
  */
 @Testcontainers
 public class CamelSinkJMSITCase extends AbstractKafkaTest {
-    private static final Logger LOG = LoggerFactory.getLogger(CamelSinkJMSITCase.class);
-
     @RegisterExtension
-    public JMSService jmsService = JMSServiceFactory.createService();
+    public static JMSService jmsService = JMSServiceFactory.createService();
+
+    private static final Logger LOG = LoggerFactory.getLogger(CamelSinkJMSITCase.class);
 
     private int received;
     private final int expect = 10;

tests/src/test/java/org/apache/camel/kafkaconnector/source/jms/CamelSourceJMSITCase.java

@@ -29,6 +29,7 @@
 import org.apache.camel.kafkaconnector.services.jms.JMSService;
 import org.apache.camel.kafkaconnector.services.jms.JMSServiceFactory;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.api.extension.RegisterExtension;
@@ -46,10 +47,10 @@
  */
 @Testcontainers
 public class CamelSourceJMSITCase extends AbstractKafkaTest {
-    private static final Logger LOG = LoggerFactory.getLogger(CamelSourceJMSITCase.class);
-
     @RegisterExtension
-    public JMSService jmsService = JMSServiceFactory.createService();
+    public static JMSService jmsService = JMSServiceFactory.createService();
+
+    private static final Logger LOG = LoggerFactory.getLogger(CamelSourceJMSITCase.class);
 
     private int received;
     private final int expect = 10;