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

   1 package de.juplo.kafka;
   2
   3 import lombok.RequiredArgsConstructor;
   4 import org.apache.kafka.common.TopicPartition;
   5
   6 import java.util.List;
   7 import java.util.Map;
   8
   9
  10 @RequiredArgsConstructor
  11 public class TestRecordHandler implements RecordHandler
  12 {
  13   private final RecordHandler handler;
  14
  15   Map<TopicPartition, Long> seenOffsets;
  16   int receivedMessages;
  17
  18
  19   public void onNewRecord(
  20       String topic,
  21       Integer partition,
  22       Long offset,
  23       Message messgage)
  24   {
  25     seenOffsets.put(new TopicPartition(topic, partition), offset);
  26     receivedMessages++;
  27   }
  28
  29   @Override
  30   public void addNumber(
  31       String topic,
  32       Integer partition,
  33       Long offset,
  34       String user,
  35       MessageAddNumber message)
  36   {
  37     this.onNewRecord(topic, partition, offset, message);
  38     handler.addNumber(topic, partition, offset, user, message);
  39   }
  40
  41   @Override
  42   public void calculateSum(
  43       String topic,
  44       Integer partition,
  45       Long offset,
  46       String user,
  47       MessageCalculateSum message)
  48   {
  49     this.onNewRecord(topic, partition, offset, message);
  50     handler.calculateSum(topic, partition, offset, user, message);
  51   }
  52 }