projects
/
demos
/
kafka
/
training
/ commitdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
| commitdiff |
tree
raw
|
patch
|
inline
| side by side (parent:
7bde748
)
Rebalance-Listener anstatt Wegwerfen der Map
author
Kai Moritz
<kai@juplo.de>
Tue, 5 Apr 2022 20:25:47 +0000
(22:25 +0200)
committer
Kai Moritz
<kai@juplo.de>
Wed, 6 Apr 2022 17:03:34 +0000
(19:03 +0200)
src/main/java/de/juplo/kafka/EndlessConsumer.java
patch
|
blob
|
history
diff --git
a/src/main/java/de/juplo/kafka/EndlessConsumer.java
b/src/main/java/de/juplo/kafka/EndlessConsumer.java
index
357a0b4
..
9aa8152
100644
(file)
--- a/
src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/
src/main/java/de/juplo/kafka/EndlessConsumer.java
@@
-1,18
+1,17
@@
package de.juplo.kafka;
import lombok.extern.slf4j.Slf4j;
package de.juplo.kafka;
import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.StringDeserializer;
import javax.annotation.PreDestroy;
import java.time.Duration;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.StringDeserializer;
import javax.annotation.PreDestroy;
import java.time.Duration;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Properties;
+import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
@@
-34,7
+33,7
@@
public class EndlessConsumer implements Runnable
private KafkaConsumer<String, String> consumer = null;
private Future<?> future = null;
private KafkaConsumer<String, String> consumer = null;
private Future<?> future = null;
- private
Map<Integer, Map<String, Integer>> seen
;
+ private
final Map<Integer, Map<String, Integer>> seen = new HashMap<>()
;
public EndlessConsumer(
public EndlessConsumer(
@@
-70,9
+69,20
@@
public class EndlessConsumer implements Runnable
this.consumer = new KafkaConsumer<>(props);
log.info("{} - Subscribing to topic {}", id, topic);
this.consumer = new KafkaConsumer<>(props);
log.info("{} - Subscribing to topic {}", id, topic);
- consumer.subscribe(Arrays.asList(topic));
+ consumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener()
+ {
+ @Override
+ public void onPartitionsRevoked(Collection<TopicPartition> partitions)
+ {
+ partitions.forEach(tp -> seen.remove(tp.partition()));
+ }
- seen = new HashMap<>();
+ @Override
+ public void onPartitionsAssigned(Collection<TopicPartition> partitions)
+ {
+ partitions.forEach(tp -> seen.put(tp.partition(), new HashMap<>()));
+ }
+ });
while (true)
{
while (true)
{
@@
-96,10
+106,6
@@
public class EndlessConsumer implements Runnable
Integer partition = record.partition();
String key = record.key() == null ? "NULL" : record.key();
Integer partition = record.partition();
String key = record.key() == null ? "NULL" : record.key();
-
- if (!seen.containsKey(partition))
- seen.put(partition, new HashMap<>());
-
Map<String, Integer> byKey = seen.get(partition);
if (!byKey.containsKey(key))
Map<String, Integer> byKey = seen.get(partition);
if (!byKey.containsKey(key))
@@
-138,7
+144,6
@@
public class EndlessConsumer implements Runnable
key);
}
}
key);
}
}
- seen = null;
log.info("{} - Consumer-Thread exiting", id);
}
log.info("{} - Consumer-Thread exiting", id);
}