Workaround to KafkaEmbedded memory leaks using Kafka Connect.

apache · Jun 15, 2020 · b9c63c9 · b9c63c9
1 parent fea1abb
commit b9c63c9
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 2 deletions.
diff --git a/...test/java/org/apache/camel/kafkaconnector/common/services/kafka/EmbeddedKafkaService.java b/...test/java/org/apache/camel/kafkaconnector/common/services/kafka/EmbeddedKafkaService.java
@@ -25,6 +25,7 @@
 import org.apache.camel.kafkaconnector.common.PluginPathHelper;
 import org.apache.kafka.connect.runtime.WorkerConfig;
 import org.apache.kafka.connect.util.clusters.EmbeddedConnectCluster;
+import org.junit.jupiter.api.extension.ExtensionContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -95,6 +96,15 @@ public void shutdown() {
         }
     }
 
+    @Override
+    public void beforeTestExecution(ExtensionContext extensionContext) throws Exception {
+        initialize();
+    }
+
+    @Override
+    public void afterTestExecution(ExtensionContext context) throws Exception {
+        shutdown();
+    }
 
     // WARNING: this may come uninitialized
     public EmbeddedConnectCluster getCluster() {

diff --git a/...mon/src/test/java/org/apache/camel/kafkaconnector/common/services/kafka/KafkaService.java b/...mon/src/test/java/org/apache/camel/kafkaconnector/common/services/kafka/KafkaService.java
@@ -18,13 +18,15 @@
 package org.apache.camel.kafkaconnector.common.services.kafka;
 
 import org.junit.jupiter.api.extension.AfterAllCallback;
+import org.junit.jupiter.api.extension.AfterTestExecutionCallback;
 import org.junit.jupiter.api.extension.BeforeAllCallback;
+import org.junit.jupiter.api.extension.BeforeTestExecutionCallback;
 import org.junit.jupiter.api.extension.ExtensionContext;
 
 /**
  * Provides an interface for any type of Kafka service: remote instances, local container, etc
  */
-public interface KafkaService extends BeforeAllCallback, AfterAllCallback {
+public interface KafkaService extends BeforeAllCallback, BeforeTestExecutionCallback, AfterAllCallback, AfterTestExecutionCallback {
 
 
     /**
@@ -44,7 +46,6 @@ public interface KafkaService extends BeforeAllCallback, AfterAllCallback {
      */
     void shutdown();
 
-
     @Override
     default void beforeAll(ExtensionContext extensionContext) throws Exception {
         initialize();