From: Kai Moritz <kai@juplo.de>
Date: Sun, 14 Aug 2022 08:54:27 +0000 (+0200)
Subject: `ApplicationTest` auf basis der typisierbaren Basis neu implementiert
X-Git-Tag: sumup-adder---lvm-2-tage~9^2~9
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=2bf77d19d90e7356e1a7c6e13202971fd1b9897b;p=demos%2Fkafka%2Ftraining

`ApplicationTest` auf basis der typisierbaren Basis neu implementiert
---

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);
+              }
+            }
+          }
+        });
+  }
+}