X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FEndlessConsumer.java;h=27c1e446cab68442bdf41ee6bb44baeb5c72e6c5;hb=66ff7d205e66616de8aaca94503dbbcd7d281f6d;hp=b152310b3441316032b59ba8ff7948d3291c8c84;hpb=3c1d3fa68df685146bdef7cc2e396e55fa0933dc;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 b152310..27c1e44 100644 --- a/src/main/java/de/juplo/kafka/EndlessConsumer.java +++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java @@ -1,203 +1,102 @@ package de.juplo.kafka; +import lombok.RequiredArgsConstructor; 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 org.springframework.kafka.annotation.KafkaHandler; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.config.KafkaListenerEndpointRegistry; +import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.messaging.handler.annotation.Header; +import org.springframework.messaging.handler.annotation.Payload; -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.Future; -import java.util.concurrent.atomic.AtomicBoolean; +import java.util.List; +import java.util.Optional; +@RequiredArgsConstructor @Slf4j -public class EndlessConsumer implements Runnable +@KafkaListener( + id = "${spring.kafka.client-id}", + idIsGroup = false, + topics = "${sumup.adder.topic}", + autoStartup = "false") +public class EndlessConsumer { - 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 final KafkaListenerEndpointRegistry registry; + private final RecordHandler recordHandler; - private AtomicBoolean running = new AtomicBoolean(); private long consumed = 0; - private KafkaConsumer consumer = null; - private Future future = null; - private final Map seen = new HashMap<>(); - - public EndlessConsumer( - ExecutorService executor, - PartitionStatisticsRepository repository, - String bootstrapServer, - String groupId, - String clientId, - String topic, - String autoOffsetReset) + @KafkaHandler + public void addNumber( + @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key, + @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, + @Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partition, + @Header(KafkaHeaders.OFFSET) Long offset, + @Payload MessageAddNumber message) { - this.executor = executor; - this.repository = repository; - this.bootstrapServer = bootstrapServer; - this.groupId = groupId; - this.id = clientId; - this.topic = topic; - this.autoOffsetReset = autoOffsetReset; + log.info( + "{} - {}: {}/{} - {}={}", + id, + offset, + topic, + partition, + key, + message + ); + + recordHandler.addNumber(topic, partition, offset, key, message); + + consumed++; } - @Override - public void run() + @KafkaHandler + public void calculateSum( + @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key, + @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, + @Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partition, + @Header(KafkaHeaders.OFFSET) Long offset, + @Payload MessageCalculateSum message) { - try - { - 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); - PartitionStatistics removed = seen.remove(tp); - for (KeyCounter counter : removed.getStatistics()) - { - log.info( - "{} - Seen {} messages for partition={}|key={}", - id, - counter.getResult(), - removed.getPartition(), - counter.getKey()); - } - repository.save(new StatisticsDocument(removed, consumer.position(tp))); - }); - } - - @Override - public void onPartitionsAssigned(Collection partitions) - { - partitions.forEach(tp -> - { - log.info("{} - adding partition: {}", id, tp); - StatisticsDocument document = - repository - .findById(tp.toString()) - .orElse(new StatisticsDocument(tp)); - consumer.seek(tp, document.offset); - seen.put(tp, new PartitionStatistics(document)); - }); - } - }); - - 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) - { - consumed++; log.info( "{} - {}: {}/{} - {}={}", id, - record.offset(), - record.topic(), - record.partition(), - record.key(), - record.value() + offset, + topic, + partition, + key, + message ); - TopicPartition partition = new TopicPartition(record.topic(), record.partition()); - String key = record.key() == null ? "NULL" : record.key(); - seen.get(partition).increment(key); - } - - seen.forEach((tp, statistics) -> repository.save(new StatisticsDocument(statistics, consumer.position(tp)))); - } - } - catch(WakeupException e) - { - log.info("{} - RIIING!", id); - } - catch(Exception e) - { - log.error("{} - Unexpected error: {}", id, e.toString(), e); - running.set(false); // Mark the instance as not running - } - finally - { - log.info("{} - Closing the KafkaConsumer", id); - consumer.close(); - log.info("{} - Consumer-Thread exiting", id); - } - } + recordHandler.calculateSum(topic, partition, offset, key, message); - public Map getSeen() - { - return seen; + consumed++; } - public synchronized void start() + public void start() { - boolean stateChanged = running.compareAndSet(false, true); - if (!stateChanged) - throw new RuntimeException("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); - future = executor.submit(this); + registry.getListenerContainer(id).start(); } - public synchronized void stop() throws ExecutionException, InterruptedException + public void stop() { - boolean stateChanged = running.compareAndSet(true, false); - if (!stateChanged) - throw new RuntimeException("Consumer instance " + id + " is not running!"); + if (!running()) + throw new IllegalStateException("Consumer instance " + id + " is not running!"); log.info("{} - Stopping", id); - consumer.wakeup(); - future.get(); + registry.getListenerContainer(id).stop(); log.info("{} - Stopped - consumed {} messages so far", id, consumed); } - @PreDestroy - public void destroy() throws ExecutionException, InterruptedException + public boolean running() { - 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); - } + return registry.getListenerContainer(id).isRunning(); } }