From: Kai Moritz Date: Sat, 23 Jul 2022 11:49:23 +0000 (+0200) Subject: Merge der überarbeiteten Compose-Konfiguration ('endless-stream-consumer') X-Git-Tag: sumup-requests---lvm-2-tage~5^2~1 X-Git-Url: https://juplo.de/gitweb/?a=commitdiff_plain;h=f9c0ba7779552d8fcfc9cb29c8b689e20c314904;hp=-c;p=demos%2Fkafka%2Ftraining Merge der überarbeiteten Compose-Konfiguration ('endless-stream-consumer') * Die letzten Änderungen an 'endless-stream-consumer' sind länger nicht mehr gemerged worden. --- f9c0ba7779552d8fcfc9cb29c8b689e20c314904 diff --combined docker-compose.yml index 0b1f0ae,159f9cb..f9eeedd --- a/docker-compose.yml +++ b/docker-compose.yml @@@ -1,14 -1,14 +1,14 @@@ version: '3.2' services: zookeeper: - image: confluentinc/cp-zookeeper:7.0.2 + image: confluentinc/cp-zookeeper:7.1.3 environment: ZOOKEEPER_CLIENT_PORT: 2181 ports: - 2181:2181 kafka: - image: confluentinc/cp-kafka:7.0.2 + image: confluentinc/cp-kafka:7.1.3 environment: KAFKA_BROKER_ID: 1 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 @@@ -24,13 -24,13 +24,13 @@@ 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,19 -39,21 +39,21 @@@ producer: image: juplo/endless-producer:1.0-SNAPSHOT ports: - - 8080:8080 + - 8000:8080 environment: + server.port: 8080 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:8080 environment: + server.port: 8080 consumer.bootstrap-server: kafka:9092 consumer.client-id: my-group consumer.client-id: consumer diff --combined pom.xml index b7b0b8d,29c1851..9db9d9d --- a/pom.xml +++ b/pom.xml @@@ -7,20 -7,24 +7,24 @@@ org.springframework.boot spring-boot-starter-parent - 2.6.5 + 2.7.2 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 org.springframework.boot spring-boot-starter-web + + org.springframework.boot + spring-boot-starter-validation + org.springframework.boot spring-boot-starter-actuator @@@ -50,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/DriverController.java index a504842,a02fd2c..06e562c --- 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.web.bind.annotation.GetMapping; + import org.springframework.http.HttpStatus; + import org.springframework.web.bind.annotation.ExceptionHandler; 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,18 @@@ 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 357a0b4,adebff1..2310ccd --- a/src/main/java/de/juplo/kafka/EndlessConsumer.java +++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java @@@ -10,13 -10,13 +10,15 @@@ import org.apache.kafka.common.serializ import javax.annotation.PreDestroy; import java.time.Duration; import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; + import java.util.Optional; import java.util.Properties; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; - import java.util.concurrent.Future; - import java.util.concurrent.atomic.AtomicBoolean; + import java.util.concurrent.locks.Condition; + import java.util.concurrent.locks.Lock; + import java.util.concurrent.locks.ReentrantLock; @Slf4j @@@ -29,14 -29,14 +31,17 @@@ public class EndlessConsumer implement private final String topic; private final String autoOffsetReset; - private AtomicBoolean running = new AtomicBoolean(); + 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 Future future = null; + + private Map> seen; + + public EndlessConsumer( ExecutorService executor, String bootstrapServer, @@@ -63,7 -63,6 +68,7 @@@ 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()); @@@ -72,8 -71,6 +77,8 @@@ log.info("{} - Subscribing to topic {}", id, topic); consumer.subscribe(Arrays.asList(topic)); + seen = new HashMap<>(); + while (true) { ConsumerRecords records = @@@ -93,82 -90,83 +98,119 @@@ record.key(), record.value() ); + + Integer partition = record.partition(); + String key = record.key() == null ? "NULL" : record.key(); + + if (!seen.containsKey(partition)) + seen.put(partition, new HashMap<>()); + + Map byKey = seen.get(partition); + + if (!byKey.containsKey(key)) + byKey.put(key, 0); + + int seenByKey = byKey.get(key); + seenByKey++; + byKey.put(key, seenByKey); } } } catch(WakeupException e) { log.info("{} - RIIING!", id); + shutdown(); } catch(Exception e) { log.error("{} - Unexpected error: {}", id, e.toString(), e); - running.set(false); // Mark the instance as not running + shutdown(e); } finally { log.info("{} - Closing the KafkaConsumer", id); consumer.close(); + + for (Integer partition : seen.keySet()) + { + Map byKey = seen.get(partition); + for (String key : byKey.keySet()) + { + log.info( + "{} - Seen {} messages for partition={}|key={}", + id, + byKey.get(key), + partition, + key); + } + } + seen = null; + log.info("{} - Consumer-Thread exiting", id); } } + public Map> getSeen() + { + return seen; + } + - public synchronized void start() + private void shutdown() + { + shutdown(null); + } + + private void shutdown(Exception e) + { + lock.lock(); + try + { + running = false; + exception = e; + condition.signal(); + } + finally + { + lock.unlock(); + } + } + + public void start() { - boolean stateChanged = running.compareAndSet(false, true); - if (!stateChanged) - throw new RuntimeException("Consumer instance " + id + " is already running!"); + lock.lock(); + try + { + if (running) + throw new IllegalStateException("Consumer instance " + id + " is already running!"); - log.info("{} - Starting - consumed {} messages before", id, consumed); - future = executor.submit(this); + log.info("{} - Starting - consumed {} messages before", id, consumed); + running = true; + exception = null; + executor.submit(this); + } + finally + { + lock.unlock(); + } } 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); + 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 @@@ -183,9 -181,42 +225,42 @@@ { log.info("{} - Was already stopped", id); } + catch (Exception e) + { + log.error("{} - Unexpected exception while trying to stop the consumer", id, e); + } finally { 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(); + } + } }