Tests: Refaktorisiert - Nachrichten werden für alle Tests aufgezeichnet

[demos/kafka/training] / src / test / java / de / juplo / kafka / ApplicationTests.java

diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java
index f541093..4cc4f91 100644 (file)
--- a/src/test/java/de/juplo/kafka/ApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/ApplicationTests.java
@@ -69,6 +69,7 @@ class ApplicationTests
        EndlessConsumer<String, Long> endlessConsumer;
        Map<TopicPartition, Long> oldOffsets;
        Map<TopicPartition, Long> newOffsets;
+       Set<ConsumerRecord<String, Long>> receivedRecords;
 
 
        /** Tests methods */
@@ -79,12 +80,9 @@ class ApplicationTests
        {
                send100Messages(i ->  new Bytes(longSerializer.serialize(TOPIC, i)));
 
-               Set<ConsumerRecord<String, Long>> received = new HashSet<>();
-               testHandler = record -> received.add(record);
-
                await("100 records received")
                                .atMost(Duration.ofSeconds(30))
-                               .until(() -> received.size() >= 100);
+                               .until(() -> receivedRecords.size() >= 100);
 
                await("Offsets committed")
                                .atMost(Duration.ofSeconds(10))
@@ -118,6 +116,9 @@ class ApplicationTests
 
                checkSeenOffsetsForProgress();
                compareToCommitedOffsets(newOffsets);
+               assertThat(receivedRecords.size())
+                               .describedAs("Received not all sent events")
+                               .isLessThan(100);
        }
 
 
@@ -223,6 +224,7 @@ class ApplicationTests
 
                oldOffsets = new HashMap<>();
                newOffsets = new HashMap<>();
+               receivedRecords = new HashSet<>();
 
                doForCurrentOffsets((tp, offset) ->
                {
@@ -236,6 +238,7 @@ class ApplicationTests
                                        newOffsets.put(
                                                        new TopicPartition(record.topic(), record.partition()),
                                                        record.offset());
+                                       receivedRecords.add(record);
                                        testHandler.accept(record);
                                };