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