X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FEndlessConsumer.java;h=9ea944b9039812dd3c0a6d4f1dcd10286d1f67bf;hb=eba11d4859d1e2b936bcd9e8075986b5179b32ea;hp=01f9057a055aa360bb8d98cca1641073df19afe0;hpb=a6a0a22a5fa34a01b0e8b2bc1e0e2b82d7b60f33;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 01f9057..9ea944b 100644 --- a/src/main/java/de/juplo/kafka/EndlessConsumer.java +++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java @@ -6,82 +6,29 @@ import org.apache.kafka.clients.consumer.*; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.RecordDeserializationException; import org.apache.kafka.common.errors.WakeupException; +import org.springframework.context.ConfigurableApplicationContext; -import javax.annotation.PreDestroy; -import java.time.Clock; import java.time.Duration; -import java.time.Instant; import java.util.*; -import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; -import java.util.concurrent.locks.Condition; -import java.util.concurrent.locks.Lock; -import java.util.concurrent.locks.ReentrantLock; -import java.util.regex.Pattern; @Slf4j @RequiredArgsConstructor -public class EndlessConsumer implements ConsumerRebalanceListener, Runnable +public class EndlessConsumer implements Runnable { - final static Pattern PATTERN = Pattern.compile("\\W+"); - - private final ExecutorService executor; - private final PartitionStatisticsRepository repository; + private final ConfigurableApplicationContext applicationContext; private final String id; private final String topic; - private final Clock clock; - private final Duration commitInterval; - private final Consumer consumer; + private final Consumer consumer; + private final ConsumerRebalanceListener rebalanceListener; + private final RecordHandler recordHandler; - private final Lock lock = new ReentrantLock(); - private final Condition condition = lock.newCondition(); private boolean running = false; - private Exception exception; + private Exception exception = null; private long consumed = 0; - private final Map>> seen = new HashMap<>(); - - - @Override - public void onPartitionsRevoked(Collection partitions) - { - partitions.forEach(tp -> - { - Integer partition = tp.partition(); - Long newOffset = consumer.position(tp); - log.info( - "{} - removing partition: {}, offset of next message {})", - id, - partition, - newOffset); - Map> removed = seen.remove(partition); - repository.save(new StatisticsDocument(partition, removed, consumer.position(tp))); - }); - } - - @Override - public void onPartitionsAssigned(Collection partitions) - { - partitions.forEach(tp -> - { - Integer partition = tp.partition(); - Long offset = consumer.position(tp); - log.info("{} - adding partition: {}, offset={}", id, partition, offset); - StatisticsDocument document = - repository - .findById(Integer.toString(partition)) - .orElse(new StatisticsDocument(partition)); - if (document.offset >= 0) - { - // Only seek, if a stored offset was found - // Otherwise: Use initial offset, generated by Kafka - consumer.seek(tp, document.offset); - } - seen.put(partition, document.statistics); - }); - } @Override @@ -90,18 +37,16 @@ public class EndlessConsumer implements ConsumerRebalanceListener, Runnable try { log.info("{} - Subscribing to topic {}", id, topic); - consumer.subscribe(Arrays.asList(topic), this); - - Instant lastCommit = clock.instant(); + consumer.subscribe(Arrays.asList(topic), rebalanceListener); while (true) { - ConsumerRecords records = + ConsumerRecords records = consumer.poll(Duration.ofSeconds(1)); // Do something with the data... log.info("{} - Received {} messages", id, records.count()); - for (ConsumerRecord record : records) + for (ConsumerRecord record : records) { log.info( "{} - {}: {}/{} - {}={}", @@ -113,50 +58,15 @@ public class EndlessConsumer implements ConsumerRebalanceListener, Runnable record.value() ); - consumed++; - - Integer partition = record.partition(); - String user = record.key(); - Map> users = seen.get(partition); + recordHandler.accept(record); - Map words = users.get(user); - if (words == null) - { - words = new HashMap<>(); - users.put(user, words); - } - - for (String word : PATTERN.split(record.value())) - { - Long num = words.get(word); - if (num == null) - { - num = 1l; - } - else - { - num++; - } - words.put(word, num); - } - } - - if (lastCommit.plus(commitInterval).isBefore(clock.instant())) - { - log.debug("Storing data and offsets, last commit: {}", lastCommit); - seen.forEach((partiton, statistics) -> repository.save( - new StatisticsDocument( - partiton, - statistics, - consumer.position(new TopicPartition(topic, partiton))))); - lastCommit = clock.instant(); + consumed++; } } } catch(WakeupException e) { log.info("{} - RIIING! Request to stop consumption - commiting current offsets!", id); - shutdown(); } catch(RecordDeserializationException e) { @@ -168,132 +78,50 @@ public class EndlessConsumer implements ConsumerRebalanceListener, Runnable tp, offset, e.getCause().toString()); - - shutdown(e); + this.exception = e; } catch(Exception e) { log.error("{} - Unexpected error: {}", id, e.toString(), e); - shutdown(e); + this.exception = e; + log.info("{} - Unsubscribing...", id); + consumer.unsubscribe(); } finally { + running = false; + log.info("{} - Closing the consumer...", id); + consumer.close(); + log.info("{} - Shutting down the app...", id); + applicationContext.close(); log.info("{} - Consumer-Thread exiting", id); } } - private void shutdown() - { - shutdown(null); - } - - private void shutdown(Exception e) - { - lock.lock(); - try - { - try - { - log.info("{} - Unsubscribing from topic {}", id, topic); - consumer.unsubscribe(); - } - catch (Exception ue) - { - log.error( - "{} - Error while unsubscribing from topic {}: {}", - id, - topic, - ue.toString()); - } - finally - { - running = false; - exception = e; - condition.signal(); - } - } - finally - { - lock.unlock(); - } - } - - public Map>> getSeen() - { - return seen; - } - public void start() { - lock.lock(); - try - { - if (running) - throw new IllegalStateException("Consumer instance " + id + " is already running!"); + if (running) + throw new IllegalStateException("Consumer instance " + id + " is already running!"); - log.info("{} - Starting - consumed {} messages before", id, consumed); - running = true; - exception = null; - executor.submit(this); - } - finally - { - lock.unlock(); - } + log.info("{} - Starting - consumed {} messages before", id, consumed); + running = true; + executor.submit(this); } - public synchronized void stop() throws InterruptedException + public void stop() { - lock.lock(); - try - { - if (!running) - throw new IllegalStateException("Consumer instance " + id + " is not running!"); - - log.info("{} - Stopping", id); - consumer.wakeup(); - condition.await(); - log.info("{} - Stopped - consumed {} messages so far", id, consumed); - } - finally - { - lock.unlock(); - } - } - - @PreDestroy - public void destroy() throws ExecutionException, InterruptedException - { - log.info("{} - Destroy!", id); - log.info("{}: Consumed {} messages in total, exiting!", id, consumed); + consumer.wakeup(); } - public boolean running() { - lock.lock(); - try - { - return running; - } - finally - { - lock.unlock(); - } + return running; } public Optional exitStatus() { - lock.lock(); - try - { - if (running) - throw new IllegalStateException("No exit-status available: Consumer instance " + id + " is running!"); + if (running) + throw new IllegalStateException("No exit-status available: Consumer instance " + id + " is running!"); - return Optional.ofNullable(exception); - } - finally - { - lock.unlock(); - } + return Optional.ofNullable(exception); } }