juplo.de Git - demos/kafka/training/blob - src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java

   1 package de.juplo.kafka;
   2
   3 import lombok.RequiredArgsConstructor;
   4 import lombok.extern.slf4j.Slf4j;
   5 import org.apache.kafka.clients.consumer.Consumer;
   6 import org.apache.kafka.common.TopicPartition;
   7
   8 import java.util.*;
   9
  10
  11 @RequiredArgsConstructor
  12 @Slf4j
  13 public class ApplicationRebalanceListener implements RebalanceListener
  14 {
  15   private final ApplicationRecordHandler recordHandler;
  16   private final AdderResults adderResults;
  17   private final StateRepository stateRepository;
  18   private final String id;
  19   private final String topic;
  20   private final Consumer consumer;
  21
  22   private final Set<Integer> partitions = new HashSet<>();
  23
  24   @Override
  25   public void onPartitionsAssigned(Collection<TopicPartition> partitions)
  26   {
  27     partitions.forEach(tp ->
  28     {
  29       Integer partition = tp.partition();
  30       log.info("{} - adding partition: {}", id, partition);
  31       this.partitions.add(partition);
  32       StateDocument document =
  33           stateRepository
  34               .findById(Integer.toString(partition))
  35               .orElse(new StateDocument(partition));
  36       log.info(
  37         "{} - Offset of next unseen message for partition {}: {}",
  38         id,
  39         partition,
  40         document.offset);
  41       consumer.seek(tp, document.offset);
  42       recordHandler.addPartition(partition, document.state);
  43       for (String user : document.state.keySet())
  44       {
  45         log.info(
  46             "{} - Restored state for partition={}|user={}: {}",
  47             id,
  48             partition,
  49             user,
  50             document.state.get(user));
  51       }
  52       adderResults.addPartition(partition, document.results);
  53     });
  54   }
  55
  56   @Override
  57   public void onPartitionsRevoked(Collection<TopicPartition> partitions)
  58   {
  59     partitions.forEach(tp ->
  60     {
  61       Integer partition = tp.partition();
  62       log.info("{} - removing partition: {}", id, partition);
  63       this.partitions.remove(partition);
  64       Map<String, AdderResult> state = recordHandler.removePartition(partition);
  65       Long offset = consumer.position(tp);
  66       log.info(
  67           "{} - offset of next unseen message for partition {} is {}",
  68           id,
  69           partition,
  70           offset);
  71       for (String user : state.keySet())
  72       {
  73         log.info(
  74             "{} - Saved state for partition={}|user={}: {}",
  75             id,
  76             partition,
  77             user,
  78             state.get(user));
  79       }
  80       Map<String, List<AdderResult>> results = adderResults.removePartition(partition);
  81       stateRepository.save(new StateDocument(partition, state, results, offset));
  82     });
  83   }
  84 }