From: Kai Moritz <kai@juplo.de>
Date: Mon, 7 Apr 2025 21:26:07 +0000 (+0200)
Subject: GRÜN: Existierende Offsets sind nicht die einzigen
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=1c5be32878793e9a61c77d54ba51ce60ca5e6a41;p=demos%2Fkafka%2Ftraining

GRÜN: Existierende Offsets sind nicht die einzigen
---

diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java
index 76e5278..a4dc033 100644
--- a/src/test/java/de/juplo/kafka/ApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/ApplicationTests.java
@@ -27,6 +27,7 @@ import org.springframework.kafka.support.SendResult;
 import org.springframework.kafka.test.context.EmbeddedKafka;
 
 import java.util.HashMap;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
@@ -66,8 +67,21 @@ public class ApplicationTests
   @FieldSource("PARTITIONS")
   void testExistingOffset(int partition) throws Exception
   {
-    SendResult<byte[], byte[]> result = send(partition);
+    List<SendResult<byte[], byte[]>> results = new LinkedList<>();
+    for (int i = 0; i < (partition + 1) * 7; i++)
+    {
+      SendResult<byte[], byte[]> result = send(partition);
+      if (i % (partition + 1) == 0)
+      {
+        results.add(result);
+      }
+    }
 
+    results.forEach(result -> fetchAndCheck(result));
+  }
+
+  private void fetchAndCheck(SendResult<byte[], byte[]> result)
+  {
     RecordMetadata recordMetadata = result.getRecordMetadata();
     ResponseEntity<String> response = fetchRecord(recordMetadata);
     check(result, response);