juplo.de Git - demos/kafka/training/blob - src/test/java/de/juplo/kafka/ApplicationTest.java

   1 package de.juplo.kafka;
   2
   3 import org.apache.kafka.clients.producer.ProducerRecord;
   4 import org.apache.kafka.common.serialization.LongSerializer;
   5 import org.apache.kafka.common.serialization.StringSerializer;
   6 import org.apache.kafka.common.utils.Bytes;
   7
   8 import java.util.Set;
   9 import java.util.function.Consumer;
  10
  11
  12 public class ApplicationTest extends GenericApplicationTest<String, Long>
  13 {
  14   public ApplicationTest()
  15   {
  16     super(
  17         new RecordGenerator()
  18         {
  19           final StringSerializer stringSerializer = new StringSerializer();
  20           final LongSerializer longSerializer = new LongSerializer();
  21
  22
  23           @Override
  24           public void generate(
  25               int numberOfMessagesToGenerate,
  26               Set<Integer> poisonPills,
  27               Consumer<ProducerRecord<Bytes, Bytes>> messageSender)
  28           {
  29             int i = 0;
  30
  31             for (int partition = 0; partition < 10; partition++)
  32             {
  33               for (int key = 0; key < 10; key++)
  34               {
  35                 if (++i > numberOfMessagesToGenerate)
  36                   return;
  37
  38                 Bytes value =
  39                     poisonPills.contains(i)
  40                         ? new Bytes(stringSerializer.serialize(TOPIC, "BOOM!"))
  41                         : new Bytes(longSerializer.serialize(TOPIC, (long)i));
  42
  43                 ProducerRecord<Bytes, Bytes> record =
  44                     new ProducerRecord<>(
  45                         TOPIC,
  46                         partition,
  47                         new Bytes(stringSerializer.serialize(TOPIC,Integer.toString(partition*10+key%2))),
  48                         value);
  49
  50                 messageSender.accept(record);
  51               }
  52             }
  53           }
  54         });
  55   }
  56 }