From 2bf77d19d90e7356e1a7c6e13202971fd1b9897b Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Sun, 14 Aug 2022 10:54:27 +0200
Subject: [PATCH] `ApplicationTest` auf basis der typisierbaren Basis neu
 implementiert

---
 .../java/de/juplo/kafka/ApplicationTest.java  | 57 +++++++++++++++++++
 1 file changed, 57 insertions(+)
 create mode 100644 src/test/java/de/juplo/kafka/ApplicationTest.java

diff --git a/src/test/java/de/juplo/kafka/ApplicationTest.java b/src/test/java/de/juplo/kafka/ApplicationTest.java
new file mode 100644
index 0000000..d3ff3b1
--- /dev/null
+++ b/src/test/java/de/juplo/kafka/ApplicationTest.java
@@ -0,0 +1,57 @@
+package de.juplo.kafka;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.LongSerializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.utils.Bytes;
+
+import java.util.Set;
+import java.util.function.Consumer;
+
+
+public class ApplicationTest extends GenericApplicationTest<String, Long>
+{
+  public ApplicationTest()
+  {
+    super(
+        new StringSerializer(),
+        new RecordGenerator<> ()
+        {
+          final StringSerializer stringSerializer = new StringSerializer();
+          final LongSerializer longSerializer = new LongSerializer();
+
+
+          @Override
+          public void generate(
+              int numberOfMessagesToGenerate,
+              Set<Integer> poisonPills,
+              Consumer<ProducerRecord<String, Bytes>> messageSender)
+          {
+            int i = 0;
+
+            for (int partition = 0; partition < 10; partition++)
+            {
+              for (int key = 0; key < 10; key++)
+              {
+                if (++i > numberOfMessagesToGenerate)
+                  return;
+
+                Bytes value =
+                    poisonPills.contains(i)
+                        ? new Bytes(stringSerializer.serialize(TOPIC, "BOOM!"))
+                        : new Bytes(longSerializer.serialize(TOPIC, (long)i));
+
+                ProducerRecord<String, Bytes> record =
+                    new ProducerRecord<>(
+                        TOPIC,
+                        partition,
+                        Integer.toString(partition*10+key%2),
+                        value);
+
+                messageSender.accept(record);
+              }
+            }
+          }
+        });
+  }
+}
-- 
2.20.1