Clean up previous messages (#742)

Author: geso02

basyx.aasrepository/basyx.aasrepository-feature-kafka/src/test/java/org/eclipse/digitaltwin/basyx/aasrepository/feature/kafka/KafkaEventsInMemoryStorageIntegrationTest.java

@@ -86,6 +86,7 @@ public class KafkaEventsInMemoryStorageIntegrationTest {
 	@Before
 	public void awaitAssignment() throws InterruptedException {
 		listener.awaitTopicAssignment();
+		cleanupPreviousMessages();
 		FileRepository fileRepo = new InMemoryFileRepository();
 		AasBackend aasRepositoryBackend = new InMemoryAasBackend();
 		AasServiceFactory sf = new CrudAasServiceFactory(aasRepositoryBackend, fileRepo);
@@ -94,7 +95,9 @@ public void awaitAssignment() throws InterruptedException {
 
 		cleanup();
 	}
-
+	private void cleanupPreviousMessages() throws InterruptedException {
+		while (listener.next(1, TimeUnit.SECONDS) != null);	
+	}
 
 	@Test
 	public void testCreateAas() throws InterruptedException {
@@ -245,9 +248,7 @@ public void cleanup() throws InterruptedException {
 			Assert.assertEquals(AasEventType.AAS_DELETED, deletedEvt.getType());
 			Assert.assertEquals(aas.getId(), deletedEvt.getId());
 		}
-	}
-	@After
-	public void assertNoAdditionalEvent() throws InterruptedException {
-		Assert.assertNull(listener.next(1, TimeUnit.SECONDS));
+        AasEvent evt = listener.next(1, TimeUnit.SECONDS);
+		Assert.assertNull(evt);
 	}
 }

basyx.submodelrepository/basyx.submodelrepository-feature-kafka/src/test/java/org/eclipse/digitaltwin/basyx/submodelrepository/feature/kafka/KafkaEventsInMemoryStorageIntegrationTest.java

@@ -105,9 +105,13 @@ public class KafkaEventsInMemoryStorageIntegrationTest {
 	public void awaitAssignment() throws InterruptedException {
 		listener.awaitTopicAssignment();
 		repo = feature.decorate(factory).create();
-		
+		cleanupPreviousMessages();
 		cleanup();
 	}
+	
+	private void cleanupPreviousMessages() throws InterruptedException {
+		while (listener.next(1, TimeUnit.SECONDS) != null);	
+	}
 
 	@Test
 	public void testSubmodelCreated() throws InterruptedException {
@@ -351,6 +355,8 @@ public void cleanup() throws InterruptedException {
 				Assert.assertEquals(sm.getId(), evt.getId());		
 			} 
 		}
+		SubmodelEvent evt = listener.next(1, TimeUnit.SECONDS);
+		Assert.assertNull(evt);
 	}
 
 

basyx.submodelservice/basyx.submodelservice-feature-kafka/src/test/java/org/eclipse/digitaltwin/basyx/submodelservice/feature/kafka/KafkaSubmodelServiceIdsOnlySmokeTest.java

@@ -89,14 +89,18 @@ public class KafkaSubmodelServiceIdsOnlySmokeTest {
 	public void awaitAssignment() throws InterruptedException, SerializationException {
 		listener.awaitTopicAssignment();
 
-		assertNoAdditionalMessage();
+		cleanupPreviousMessages();
 
 		FileRepository repository = new InMemoryFileRepository();
 		SubmodelBackend backend = new InMemorySubmodelBackend();
 		SubmodelServiceFactory smFactory = new CrudSubmodelServiceFactory(backend ,repository);
 		service = feature.decorate(smFactory).create(submodel);
 	}
 
+	private void cleanupPreviousMessages() throws InterruptedException {
+		while (listener.next(1, TimeUnit.SECONDS) != null);	
+	}
+
 	@After
 	public void assertNoAdditionalMessage() throws InterruptedException {
 		SubmodelEvent evt = listener.next(1, TimeUnit.SECONDS);

basyx.submodelservice/basyx.submodelservice-feature-kafka/src/test/java/org/eclipse/digitaltwin/basyx/submodelservice/feature/kafka/KafkaSubmodelServiceSubmodelElementsEventsIntegrationTest.java

@@ -90,18 +90,16 @@ public class KafkaSubmodelServiceSubmodelElementsEventsIntegrationTest {
 	public void awaitAssignment() throws InterruptedException {
 		listener.awaitTopicAssignment();
 
-		skipAdditionalMessage();
+		cleanupPreviousMessages();
 
 		FileRepository repository = new InMemoryFileRepository();
 		SubmodelBackend backend = new InMemorySubmodelBackend();
 		SubmodelServiceFactory smFactory = new CrudSubmodelServiceFactory(backend ,repository);
 		service = feature.decorate(smFactory).create(submodel);
 	}
-
-	@After
-	public void skipAdditionalMessage() throws InterruptedException {
-		while(listener.next(1, TimeUnit.SECONDS) != null);
-
+	
+	private void cleanupPreviousMessages() throws InterruptedException {
+		while (listener.next(1, TimeUnit.SECONDS) != null);	
 	}
 
 	@Test

basyx.submodelservice/basyx.submodelservice-feature-kafka/src/test/java/org/eclipse/digitaltwin/basyx/submodelservice/feature/kafka/KafkaSubmodelServiceSubmodelEventsIntegrationTest.java

@@ -81,12 +81,6 @@ public void awaitAssignment() throws InterruptedException {
 		listener.awaitTopicAssignment();
 	}
 
-	@After
-	public void assertGotTearDownMessage() throws InterruptedException {
-		SubmodelEvent evt = listener.next(1, TimeUnit.MINUTES);
-		Assert.assertNull(evt);
-	}
-
 	@Test
 	public void testSubmodelEvents() throws InterruptedException {
 		// we expect the "onStartup" submodel created event