X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FEndlessConsumer.java;h=25632048aaa0bf637bfacb9acdeb4ea11aab30b6;hb=620191782035383e0083dc348e4941c9cec0d994;hp=e4d969748e896ff18863cd465b6e8ab9ee91c2be;hpb=95b1d2d6fd83c68aa0db3e426f173921600e44a6;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 e4d9697..2563204 100644 --- a/src/main/java/de/juplo/kafka/EndlessConsumer.java +++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java @@ -1,64 +1,122 @@ 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.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 java.util.Arrays; -import java.util.Properties; -import java.util.concurrent.locks.Condition; -import java.util.concurrent.locks.Lock; -import java.util.concurrent.locks.ReentrantLock; +import java.util.*; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; @Slf4j -public class SimpleConsumer +public class EndlessConsumer implements Runnable { + private final ExecutorService executor; + private final PartitionStatisticsRepository repository; + private final String bootstrapServer; + private final String groupId; + private final String id; + private final String topic; + private final String autoOffsetReset; + + private AtomicBoolean running = new AtomicBoolean(); private long consumed = 0; - private KafkaConsumer consumer; - private Lock lock = new ReentrantLock(); - private Condition stopped = lock.newCondition(); + private KafkaConsumer consumer = null; + private Future future = null; + private final Map> seen = new HashMap<>(); - public SimpleConsumer() + + public EndlessConsumer( + ExecutorService executor, + PartitionStatisticsRepository repository, + String bootstrapServer, + String groupId, + String clientId, + String topic, + String autoOffsetReset) { - // tag::create[] - Properties props = new Properties(); - props.put("bootstrap.servers", ":9092"); - props.put("group.id", "my-consumer"); // << Used for Offset-Commits - // end::create[] - props.put("auto.offset.reset", "earliest"); - // tag::create[] - props.put("key.deserializer", StringDeserializer.class.getName()); - props.put("value.deserializer", StringDeserializer.class.getName()); - - KafkaConsumer consumer = new KafkaConsumer<>(props); - // end::create[] - this.consumer = consumer; + this.executor = executor; + this.repository = repository; + this.bootstrapServer = bootstrapServer; + this.groupId = groupId; + this.id = clientId; + this.topic = topic; + this.autoOffsetReset = autoOffsetReset; } - + @Override public void run() { - String id = "C"; - try { - log.info("{} - Subscribing to topic test", id); - consumer.subscribe(Arrays.asList("test")); + Properties props = new Properties(); + props.put("bootstrap.servers", bootstrapServer); + props.put("group.id", groupId); + props.put("client.id", id); + props.put("enable.auto.commit", false); + props.put("auto.offset.reset", autoOffsetReset); + props.put("metadata.max.age.ms", "1000"); + props.put("key.deserializer", StringDeserializer.class.getName()); + props.put("value.deserializer", StringDeserializer.class.getName()); + + this.consumer = new KafkaConsumer<>(props); + + log.info("{} - Subscribing to topic {}", id, topic); + consumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener() + { + @Override + public void onPartitionsRevoked(Collection partitions) + { + partitions.forEach(tp -> + { + log.info("{} - removing partition: {}", id, tp); + Map removed = seen.remove(tp.partition()); + for (String key : removed.keySet()) + { + log.info( + "{} - Seen {} messages for partition={}|key={}", + id, + removed.get(key), + tp.partition(), + key); + } + repository.save(new StatisticsDocument(tp.partition(), removed, consumer.position(tp))); + }); + } + + @Override + public void onPartitionsAssigned(Collection partitions) + { + partitions.forEach(tp -> + { + log.info("{} - adding partition: {}", id, tp); + StatisticsDocument document = + repository + .findById(Integer.toString(tp.partition())) + .orElse(new StatisticsDocument(tp.partition())); + consumer.seek(tp, document.offset); + seen.put(tp.partition(), document.statistics); + }); + } + }); - // tag::loop[] while (true) { ConsumerRecords records = consumer.poll(Duration.ofSeconds(1)); // Do something with the data... - // end::loop[] log.info("{} - Received {} messages", id, records.count()); for (ConsumerRecord record : records) { @@ -72,10 +130,25 @@ public class SimpleConsumer record.key(), record.value() ); + + Integer partition = record.partition(); + String key = record.key() == null ? "NULL" : record.key(); + Map byKey = seen.get(partition); + + if (!byKey.containsKey(key)) + byKey.put(key, 0); + + int seenByKey = byKey.get(key); + seenByKey++; + byKey.put(key, seenByKey); } - // tag::loop[] + + seen.forEach((partiton, statistics) -> repository.save( + new StatisticsDocument( + partiton, + statistics, + consumer.position(new TopicPartition(topic, partiton))))); } - // end::loop[] } catch(WakeupException e) { @@ -83,49 +156,59 @@ public class SimpleConsumer } catch(Exception e) { - log.error("{} - Unexpected error: {}", id, e.toString()); + log.error("{} - Unexpected error: {}", id, e.toString(), e); + running.set(false); // Mark the instance as not running } finally { - this.lock.lock(); - try - { - log.info("{} - Closing the KafkaConsumer", id); - consumer.close(); - log.info("C - DONE!"); - stopped.signal(); - } - finally - { - this.lock.unlock(); - log.info("{}: Consumed {} messages in total, exiting!", id, consumed); - } + log.info("{} - Closing the KafkaConsumer", id); + consumer.close(); + log.info("{} - Consumer-Thread exiting", id); } } + public Map> getSeen() + { + return seen; + } - public static void main(String[] args) throws Exception + public synchronized void start() { - SimpleConsumer instance = new SimpleConsumer(); + boolean stateChanged = running.compareAndSet(false, true); + if (!stateChanged) + throw new RuntimeException("Consumer instance " + id + " is already running!"); - Runtime.getRuntime().addShutdownHook(new Thread(() -> - { - instance.lock.lock(); - try - { - instance.consumer.wakeup(); - instance.stopped.await(); - } - catch (InterruptedException e) - { - log.warn("Interrrupted while waiting for the consumer to stop!", e); - } - finally - { - instance.lock.unlock(); - } - })); + log.info("{} - Starting - consumed {} messages before", id, consumed); + future = executor.submit(this); + } + + public synchronized void stop() throws ExecutionException, InterruptedException + { + boolean stateChanged = running.compareAndSet(true, false); + if (!stateChanged) + throw new RuntimeException("Consumer instance " + id + " is not running!"); + + log.info("{} - Stopping", id); + consumer.wakeup(); + future.get(); + log.info("{} - Stopped - consumed {} messages so far", id, consumed); + } - instance.run(); + @PreDestroy + public void destroy() throws ExecutionException, InterruptedException + { + log.info("{} - Destroy!", id); + try + { + stop(); + } + catch (IllegalStateException e) + { + log.info("{} - Was already stopped", id); + } + finally + { + log.info("{}: Consumed {} messages in total, exiting!", id, consumed); + } } }