test case specific kafka consumers

Robert Diers · Robert Diers · commit 4de18db218a3 · 2022-07-25T09:13:23.000+02:00
diff --git a/README.md b/README.md
@@ -17,6 +17,7 @@ Container:   https://github.com/opentestingapi/impl_java/pkgs/container/opentest
 | 1.8     | 1.0 | data generator endpoint to create mass data from testcases                                                                                                   |
 | 1.9     | 1.0 | dependency upgrades, CVE fix: https://github.com/aws/aws-sdk-java/security/advisories/GHSA-c28r-hw5m-5gv3                                                    |
 | 1.10    | 1.0 | endpoint to export all test cases                                                                                                                            |
+| 1.11    | 1.0 | Kafka consumers not shared between test cases, they have to use different group.id (recreated with every upload)                                             |
 
 ## Architecture
 
diff --git a/src/main/java/org/opentesting/services/adapter/kafka/Kafka.java b/src/main/java/org/opentesting/services/adapter/kafka/Kafka.java
@@ -196,7 +196,8 @@ public void createRequiredComponents(TestCaseDTO test) {
         // create required Kafka consumer
         for (TestCaseCheckDTO check : test.getChecks()) {
             if (getAllServicenames().contains(check.getService().getType()) && check.isActive()) {
-                kafkaConsumerFactory.createConsumer(check.getService(), this);
+                kafkaConsumerFactory.closeConsumer(test.getId(), check.getService(), this);
+                kafkaConsumerFactory.createConsumer(test.getId(), check.getService(), this);
             }
         }
 
diff --git a/src/main/java/org/opentesting/services/adapter/kafka/OpenTestingKafkaConsumer.java b/src/main/java/org/opentesting/services/adapter/kafka/OpenTestingKafkaConsumer.java
@@ -4,6 +4,7 @@
 import java.util.Map;
 
 import org.opentesting.dto.TestCaseServiceDTO;
+import org.opentesting.dto.api1dot0.TestCaseServiceDTOapi1dot0;
 import org.opentesting.services.adapter.Adapter;
 import org.opentesting.services.encryption.Encryption;
 import org.opentesting.services.execution.TestCheck;
@@ -42,10 +43,10 @@ public class OpenTestingKafkaConsumer {
     private Map<String, KafkaMessageListenerContainer<String, String>> createdConsumers = new HashMap<>();
 
     @LogExecutionTime
-    public void createConsumer(TestCaseServiceDTO service, Adapter adapter) {
+    public void createConsumer(String testid, TestCaseServiceDTO service, Adapter adapter) {
        
-        //shared between test cases
-        String key = service.getConnectstring()+"#"+service.getCustom("topic").getValue()+"#"+service.getUsername()+"#"+service.getCustom("group.id").getValue();
+        //we will not share between test ids, they need to use different group.ids
+        String key = createUniqueKey(testid, service);
         if (createdConsumers.containsKey(key)) {
             log.info("kafka consumer already exists: "+key);
             return;
@@ -71,7 +72,7 @@ public void createConsumer(TestCaseServiceDTO service, Adapter adapter) {
 
         //consumer container
         ContainerProperties containerProps = new ContainerProperties(service.getCustom("topic").getValue());
-        containerProps.setMessageListener(new OpenTestingKafkaMessageListener(service, testCheck, key, prometheus, adapter, kafkaTrace));
+        containerProps.setMessageListener(new OpenTestingKafkaMessageListener(testid, service, testCheck, key, prometheus, adapter, kafkaTrace));
         KafkaMessageListenerContainer<String, String> container = new KafkaMessageListenerContainer<>(fact, containerProps);
 
         //start container
@@ -91,6 +92,40 @@ public void createConsumer(TestCaseServiceDTO service, Adapter adapter) {
         log.info("KafkaConsumer created: "+key);
     }
 
+    /**
+     * close consumer
+     * @param testid
+     * @param service
+     * @param kafka
+     */
+    @LogExecutionTime
+    public void closeConsumer(String testid, TestCaseServiceDTOapi1dot0 service, Kafka kafka) {
+        String key = createUniqueKey(testid, service);
+        if (createdConsumers.containsKey(key)) {
+            log.info("closing kafka consumer: "+key);
+            createdConsumers.get(key).destroy();
+            createdConsumers.remove(key);
+        }
+    }
+
+    /**
+     * close consumer
+     * @param testid
+     */
+    @LogExecutionTime
+    public void closeConsumer(String testid) {
+        for (String key : createdConsumers.keySet()) {            
+            if (key.startsWith(testid+"#")) {
+                log.info("closing kafka consumer: "+key);
+                createdConsumers.get(key).destroy();
+            }
+        }        
+    }
+
+    private String createUniqueKey(String testid, TestCaseServiceDTO service) {
+        return testid+"#"+service.getConnectstring()+"#"+service.getCustom("topic").getValue()+"#"+service.getUsername()+"#"+service.getCustom("group.id").getValue();
+    }
+
     /**
      * stop consumers
      */
@@ -125,6 +160,6 @@ public void startConsumers() {
                     }
                 }
         );
-    }
+    }    
 
 }
diff --git a/src/main/java/org/opentesting/services/adapter/kafka/OpenTestingKafkaMessageListener.java b/src/main/java/org/opentesting/services/adapter/kafka/OpenTestingKafkaMessageListener.java
@@ -27,15 +27,17 @@ public class OpenTestingKafkaMessageListener implements MessageListener<String,
 
     private static final String TOPIC = "topic";
 
+    private String testid;
     private TestCaseServiceDTO service;
     private TestCheck testCheck;
     private String key;
     private Prometheus prometheus;
     private Adapter adapter;    
     private KafkaTrace kafkatrace;
 
-    public OpenTestingKafkaMessageListener(TestCaseServiceDTO service, TestCheck testCheck, String key, 
+    public OpenTestingKafkaMessageListener(String testid, TestCaseServiceDTO service, TestCheck testCheck, String key, 
             Prometheus prometheus, Adapter adapter, KafkaTrace kafkatrace) {
+        this.testid = testid;
         this.service = service;
         this.testCheck = testCheck;
         this.key = key;
@@ -49,7 +51,7 @@ public OpenTestingKafkaMessageListener(TestCaseServiceDTO service, TestCheck tes
     public void onMessage(ConsumerRecord<String, String> data) {
 
         //read checks
-        List<TestCaseCheckDTO> checks = getChecksByServiceAndConnectstringAndOpen(service); 
+        List<TestCaseCheckDTO> checks = getChecksByServiceAndConnectstringAndOpen(testid, service); 
 
         //extract Trace info
         final TraceDTO trace = kafkatrace.extractTraceInfo(data);
@@ -81,14 +83,14 @@ public void onMessage(ConsumerRecord<String, String> data) {
      * @throws DataAccessException
      */
     @LogExecutionTime
-    public List<TestCaseCheckDTO> getChecksByServiceAndConnectstringAndOpen(TestCaseServiceDTO service) {
+    public List<TestCaseCheckDTO> getChecksByServiceAndConnectstringAndOpen(String testid, TestCaseServiceDTO service) {
 
         List<TestCaseCheckDTO> openKafkaChecks = new ArrayList<>();
         String topic = service.getCustom(TOPIC).getValue();
 
         //filter by topic
         try {
-            for (TestCaseCheckDTO check : testCheck.getOpenChecks(service)) {
+            for (TestCaseCheckDTO check : testCheck.getOpenChecks(testid, service)) {
                 if (check.getService().getCustom(TOPIC).getValue().equals(topic)) {
                     openKafkaChecks.add(check);
                 }
diff --git a/src/main/java/org/opentesting/services/execution/TestCheck.java b/src/main/java/org/opentesting/services/execution/TestCheck.java
@@ -223,11 +223,25 @@ public List<TestCaseCheckDTO> getOpenChecks(TestCaseServiceDTO service) throws D
         return getOpenChecks(service.getType());
     }
 
+    /**
+     * read all open checks for a defined type
+     */
+    public List<TestCaseCheckDTO> getOpenChecks(String testid, TestCaseServiceDTO service) throws DataAccessException {
+        return getOpenChecks(testid, service.getType());
+    }
+
     /**
      * read all open checks for a defined type
      */
     public List<TestCaseCheckDTO> getOpenChecks(String service) throws DataAccessException {
         return checkRepository.selectByOpentype(service);
     }
+
+    /**
+     * read all open checks for a defined type
+     */
+    public List<TestCaseCheckDTO> getOpenChecks(String testid, String service) throws DataAccessException {
+        return checkRepository.selectByTestAndOpentype(testid, service);
+    }
     
 }
diff --git a/src/main/java/org/opentesting/services/execution/TestManager.java b/src/main/java/org/opentesting/services/execution/TestManager.java
@@ -5,6 +5,7 @@
 import org.opentesting.persistence.IndexRepository;
 import org.opentesting.services.adapter.Adapter;
 import org.opentesting.services.adapter.AdapterResolver;
+import org.opentesting.services.adapter.kafka.OpenTestingKafkaConsumer;
 import org.opentesting.services.pause.Pause;
 import org.opentesting.services.scheduler.AsynTaskExec;
 import org.opentesting.services.scheduler.TimerFactory;
@@ -41,6 +42,9 @@ public class TestManager extends TestAbstractHandler {
     @Autowired
     private TestCheck testCheck;
 
+    @Autowired
+    private OpenTestingKafkaConsumer openTestingKafkaConsumer;
+
      /**
      * adapters might need timers
      */
@@ -61,6 +65,10 @@ public void scheduleAdapterTimers() {
      */
     @LogExecutionTime
     public String removeTest(String testid) {
+
+        //remove Kafka consumer
+        openTestingKafkaConsumer.closeConsumer(testid);
+
         StringBuilder st = new StringBuilder();
         indexRepository.deleteByValue(IndexRepository.LABELINDEX, testid);
         st.append("index deleted");

Original file line number	Diff line number	Diff line change
`@@ -196,7 +196,8 @@ public void createRequiredComponents(TestCaseDTO test) {`
`196`	`196`	`// create required Kafka consumer`
`197`	`197`	`for (TestCaseCheckDTO check : test.getChecks()) {`
`198`	`198`	`if (getAllServicenames().contains(check.getService().getType()) && check.isActive()) {`
`199`		`- kafkaConsumerFactory.createConsumer(check.getService(), this);`
	`199`	`+ kafkaConsumerFactory.closeConsumer(test.getId(), check.getService(), this);`
	`200`	`+ kafkaConsumerFactory.createConsumer(test.getId(), check.getService(), this);`
`200`	`201`	`}`
`201`	`202`	`}`
`202`	`203`