1 package de.juplo.kafka;
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;
11 @RequiredArgsConstructor
13 public class ApplicationRebalanceListener implements RebalanceListener
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;
22 private final Set<Integer> partitions = new HashSet<>();
25 public void onPartitionsAssigned(Collection<TopicPartition> partitions)
27 partitions.forEach(tp ->
29 Integer partition = tp.partition();
30 log.info("{} - adding partition: {}", id, partition);
31 this.partitions.add(partition);
32 StateDocument document =
34 .findById(Integer.toString(partition))
35 .orElse(new StateDocument(partition));
36 if (document.offset >= 0)
38 // Only seek, if a stored offset was found
39 // Otherwise: Use initial offset, generated by Kafka
40 consumer.seek(tp, document.offset);
42 "{} - Seeking to offset {} for partition {}",
47 recordHandler.addPartition(partition, document.state);
48 for (String user : document.state.keySet())
51 "{} - Restored state for partition={}|user={}: {}",
55 document.state.get(user));
57 adderResults.addPartition(partition, document.results);
62 public void onPartitionsRevoked(Collection<TopicPartition> partitions)
64 partitions.forEach(tp ->
66 Integer partition = tp.partition();
67 log.info("{} - removing partition: {}", id, partition);
68 this.partitions.remove(partition);
69 Map<String, AdderResult> state = recordHandler.removePartition(partition);
70 Long offset = consumer.position(tp);
72 "{} - offset of next unseen message for partition {} is {}",
76 for (String user : state.keySet())
79 "{} - Saved state for partition={}|user={}: {}",
85 Map<String, List<AdderResult>> results = adderResults.removePartition(partition);
86 stateRepository.save(new StateDocument(partition, state, results, offset));