+++ /dev/null
-#!/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
+++ /dev/null
-plugins {
- id 'java'
- id 'org.springframework.boot' version '4.0.6'
- id 'io.spring.dependency-management' version '1.1.7'
- id 'com.gorylenko.gradle-git-properties' version '2.4.2'
- id 'com.google.cloud.tools.jib' version '3.4.5'
-}
-
-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'
-}
-
-tasks.named('test') {
- useJUnitPlatform()
-}
-
-jib {
- from { image = 'eclipse-temurin:21-jre' }
- to { image = "juplo/${project.name}:${project.version}" }
- container { mainClass = 'de.juplo.kafka.ExampleConsumer' }
-}
echo -n Bereits konfiguriert:
cat INITIALIZED
kafka-topics --bootstrap-server kafka:9092 --describe --topic test
+ kafka-topics --bootstrap-server kafka:9092 --describe --topic foo
+ kafka-topics --bootstrap-server kafka:9092 --describe --topic bar
else
kafka-topics --bootstrap-server kafka:9092 \
--delete \
--if-exists \
--topic test
+ kafka-topics --bootstrap-server kafka:9092 \
+ --delete \
+ --if-exists \
+ --topic foo
+ kafka-topics --bootstrap-server kafka:9092 \
+ --delete \
+ --if-exists \
+ --topic bar
kafka-topics --bootstrap-server kafka:9092 \
--create \
--topic test \
--config min.insync.replicas=2 \
&& echo Das Topic \'test\' wurde erfolgreich angelegt: \
&& kafka-topics --bootstrap-server kafka:9092 --describe --topic test \
+ && kafka-topics --bootstrap-server kafka:9092 \
+ --create \
+ --topic foo \
+ --partitions 2 \
+ --replication-factor 3 \
+ --config min.insync.replicas=2 \
+ && echo Das Topic \'foo\' wurde erfolgreich angelegt: \
+ && kafka-topics --bootstrap-server kafka:9092 --describe --topic foo \
+ && kafka-topics --bootstrap-server kafka:9092 \
+ --create \
+ --topic bar \
+ --partitions 2 \
+ --replication-factor 3 \
+ --config min.insync.replicas=2 \
+ && echo Das Topic \'bar\' wurde erfolgreich angelegt: \
+ && kafka-topics --bootstrap-server kafka:9092 --describe --topic bar \
&& date > INITIALIZED
fi
stop_grace_period: 0s
depends_on:
- cli
+ - akhq
+ - producer
+ - producer-foo
+ - producer-bar
akhq:
image: tchiotludo/akhq:0.23.0
cpu_quota: 50000
mem_limit: 100m
+ producer-foo:
+ image: juplo/simple-producer:1.0-SNAPSHOT
+ command: kafka:9092 foo producer-foo
+
+ producer-bar:
+ image: juplo/simple-producer:1.0-SNAPSHOT
+ command: kafka:9092 bar producer-bar
+
consumer:
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
+ command: kafka:9092 foo my-group peter
ute:
image: juplo/simple-consumer:1.0-SNAPSHOT
- command: kafka:9092 test my-group ute
+ command: kafka:9092 bar my-group ute
spickzettel:
image: juplo/toolbox
+++ /dev/null
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
-
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.springframework.boot</groupId>
- <artifactId>spring-boot-starter-parent</artifactId>
- <version>4.0.6</version>
- <relativePath/> <!-- lookup parent from repository -->
- </parent>
-
- <groupId>de.juplo.kafka</groupId>
- <artifactId>simple-consumer</artifactId>
- <name>Simple Consumer-Group</name>
- <description>Super Simple Consumer-Group, that is implemented as a plain Java-program</description>
- <version>1.0-SNAPSHOT</version>
-
- <properties>
- <java.version>21</java.version>
- </properties>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka-clients</artifactId>
- </dependency>
- <dependency>
- <groupId>org.projectlombok</groupId>
- <artifactId>lombok</artifactId>
- <optional>true</optional>
- </dependency>
- <dependency>
- <groupId>ch.qos.logback</groupId>
- <artifactId>logback-classic</artifactId>
- </dependency>
- </dependencies>
-
- <build>
- <plugins>
- <plugin>
- <groupId>pl.project13.maven</groupId>
- <artifactId>git-commit-id-plugin</artifactId>
- </plugin>
- <plugin>
- <groupId>com.google.cloud.tools</groupId>
- <artifactId>jib-maven-plugin</artifactId>
- <version>3.4.5</version>
- <configuration>
- <from>
- <image>eclipse-temurin:21-jre</image>
- </from>
- <to>
- <image>juplo/${project.artifactId}:${project.version}</image>
- </to>
- <container>
- <mainClass>de.juplo.kafka.ExampleConsumer</mainClass>
- </container>
- </configuration>
- <executions>
- <execution>
- <phase>package</phase>
- <goals>
- <goal>dockerBuild</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
-
-</project>
+++ /dev/null
-rootProject.name = 'simple-consumer'
+++ /dev/null
-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<String, String> 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<String, String> records = consumer.poll(Duration.ofSeconds(1));
-
- log.info("{} - Received {} messages", id, records.count());
- for (ConsumerRecord<String, String> 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...", instance.id);
- try
- {
- Thread.sleep(1000);
- }
- catch (InterruptedException e) {}
- }
- log.info("{} - Shutdown completed.", instance.id);
- }));
-
- instance.run();
- }
-}
-
+++ /dev/null
-<?xml version="1.0" encoding="UTF-8"?>
-<configuration>
-
- <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
- <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
- <Pattern>%d{HH:mm:ss.SSS} | %highlight(%-5level) %msg%n</Pattern>
- </encoder>
- </appender>
-
- <logger name="de.juplo" level="TRACE"/>
-
- <root level="INFO">
- <appender-ref ref="STDOUT" />
- </root>
-
-</configuration>
projects / demos / kafka / training / commitdiff