1 package de.juplo.kafka;
3 import lombok.RequiredArgsConstructor;
4 import lombok.extern.slf4j.Slf4j;
5 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
6 import org.apache.kafka.common.TopicPartition;
8 import java.time.Clock;
9 import java.time.Duration;
10 import java.time.Instant;
14 @RequiredArgsConstructor
16 public class ApplicationRebalanceListener implements ConsumerRebalanceListener
18 private final ApplicationRecordHandler recordHandler;
19 private final AdderResults adderResults;
20 private final StateRepository stateRepository;
21 private final String id;
23 private final Set<Integer> partitions = new HashSet<>();
26 public void onPartitionsAssigned(Collection<TopicPartition> partitions)
28 partitions.forEach(tp ->
30 Integer partition = tp.partition();
31 log.info("{} - adding partition: {}", id, partition);
32 this.partitions.add(partition);
33 StateDocument document =
35 .findById(Integer.toString(partition))
36 .orElse(new StateDocument(partition));
37 recordHandler.addPartition(partition, document.state);
38 for (String user : document.state.keySet())
41 "{} - Restored state for partition={}|user={}: {}",
45 document.state.get(user));
47 adderResults.addPartition(partition, document.results);
52 public void onPartitionsRevoked(Collection<TopicPartition> partitions)
54 partitions.forEach(tp ->
56 Integer partition = tp.partition();
57 log.info("{} - removing partition: {}", id, partition);
58 this.partitions.remove(partition);
59 Map<String, AdderResult> state = recordHandler.removePartition(partition);
60 for (String user : state.keySet())
63 "{} - Saved state for partition={}|user={}: {}",
69 Map<String, List<AdderResult>> results = adderResults.removePartition(partition);
70 stateRepository.save(new StateDocument(partition, state, results));