Add logging and fix topic name in authorized client test

Author: geso02

basyx.common/basyx.kafka/src/main/java/org/eclipse/digitaltwin/basyx/kafka/KafkaAdapter.java

@@ -39,8 +39,13 @@
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.eclipse.digitaltwin.aas4j.v3.dataformat.core.DeserializationException;
 import org.eclipse.digitaltwin.aas4j.v3.dataformat.json.JsonDeserializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class KafkaAdapter<T> {
+	
+	private static final Logger LOG = LoggerFactory.getLogger(KafkaAdapter.class);
+	
 	private final KafkaConsumer<String, String> consumer;
 
 	private final String bootstrapServers;
@@ -92,33 +97,38 @@ private KafkaConsumer<String, String> init() {
 	}
 
 	private void awaitAssignment() {
-		long deadline = System.currentTimeMillis() + assignmentTimeout.toMillis();
+		LOG.info("Await Assignment");
+		long start = System.currentTimeMillis();
+		long deadline = start + assignmentTimeout.toMillis();
 		while (consumer.assignment().isEmpty() && System.currentTimeMillis() < deadline) {
 			consumer.poll(Duration.ofMillis(100));
 		}
 		if (consumer.assignment().isEmpty()) {
 			throw new RuntimeException("Failed to wait for topic assignment. Is KAFKA running?");
 		}
+		LOG.info("Partitions {} assigned after {} ms." + consumer.assignment(), System.currentTimeMillis() - start);
 	}
 
 	private String nextMessage() {
 		return nextMessage(pollTimeout);
 	}
 
 	private String nextMessage(Duration duration) {
-
+		LOG.info("Reading Kafka message");
 		long deadline = System.currentTimeMillis() + duration.toMillis();
 
 		while (deque.isEmpty() && System.currentTimeMillis() < deadline) {
-			ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
+			ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
 			for (ConsumerRecord<String, String> record : records) {
 				this.deque.add(record.value());
-				consumer.commitSync();
+				consumer.commitAsync();
 			}
 		}
 		if (!deque.isEmpty()) {
+			LOG.info("Got message");	
 			return deque.remove();
 		}
+		LOG.info("Failed to receive message");
 		return null;
 	}
 
@@ -136,17 +146,19 @@ public T next() {
 	}
 
 	public void assertNoAdditionalMessages() {
-		String next = nextMessage(Duration.ofSeconds(1)); 
+		String next = nextMessage(Duration.ofMillis(100)); 
 		if (next != null) {
 			throw new RuntimeException("Got an additional message within 1 second: \n" + next); 
 		}
 	}
 
 	public void close() {
+		LOG.info("Dispose");
 		this.consumer.close();
 	}
 
 	public void skipMessages() {
+		LOG.info("SkipMessages");
 		while (nextMessage(Duration.ofMillis(100)) != null);
 	}
 

basyx.submodelregistry/basyx.submodelregistry-service-release-kafka-mongodb/src/test/java/org/eclipse/digitaltwin/basyx/submodelregistry/service/storage/mongodb/AuthorizedClientTest.java

@@ -56,7 +56,7 @@
 public class AuthorizedClientTest extends BaseIntegrationTest {
 
 
-	private final KafkaAdapter<RegistryEvent> adapter = new KafkaAdapter<>("localhost:9092", "submodel-events", RegistryEvent.class);
+	private final KafkaAdapter<RegistryEvent> adapter = new KafkaAdapter<>("localhost:9092", "submodel-registry", RegistryEvent.class);
 
 	@Value("${local.server.port}")
 	private int port;