From: Kai Moritz <kai@juplo.de>
Date: Sun, 10 Nov 2024 16:30:57 +0000 (+0100)
Subject: Version des ``spring-consumer``, die ``long``-Werte erwartet
X-Git-Tag: consumer/spring-consumer--long--2025-04-signal-spickzettel
X-Git-Url: https://juplo.de/gitweb/?a=commitdiff_plain;h=refs%2Fheads%2Fconsumer%2Fspring-consumer--long;p=demos%2Fkafka%2Ftraining
Version des ``spring-consumer``, die ``long``-Werte erwartet
---
diff --git a/README.sh b/README.sh
index b46e2350..86b98061 100755
--- a/README.sh
+++ b/README.sh
@@ -1,6 +1,6 @@
#!/bin/bash
-IMAGE=juplo/spring-consumer:1.1-SNAPSHOT
+IMAGE=juplo/spring-consumer:1.1-long-SNAPSHOT
if [ "$1" = "cleanup" ]
then
@@ -10,7 +10,7 @@ then
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
+docker compose -f docker/docker-compose.yml rm -svf peter ute
if [[
$(docker image ls -q $IMAGE) == "" ||
@@ -27,13 +27,16 @@ 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
+docker compose -f docker/docker-compose.yml up -d peter ute
+sleep 15
-sleep 5
-docker compose -f docker/docker-compose.yml stop consumer
+docker compose -f docker/docker-compose.yml stop producer
-docker compose -f docker/docker-compose.yml start consumer
-sleep 5
+echo
+echo "Von peter empfangen:"
+docker compose -f docker/docker-compose.yml logs peter | grep '\ test\/.'
+echo
+echo "Von ute empfangen:"
+docker compose -f docker/docker-compose.yml logs ute | grep '\ test\/.'
-docker compose -f docker/docker-compose.yml stop producer consumer
-docker compose -f docker/docker-compose.yml logs consumer
+docker compose -f docker/docker-compose.yml stop peter ute
diff --git a/build.gradle b/build.gradle
index a8614fdf..12188d65 100644
--- a/build.gradle
+++ b/build.gradle
@@ -8,7 +8,7 @@ plugins {
}
group = 'de.juplo.kafka'
-version = '1.1-SNAPSHOT'
+version = '1.1-long-SNAPSHOT'
java {
toolchain {
diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
index 9984f6b8..3338c62a 100644
--- a/docker/docker-compose.yml
+++ b/docker/docker-compose.yml
@@ -136,29 +136,29 @@ services:
- kafka-3
producer:
- image: juplo/simple-producer:1.0-SNAPSHOT
- command: kafka:9092 test producer
+ image: juplo/spring-producer:2.0-long-SNAPSHOT
+ environment:
+ spring.kafka.bootstrap-servers: kafka:9092
+ spring.kafka.client-id: producer
+ juplo.producer.topic: test
consumer:
- image: juplo/spring-consumer:1.1-SNAPSHOT
+ image: juplo/spring-consumer:1.1-long-SNAPSHOT
environment:
juplo.bootstrap-server: kafka:9092
juplo.client-id: consumer
- juplo.consumer.topic: test
peter:
- image: juplo/spring-consumer:1.1-SNAPSHOT
+ image: juplo/spring-consumer:1.1-long-SNAPSHOT
environment:
juplo.bootstrap-server: kafka:9092
juplo.client-id: peter
- juplo.consumer.topic: test
ute:
- image: juplo/spring-consumer:1.1-SNAPSHOT
+ image: juplo/spring-consumer:1.1-long-SNAPSHOT
environment:
juplo.bootstrap-server: kafka:9092
juplo.client-id: ute
- juplo.consumer.topic: test
volumes:
zookeeper-data:
diff --git a/pom.xml b/pom.xml
index dd96d00f..d99ed82f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -15,7 +15,7 @@
<artifactId>spring-consumer</artifactId>
<name>Spring Consumer</name>
<description>Super Simple Consumer-Group, that is implemented as Spring-Boot application and configured by Spring Kafka</description>
- <version>1.1-SNAPSHOT</version>
+ <version>1.1-long-SNAPSHOT</version>
<properties>
<java.version>21</java.version>
diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index ea6b64ea..cb2057b8 100644
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -2,6 +2,7 @@ package de.juplo.kafka;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ConfigurableApplicationContext;
@@ -17,7 +18,7 @@ public class ApplicationConfiguration
{
@Bean
public ExampleConsumer exampleConsumer(
- Consumer<String, String> kafkaConsumer,
+ Consumer<String, Long> kafkaConsumer,
ApplicationProperties properties,
ConfigurableApplicationContext applicationContext)
{
@@ -30,7 +31,7 @@ public class ApplicationConfiguration
}
@Bean(destroyMethod = "")
- public KafkaConsumer<String, String> kafkaConsumer(ApplicationProperties properties)
+ public KafkaConsumer<String, Long> kafkaConsumer(ApplicationProperties properties)
{
Properties props = new Properties();
props.put("bootstrap.servers", properties.getBootstrapServer());
@@ -46,7 +47,7 @@ public class ApplicationConfiguration
}
props.put("metadata.max.age.ms", 5000); // 5 Sekunden
props.put("key.deserializer", StringDeserializer.class.getName());
- props.put("value.deserializer", StringDeserializer.class.getName());
+ props.put("value.deserializer", LongDeserializer.class.getName());
return new KafkaConsumer<>(props);
}
diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index 1f5a5706..83a9f89b 100644
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -15,7 +15,7 @@ public class ExampleConsumer implements Runnable
{
private final String id;
private final String topic;
- private final Consumer<String, String> consumer;
+ private final Consumer<String, Long> consumer;
private final Thread workerThread;
private final Runnable closeCallback;
@@ -25,7 +25,7 @@ public class ExampleConsumer implements Runnable
public ExampleConsumer(
String clientId,
String topic,
- Consumer<String, String> consumer,
+ Consumer<String, Long> consumer,
Runnable closeCallback)
{
this.id = clientId;
@@ -49,10 +49,10 @@ public class ExampleConsumer implements Runnable
while (true)
{
- ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
+ ConsumerRecords<String, Long> records = consumer.poll(Duration.ofSeconds(1));
log.info("{} - Received {} messages", id, records.count());
- for (ConsumerRecord<String, String> record : records)
+ for (ConsumerRecord<String, Long> record : records)
{
handleRecord(
record.topic(),
@@ -87,7 +87,7 @@ public class ExampleConsumer implements Runnable
Integer partition,
Long offset,
String key,
- String value)
+ Long value)
{
consumed++;
log.info("{} - partition={}-{}, offset={}: {}={}", id, topic, partition, offset, key, value);