Merge branch 'deserialization' into sumup-adder--ohne--stored-offsets

[demos/kafka/training] / src / test / java / de / juplo / kafka / TestRecordHandler.java

diff --git a/src/test/java/de/juplo/kafka/TestRecordHandler.java b/src/test/java/de/juplo/kafka/TestRecordHandler.java
index de28385..37d3f65 100644 (file)
--- a/src/test/java/de/juplo/kafka/TestRecordHandler.java
+++ b/src/test/java/de/juplo/kafka/TestRecordHandler.java
@@ -2,16 +2,28 @@ package de.juplo.kafka;
 
 import lombok.RequiredArgsConstructor;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.Map;
+import java.util.Set;
 
 
 @RequiredArgsConstructor
-public abstract class TestRecordHandler<K, V> implements RecordHandler<K, V>
+public class TestRecordHandler<K, V> implements RecordHandler<K, V>
 {
   private final RecordHandler<K, V> handler;
 
+  Map<TopicPartition, Long> seenOffsets;
+  Set<ConsumerRecord<K, V>> receivedRecords;
 
-  public abstract void onNewRecord(ConsumerRecord<K, V> record);
 
+  public void onNewRecord(ConsumerRecord<K, V> record)
+  {
+    seenOffsets.put(
+      new TopicPartition(record.topic(), record.partition()),
+      record.offset());
+    receivedRecords.add(record);
+  }
 
   @Override
   public void accept(ConsumerRecord<K, V> record)
@@ -19,10 +31,4 @@ public abstract class TestRecordHandler<K, V> implements RecordHandler<K, V>
     this.onNewRecord(record);
     handler.accept(record);
   }
-  @Override
-
-  public void beforeNextPoll()
-  {
-    handler.beforeNextPoll();
-  }
 }