fx: Tests fixes + allow non-avro apps to assert full topic consumption (#153)

Author: brian-mulier

tzatziki-spring-kafka/src/main/java/com/decathlon/tzatziki/steps/KafkaSteps.java

@@ -192,7 +192,7 @@ public void a_message_is_consumed_from_a_topic(Guard guard, String name, boolean
         guard.in(objects, () -> {
             KafkaInterceptor.awaitForSuccessfullOnly = successfully;
             if (!checkedTopics.contains(topic)) {
-                try (Admin admin = Admin.create(avroConsumerFactories.get(0).getConfigurationProperties())) {
+                try (Admin admin = Admin.create(getAnyConsumerFactory().getConfigurationProperties())) {
                     awaitUntil(() -> {
                         List<String> groupIds = admin.listConsumerGroups().all().get().stream().map(ConsumerGroupListing::groupId).toList();
                         Map<String, KafkaFuture<ConsumerGroupDescription>> groupDescriptions = admin.describeConsumerGroups(groupIds).describedGroups();
@@ -222,6 +222,11 @@ public void a_message_is_consumed_from_a_topic(Guard guard, String name, boolean
         });
     }
 
+    @NotNull
+    private ConsumerFactory<String, ?> getAnyConsumerFactory() {
+        return Stream.concat(Stream.concat(jsonConsumerFactories.stream(), avroConsumerFactories.stream()), avroJacksonConsumerFactories.stream()).findFirst().get();
+    }
+
     @SneakyThrows
     @When(THAT + GUARD + "the " + VARIABLE + " group id has fully consumed the " + VARIABLE + " topic$")
     public void topic_has_been_consumed_on_every_partition(Guard guard, String groupId, String topic) {

tzatziki-spring-kafka/src/test/java/com/decathlon/tzatziki/kafka/KafkaUsersListener.java

@@ -1,25 +1,23 @@
 package com.decathlon.tzatziki.kafka;
 
-import java.util.List;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.avro.generic.GenericRecord;
-import org.springframework.boot.test.mock.mockito.SpyBean;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.listener.AbstractConsumerSeekAware;
 import org.springframework.kafka.listener.BatchListenerFailedException;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.handler.annotation.Payload;
 import org.springframework.stereotype.Service;
 
-import static org.springframework.kafka.support.KafkaHeaders.OFFSET;
-import static org.springframework.kafka.support.KafkaHeaders.RECEIVED_KEY;
-import static org.springframework.kafka.support.KafkaHeaders.RECEIVED_PARTITION;
-import static org.springframework.kafka.support.KafkaHeaders.RECEIVED_TOPIC;
+import java.util.List;
+
+import static org.springframework.kafka.support.KafkaHeaders.*;
 
 @Service
 @Slf4j
 public class KafkaUsersListener extends AbstractConsumerSeekAware implements Seeker {
-    @SpyBean
+    @Autowired
     private CountService countService;
 
     @KafkaListener(topics = "users", groupId = "users-group-id", containerFactory = "defaultFactory")

tzatziki-spring-kafka/src/test/java/com/decathlon/tzatziki/steps/TestApplicationSteps.java

@@ -14,6 +14,7 @@
 import org.mockito.Mockito;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.boot.test.util.TestPropertyValues;
 import org.springframework.context.ApplicationContextInitializer;
 import org.springframework.context.ConfigurableApplicationContext;
@@ -36,7 +37,8 @@ public void initialize(ConfigurableApplicationContext configurableApplicationCon
             TestPropertyValues.of(
                     "spring.kafka.bootstrap-servers=" + KafkaSteps.bootstrapServers(),
                     "spring.kafka.consumer.properties.fetch.min.bytes=100000",
-                    "spring.kafka.consumer.properties.fetch.max.wait.ms=1"
+                    "spring.kafka.consumer.properties.fetch.max.wait.ms=1",
+                    "spring.kafka.consumer.auto-offset-reset=earliest"
             ).applyTo(configurableApplicationContext.getEnvironment());
             KafkaSteps.autoSeekTopics("exposed-users", "json-users");
         }
@@ -48,7 +50,7 @@ public TestApplicationSteps(ObjectSteps objects) {
         this.objects = objects;
     }
 
-    @Autowired
+    @SpyBean
     CountService spyCountService;
 
     @Autowired