From d58ed9de440e56ecf906b26803591290b1c4f60d Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Wed, 2 Nov 2022 18:42:34 +0100 Subject: [PATCH] WIP --- src/main/java/de/juplo/kafka/Application.java | 17 +- .../java/de/juplo/kafka/EndlessConsumer.java | 214 ------------------ .../java/de/juplo/kafka/SimpleConsumer.java | 5 - 3 files changed, 13 insertions(+), 223 deletions(-) delete mode 100644 src/main/java/de/juplo/kafka/EndlessConsumer.java diff --git a/src/main/java/de/juplo/kafka/Application.java b/src/main/java/de/juplo/kafka/Application.java index 94224e1..ab357c7 100644 --- a/src/main/java/de/juplo/kafka/Application.java +++ b/src/main/java/de/juplo/kafka/Application.java @@ -11,30 +11,39 @@ import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.ConsumerFactory; import javax.annotation.PreDestroy; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; @SpringBootApplication @Slf4j public class Application implements ApplicationRunner { + @Autowired + ExecutorService executorService; @Autowired Consumer consumer; @Autowired SimpleConsumer simpleConsumer; + Future consumerJob; @Override public void run(ApplicationArguments args) throws Exception { - log.info("Starting EndlessConsumer"); - simpleConsumer.start(); + log.info("Starting SimpleConsumer"); + consumerJob = executorService.submit(simpleConsumer); } @PreDestroy - public void shutdown() + public void shutdown() throws ExecutionException, InterruptedException { - log.info("Signaling the consumer to quit its work"); + log.info("Signaling SimpleConsumer to quit its work"); consumer.wakeup(); + log.info("Waiting for SimpleConsumer to finish its work"); + consumerJob.get(); + log.info("SimpleConsumer finished its work"); } @Bean(destroyMethod = "close") diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java deleted file mode 100644 index ba8eb27..0000000 --- a/src/main/java/de/juplo/kafka/EndlessConsumer.java +++ /dev/null @@ -1,214 +0,0 @@ -package de.juplo.kafka; - -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -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.stereotype.Component; - -import javax.annotation.PreDestroy; -import java.time.Duration; -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; - - -@Component -@Slf4j -@RequiredArgsConstructor -public class EndlessConsumer implements Runnable -{ - private final ExecutorService executor; - private final String id; - private final String topic; - 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 long consumed = 0; - - - - @Override - public void run() - { - try - { - log.info("{} - Subscribing to topic {}", id, topic); - consumer.subscribe(Arrays.asList(topic), rebalanceListener); - - while (true) - { - ConsumerRecords records = - consumer.poll(Duration.ofSeconds(1)); - - // Do something with the data... - log.info("{} - Received {} messages", id, records.count()); - for (ConsumerRecord record : records) - { - log.info( - "{} - {}: {}/{} - {}={}", - id, - record.offset(), - record.topic(), - record.partition(), - record.key(), - record.value() - ); - - recordHandler.accept(record); - - consumed++; - } - } - } - catch(WakeupException e) - { - log.info("{} - RIIING! Request to stop consumption - commiting current offsets!", id); - consumer.commitSync(); - shutdown(); - } - catch(RecordDeserializationException e) - { - TopicPartition tp = e.topicPartition(); - long offset = e.offset(); - log.error( - "{} - Could not deserialize message on topic {} with offset={}: {}", - id, - tp, - offset, - e.getCause().toString()); - - consumer.commitSync(); - shutdown(e); - } - catch(Exception e) - { - log.error("{} - Unexpected error: {}", id, e.toString(), e); - shutdown(e); - } - finally - { - 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 void start() - { - lock.lock(); - try - { - 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(); - } - } - - public synchronized void stop() throws InterruptedException - { - 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); - } - - public boolean running() - { - lock.lock(); - try - { - return running; - } - finally - { - lock.unlock(); - } - } - - public Optional exitStatus() - { - lock.lock(); - try - { - if (running) - throw new IllegalStateException("No exit-status available: Consumer instance " + id + " is running!"); - - return Optional.ofNullable(exception); - } - finally - { - lock.unlock(); - } - } -} diff --git a/src/main/java/de/juplo/kafka/SimpleConsumer.java b/src/main/java/de/juplo/kafka/SimpleConsumer.java index 0d371f4..040e24b 100644 --- a/src/main/java/de/juplo/kafka/SimpleConsumer.java +++ b/src/main/java/de/juplo/kafka/SimpleConsumer.java @@ -72,9 +72,4 @@ public class SimpleConsumer implements Runnable log.info("{}: Consumed {} messages in total, exiting!", id, consumed); } } - - public void start() - { - executor.submit(this); - } } -- 2.20.1