From: Kai Moritz Date: Sun, 10 Apr 2022 20:15:34 +0000 (+0200) Subject: Merge branch 'endless-stream-consumer' into rebalance-listener X-Git-Tag: deserialization-synchroner-test~16 X-Git-Url: https://juplo.de/gitweb/?a=commitdiff_plain;h=92882a04df8b54ae6ac000c14a9793f7411318f9;hp=-c;p=demos%2Fkafka%2Ftraining Merge branch 'endless-stream-consumer' into rebalance-listener --- 92882a04df8b54ae6ac000c14a9793f7411318f9 diff --combined docker-compose.yml index 0b1f0ae,b03bb1e..1392ae2 --- a/docker-compose.yml +++ b/docker-compose.yml @@@ -24,13 -24,13 +24,13 @@@ services depends_on: - zookeeper - setup: - image: juplo/toolbox - command: > - bash -c " - kafka-topics --bootstrap-server kafka:9092 --delete --if-exists --topic test - kafka-topics --bootstrap-server kafka:9092 --create --topic test --partitions 2 - " + kafka-ui: + image: provectuslabs/kafka-ui:0.3.3 + ports: + - 8080:8080 + environment: + KAFKA_CLUSTERS_0_NAME: local + KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092 cli: image: juplo/toolbox @@@ -39,18 -39,18 +39,18 @@@ producer: image: juplo/endless-producer:1.0-SNAPSHOT ports: - - 8000:8080 + - 8080:8880 environment: producer.bootstrap-server: kafka:9092 producer.client-id: producer producer.topic: test - producer.throttle-ms: 200 + producer.throttle-ms: 10 consumer: image: juplo/endless-consumer:1.0-SNAPSHOT ports: - - 8081:8081 + - 8081:8881 environment: consumer.bootstrap-server: kafka:9092 consumer.client-id: my-group diff --combined pom.xml index d9b70f9,cf68030..8c5dccc --- a/pom.xml +++ b/pom.xml @@@ -14,7 -14,7 +14,7 @@@ de.juplo.kafka endless-consumer 1.0-SNAPSHOT - Endless Consumer: a Simple Consumer-Group that reads and print the topic + Endless Consumer: a Simple Consumer-Group that reads and prints the topic and counts the received messages for each key by topic @@@ -54,6 -54,17 +54,17 @@@ org.springframework.boot spring-boot-maven-plugin + + + + build-info + + + + + + pl.project13.maven + git-commit-id-plugin io.fabric8 diff --combined src/main/java/de/juplo/kafka/ApplicationHealthIndicator.java index 0000000,0000000..ab9782c new file mode 100644 --- /dev/null +++ b/src/main/java/de/juplo/kafka/ApplicationHealthIndicator.java @@@ -1,0 -1,0 +1,32 @@@ ++package de.juplo.kafka; ++ ++import lombok.RequiredArgsConstructor; ++import org.springframework.boot.actuate.health.Health; ++import org.springframework.boot.actuate.health.HealthIndicator; ++import org.springframework.stereotype.Component; ++ ++ ++@Component ++@RequiredArgsConstructor ++public class ApplicationHealthIndicator implements HealthIndicator ++{ ++ private final EndlessConsumer consumer; ++ ++ ++ @Override ++ public Health health() ++ { ++ try ++ { ++ return consumer ++ .exitStatus() ++ .map(Health::down) ++ .orElse(Health.outOfService()) ++ .build(); ++ } ++ catch (IllegalStateException e) ++ { ++ return Health.up().build(); ++ } ++ } ++} diff --combined src/main/java/de/juplo/kafka/DriverController.java index a504842,a02fd2c..1525f5a --- a/src/main/java/de/juplo/kafka/DriverController.java +++ b/src/main/java/de/juplo/kafka/DriverController.java @@@ -1,11 -1,12 +1,14 @@@ package de.juplo.kafka; import lombok.RequiredArgsConstructor; + import org.springframework.http.HttpStatus; + import org.springframework.web.bind.annotation.ExceptionHandler; +import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PostMapping; + import org.springframework.web.bind.annotation.ResponseStatus; import org.springframework.web.bind.annotation.RestController; +import java.util.Map; import java.util.concurrent.ExecutionException; @@@ -28,10 -29,10 +31,16 @@@ public class DriverControlle consumer.stop(); } - + @GetMapping("seen") + public Map> seen() + { + return consumer.getSeen(); + } ++ + @ExceptionHandler + @ResponseStatus(HttpStatus.BAD_REQUEST) + public ErrorResponse illegalStateException(IllegalStateException e) + { + return new ErrorResponse(e.getMessage(), HttpStatus.BAD_REQUEST.value()); + } } diff --combined src/main/java/de/juplo/kafka/EndlessConsumer.java index 03a89dc,adebff1..c2d4447 --- a/src/main/java/de/juplo/kafka/EndlessConsumer.java +++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java @@@ -1,17 -1,17 +1,17 @@@ 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.Optional; -import java.util.Properties; +import java.util.*; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.locks.Condition; @@@ -32,13 -32,11 +32,14 @@@ public class EndlessConsumer implement private final Lock lock = new ReentrantLock(); private final Condition condition = lock.newCondition(); private boolean running = false; + private Exception exception; private long consumed = 0; private KafkaConsumer consumer = null; + private final Map> seen = new HashMap<>(); + + public EndlessConsumer( ExecutorService executor, String bootstrapServer, @@@ -65,44 -63,13 +66,44 @@@ props.put("group.id", groupId); props.put("client.id", id); 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)); + 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); + } + }); + } + + @Override + public void onPartitionsAssigned(Collection partitions) + { + partitions.forEach(tp -> + { + log.info("{} - adding partition: {}", id, tp); + seen.put(tp.partition(), new HashMap<>()); + }); + } + }); while (true) { @@@ -123,17 -90,6 +124,17 @@@ 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); } } } @@@ -145,7 -101,7 +146,7 @@@ catch(Exception e) { log.error("{} - Unexpected error: {}", id, e.toString(), e); - shutdown(); + shutdown(e); } finally { @@@ -156,11 -112,17 +157,17 @@@ } private void shutdown() + { + shutdown(null); + } + + private void shutdown(Exception e) { lock.lock(); try { running = false; + exception = e; condition.signal(); } finally @@@ -169,11 -131,6 +176,11 @@@ } } + public Map> getSeen() + { + return seen; + } + public void start() { lock.lock(); @@@ -184,6 -141,7 +191,7 @@@ log.info("{} - Starting - consumed {} messages before", id, consumed); running = true; + exception = null; executor.submit(this); } finally @@@ -232,4 -190,33 +240,33 @@@ 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(); + } + } }