From: Kai Moritz Date: Tue, 29 Oct 2024 13:05:40 +0000 (+0100) Subject: Setup, um `grundlagen/simple-consumer` als Consumer Group zu skalieren X-Git-Tag: grundlagen/simple-consumer--consumergroup--vorlage--2025-04-signal-spickzettel X-Git-Url: https://juplo.de/gitweb/?a=commitdiff_plain;h=refs%2Fheads%2Fgrundlagen%2Fsimple-consumer--consumergroup--vorlage;p=demos%2Fkafka%2Ftraining Setup, um `grundlagen/simple-consumer` als Consumer Group zu skalieren --- diff --git a/.dockerignore b/.dockerignore deleted file mode 100644 index 9127d15f..00000000 --- a/.dockerignore +++ /dev/null @@ -1,4 +0,0 @@ -* -!Dockerfile -!target/*.jar -!target/libs/*.jar diff --git a/.maven-dockerexclude b/.maven-dockerexclude deleted file mode 100644 index 72e8ffc0..00000000 --- a/.maven-dockerexclude +++ /dev/null @@ -1 +0,0 @@ -* diff --git a/.maven-dockerinclude b/.maven-dockerinclude deleted file mode 100644 index a00c65ff..00000000 --- a/.maven-dockerinclude +++ /dev/null @@ -1,2 +0,0 @@ -target/*.jar -target/libs/*.jar diff --git a/Dockerfile b/Dockerfile deleted file mode 100644 index 22819afe..00000000 --- a/Dockerfile +++ /dev/null @@ -1,6 +0,0 @@ -FROM eclipse-temurin:21-jre -VOLUME /tmp -COPY target/*.jar /opt/app.jar -COPY target/libs /opt/libs -ENTRYPOINT [ "java", "-jar", "/opt/app.jar" ] -CMD [ "kafka:9092", "test", "my-group", "DCKR" ] diff --git a/README.sh b/README.sh deleted file mode 100755 index 85b8f960..00000000 --- a/README.sh +++ /dev/null @@ -1,39 +0,0 @@ -#!/bin/bash - -IMAGE=juplo/simple-consumer:1.0-SNAPSHOT - -if [ "$1" = "cleanup" ] -then - docker compose -f docker/docker-compose.yml down -t0 -v --remove-orphans - mvn clean - exit -fi - -docker compose -f docker/docker-compose.yml up -d --remove-orphans kafka-1 kafka-2 kafka-3 -docker compose -f docker/docker-compose.yml rm -svf consumer - -if [[ - $(docker image ls -q $IMAGE) == "" || - "$1" = "build" -]] -then - mvn clean install || exit -else - echo "Using image existing images:" - docker image ls $IMAGE -fi - -docker compose -f docker/docker-compose.yml up --remove-orphans setup || exit 1 - - -docker compose -f docker/docker-compose.yml up -d producer -docker compose -f docker/docker-compose.yml up -d consumer - -sleep 5 -docker compose -f docker/docker-compose.yml stop consumer - -docker compose -f docker/docker-compose.yml start consumer -sleep 5 - -docker compose -f docker/docker-compose.yml stop producer consumer -docker compose -f docker/docker-compose.yml logs consumer diff --git a/build.gradle b/build.gradle deleted file mode 100644 index be0bc470..00000000 --- a/build.gradle +++ /dev/null @@ -1,67 +0,0 @@ -import com.bmuschko.gradle.docker.tasks.image.DockerBuildImage - -plugins { - id 'java' - id 'org.springframework.boot' version '3.4.1' - id 'io.spring.dependency-management' version '1.1.7' - id 'com.bmuschko.docker-remote-api' version '9.3.3' -} - -group = 'de.juplo.kafka' -version = '1.0-SNAPSHOT' - -java { - toolchain { - languageVersion = JavaLanguageVersion.of(21) - } -} - -configurations { - compileOnly { - extendsFrom annotationProcessor - } -} - -repositories { - mavenCentral() -} - -dependencies { - implementation 'org.apache.kafka:kafka-clients' - implementation 'ch.qos.logback:logback-classic' - compileOnly 'org.projectlombok:lombok' - annotationProcessor 'org.projectlombok:lombok' -} - -docker { - // Optional: Konfiguriere den Docker-Host, falls nötig - // url = 'unix:///var/run/docker.sock' (Standard) -} - - -def targetDir = file("${projectDir}/target") -def appJarName = "${project.name}-${project.version}.jar" - -// Task zum Bereinigen des `target`-Verzeichnisses bei `clean` -clean { - delete targetDir -} - -// Task zum Kopieren des Haupt-JARs -task copyJar(type: Copy) { - from "$buildDir/libs/${appJarName}" - into targetDir - dependsOn build - - doFirst { - def libs = file("${targetDir}/libs") - mkdir(libs) - } -} - -// Docker-Task -task buildDockerImage(type: DockerBuildImage) { - inputDir = file('.') - images = ["juplo/${project.name}:${project.version}"] - dependsOn copyJar -} diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 95539006..18c80476 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -143,6 +143,14 @@ services: image: juplo/simple-consumer:1.0-SNAPSHOT command: kafka:9092 test my-group consumer + peter: + image: juplo/simple-consumer:1.0-SNAPSHOT + command: kafka:9092 test my-group peter + + ute: + image: juplo/simple-consumer:1.0-SNAPSHOT + command: kafka:9092 test my-group ute + volumes: zookeeper-data: zookeeper-log: diff --git a/pom.xml b/pom.xml deleted file mode 100644 index a1007175..00000000 --- a/pom.xml +++ /dev/null @@ -1,99 +0,0 @@ - - - - 4.0.0 - - - org.springframework.boot - spring-boot-starter-parent - 3.4.1 - - - - de.juplo.kafka - simple-consumer - Simple Consumer-Group - Super Simple Consumer-Group, that is implemented as a plain Java-program - 1.0-SNAPSHOT - - - 21 - - - - - org.apache.kafka - kafka-clients - - - org.projectlombok - lombok - compile - - - ch.qos.logback - logback-classic - - - - - - - org.apache.maven.plugins - maven-dependency-plugin - - - copy-dependencies - package - - copy-dependencies - - - ${project.build.directory}/libs - - - - - - org.apache.maven.plugins - maven-jar-plugin - - - - true - libs/ - de.juplo.kafka.ExampleConsumer - - - - - - pl.project13.maven - git-commit-id-plugin - - - io.fabric8 - docker-maven-plugin - 0.45.0 - - - - juplo/%a:%v - - - - - - build - package - - build - - - - - - - - diff --git a/settings.gradle b/settings.gradle deleted file mode 100644 index 71fb5737..00000000 --- a/settings.gradle +++ /dev/null @@ -1 +0,0 @@ -rootProject.name = 'simple-consumer' diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java deleted file mode 100644 index 7be827ef..00000000 --- a/src/main/java/de/juplo/kafka/ExampleConsumer.java +++ /dev/null @@ -1,141 +0,0 @@ -package de.juplo.kafka; - -import lombok.extern.slf4j.Slf4j; -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.clients.consumer.Consumer; -import org.apache.kafka.clients.consumer.KafkaConsumer; -import org.apache.kafka.common.errors.WakeupException; -import org.apache.kafka.common.serialization.StringDeserializer; - -import java.time.Duration; -import java.util.Arrays; -import java.util.Properties; - - -@Slf4j -public class ExampleConsumer -{ - private final String id; - private final String topic; - private final Consumer consumer; - - private volatile boolean running = false; - private long consumed = 0; - - public ExampleConsumer( - String broker, - String topic, - String groupId, - String clientId) - { - Properties props = new Properties(); - props.put("bootstrap.servers", broker); - props.put("group.id", groupId); // ID für die Offset-Commits - props.put("client.id", clientId); // Nur zur Wiedererkennung - props.put("key.deserializer", StringDeserializer.class.getName()); - props.put("value.deserializer", StringDeserializer.class.getName()); - - this.id = clientId; - this.topic = topic; - consumer = new KafkaConsumer<>(props); - } - - - public void run() - { - try - { - log.info("{} - Subscribing to topic {}", id, topic); - consumer.subscribe(Arrays.asList(topic)); - running = true; - - while (true) - { - ConsumerRecords records = consumer.poll(Duration.ofSeconds(1)); - - log.info("{} - Received {} messages", id, records.count()); - for (ConsumerRecord record : records) - { - handleRecord( - record.topic(), - record.partition(), - record.offset(), - record.key(), - record.value()); - } - } - } - catch(WakeupException e) - { - log.info("{} - Consumer was signaled to finish its work", id); - } - catch(Exception e) - { - log.error("{} - Unexpected error, unsubscribing!", id, e); - consumer.unsubscribe(); - } - finally - { - log.info("{} - Closing the KafkaConsumer", id); - consumer.close(); - log.info("{}: Consumed {} messages in total, exiting!", id, consumed); - running = false; - } - } - - private void handleRecord( - String topic, - Integer partition, - Long offset, - String key, - String value) - { - consumed++; - log.info("{} - partition={}-{}, offset={}: {}={}", id, topic, partition, offset, key, value); - } - - - public static void main(String[] args) throws Exception - { - if (args.length != 4) - { - log.error("Four arguments required!"); - log.error("args[0]: Broker-Address"); - log.error("args[1]: Topic"); - log.error("args[2]: Group-ID"); - log.error("args[3]: Unique Client-ID"); - System.exit(1); - return; - } - - - log.info( - "Running ExampleConsumer: broker={}, topic={}, group-id={}, client-id={}", - args[0], - args[1], - args[2], - args[3]); - - ExampleConsumer instance = new ExampleConsumer(args[0], args[1], args[2], args[3]); - - Runtime.getRuntime().addShutdownHook(new Thread(() -> - { - instance.consumer.wakeup(); - - while (instance.running) - { - log.info("Waiting for main-thread..."); - try - { - Thread.sleep(1000); - } - catch (InterruptedException e) {} - } - log.info("Shutdown completed."); - })); - - instance.run(); - } -} - diff --git a/src/main/resources/logback.xml b/src/main/resources/logback.xml deleted file mode 100644 index 7a25e76f..00000000 --- a/src/main/resources/logback.xml +++ /dev/null @@ -1,16 +0,0 @@ - - - - - - %d{HH:mm:ss.SSS} | %highlight(%-5level) %msg%n - - - - - - - - - -