X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FEndlessConsumer.java;fp=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FEndlessConsumer.java;h=0c107f3b261965f77cf4ac9f327ca510d39a3458;hb=2d84eda74475aaffff11ddfebe56d309b9aff2e9;hp=01f9057a055aa360bb8d98cca1641073df19afe0;hpb=a6a0a22a5fa34a01b0e8b2bc1e0e2b82d7b60f33;p=demos%2Fkafka%2Ftraining

diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java
index 01f9057..0c107f3 100644
--- a/src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java
@@ -8,32 +8,24 @@ import org.apache.kafka.common.errors.RecordDeserializationException;
 import org.apache.kafka.common.errors.WakeupException;
 
 import javax.annotation.PreDestroy;
-import java.time.Clock;
 import java.time.Duration;
-import java.time.Instant;
 import java.util.*;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
-import java.util.regex.Pattern;
 
 
 @Slf4j
 @RequiredArgsConstructor
-public class EndlessConsumer implements ConsumerRebalanceListener, Runnable
+public class EndlessConsumer<K, V> implements ConsumerRebalanceListener, Runnable
 {
-  final static Pattern PATTERN = Pattern.compile("\\W+");
-
-
   private final ExecutorService executor;
-  private final PartitionStatisticsRepository repository;
   private final String id;
   private final String topic;
-  private final Clock clock;
-  private final Duration commitInterval;
-  private final Consumer<String, String> consumer;
+  private final Consumer<K, V> consumer;
+  private final RecordHandler<K, V> handler;
 
   private final Lock lock = new ReentrantLock();
   private final Condition condition = lock.newCondition();
@@ -41,46 +33,17 @@ public class EndlessConsumer implements ConsumerRebalanceListener, Runnable
   private Exception exception;
   private long consumed = 0;
 
-  private final Map<Integer, Map<String, Map<String, Long>>> seen = new HashMap<>();
-
 
   @Override
   public void onPartitionsRevoked(Collection<TopicPartition> partitions)
   {
-    partitions.forEach(tp ->
-    {
-      Integer partition = tp.partition();
-      Long newOffset = consumer.position(tp);
-      log.info(
-          "{} - removing partition: {}, offset of next message {})",
-          id,
-          partition,
-          newOffset);
-      Map<String, Map<String, Long>> removed = seen.remove(partition);
-      repository.save(new StatisticsDocument(partition, removed, consumer.position(tp)));
-    });
+    partitions.forEach(tp -> handler.onPartitionRevoked(tp));
   }
 
   @Override
   public void onPartitionsAssigned(Collection<TopicPartition> partitions)
   {
-    partitions.forEach(tp ->
-    {
-      Integer partition = tp.partition();
-      Long offset = consumer.position(tp);
-      log.info("{} - adding partition: {}, offset={}", id, partition, offset);
-      StatisticsDocument document =
-          repository
-              .findById(Integer.toString(partition))
-              .orElse(new StatisticsDocument(partition));
-      if (document.offset >= 0)
-      {
-        // Only seek, if a stored offset was found
-        // Otherwise: Use initial offset, generated by Kafka
-        consumer.seek(tp, document.offset);
-      }
-      seen.put(partition, document.statistics);
-    });
+    partitions.forEach(tp -> handler.onPartitionAssigned(tp));
   }
 
 
@@ -92,16 +55,14 @@ public class EndlessConsumer implements ConsumerRebalanceListener, Runnable
       log.info("{} - Subscribing to topic {}", id, topic);
       consumer.subscribe(Arrays.asList(topic), this);
 
-      Instant lastCommit = clock.instant();
-
       while (true)
       {
-        ConsumerRecords<String, String> records =
+        ConsumerRecords<K, V> records =
             consumer.poll(Duration.ofSeconds(1));
 
         // Do something with the data...
         log.info("{} - Received {} messages", id, records.count());
-        for (ConsumerRecord<String, String> record : records)
+        for (ConsumerRecord<K, V> record : records)
         {
           log.info(
               "{} - {}: {}/{} - {}={}",
@@ -113,44 +74,12 @@ public class EndlessConsumer implements ConsumerRebalanceListener, Runnable
               record.value()
           );
 
-          consumed++;
-
-          Integer partition = record.partition();
-          String user = record.key();
-          Map<String, Map<String, Long>> users = seen.get(partition);
-
-          Map<String, Long> words = users.get(user);
-          if (words == null)
-          {
-            words = new HashMap<>();
-            users.put(user, words);
-          }
+          handler.accept(record);
 
-          for (String word : PATTERN.split(record.value()))
-          {
-            Long num = words.get(word);
-            if (num == null)
-            {
-              num = 1l;
-            }
-            else
-            {
-              num++;
-            }
-            words.put(word, num);
-          }
+          consumed++;
         }
 
-        if (lastCommit.plus(commitInterval).isBefore(clock.instant()))
-        {
-          log.debug("Storing data and offsets, last commit: {}", lastCommit);
-          seen.forEach((partiton, statistics) -> repository.save(
-              new StatisticsDocument(
-                  partiton,
-                  statistics,
-                  consumer.position(new TopicPartition(topic, partiton)))));
-          lastCommit = clock.instant();
-        }
+        handler.beforeNextPoll();
       }
     }
     catch(WakeupException e)
@@ -218,11 +147,6 @@ public class EndlessConsumer implements ConsumerRebalanceListener, Runnable
     }
   }
 
-  public Map<Integer, Map<String, Map<String, Long>>> getSeen()
-  {
-    return seen;
-  }
-
   public void start()
   {
     lock.lock();