feat: wait for poll step (#127)

Author: brian-mulier

tzatziki-spring-kafka/src/main/java/com/decathlon/tzatziki/kafka/KafkaInterceptor.java

@@ -1,5 +1,6 @@
 package com.decathlon.tzatziki.kafka;
 
+import com.decathlon.tzatziki.steps.KafkaSteps;
 import com.decathlon.tzatziki.utils.Fields;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -96,6 +97,9 @@ private Object createConsumerProxy(Consumer<?, ?> consumer) {
                 return switch (method.getName()) {
                     case "poll" -> {
                         ConsumerRecords<String, ?> consumerRecords = (ConsumerRecords<String, ?>) method.invoke(consumer, args);
+                        consumer.subscription().stream()
+                                .filter(KafkaSteps.semaphoreByTopic::containsKey)
+                                .forEach(topic -> KafkaSteps.semaphoreByTopic.remove(topic).release());
                         if (consumerRecords.count() == 0) {
                             yield consumerRecords;
                         }

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

@@ -29,6 +29,7 @@
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import org.junit.Assert;
+import org.junit.jupiter.api.Assertions;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
 import org.springframework.kafka.core.ConsumerFactory;
@@ -39,6 +40,7 @@
 import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Semaphore;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
@@ -70,6 +72,7 @@ public class KafkaSteps {
 
     private static boolean isStarted;
 
+    public static final Map<String, Semaphore> semaphoreByTopic = new LinkedHashMap<>();
     public static synchronized void start() {
         start(null);
     }
@@ -239,6 +242,19 @@ public void topic_has_been_consumed_on_every_partition(Guard guard, String group
         }))));
     }
 
+    @When(THAT + GUARD + "the " + VARIABLE + " topic was just polled$")
+    public void topic_was_just_polled(Guard guard, String topic) {
+        guard.in(objects, () -> {
+            Semaphore semaphore = new Semaphore(0);
+            semaphoreByTopic.put(topic, semaphore);
+            try {
+                semaphore.acquire();
+            } catch (InterruptedException e) {
+                Assertions.fail(e);
+            }
+        });
+    }
+
     @Given(THAT + "the current offset of " + VARIABLE + " on the topic " + VARIABLE + " is (\\d+)$")
     public void that_the_current_offset_the_groupid_on_topic_is(String groupId, String topic, long offset) throws ExecutionException, InterruptedException {
         try (Admin admin = Admin.create(avroConsumerFactories.get(0).getConfigurationProperties())) {

tzatziki-spring-kafka/src/test/resources/com/decathlon/tzatziki/steps/kafka.feature

@@ -406,4 +406,8 @@ Feature: to interact with a spring boot service having a connection to a kafka q
 
     Then the json-users-input topic contains 1 json message
 
-    But within 500ms the json-users-input topic contains 2 json messages
+    But within 500ms the json-users-input topic contains 2 json messages
+
+  @ignore
+  Scenario: we wait for a poll to occur on a specific topic
+    When the json-users-input topic was just polled