`ApplicationTest` auf basis der typisierbaren Basis neu implementiert

[demos/kafka/training] / 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 (file)
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);
+              }
+            }
+          }
+        });
+  }
+}