Der Producer schreibt gezielt in eine fixe (konfigurierbare) Partition

diff --git a/README.sh b/README.sh
index c8a0b221b5f6e6cb2ca15e8b8c3c100a5e0c45e5..ab112868433ba19f5eabe164bc240709f5d811b3 100755 (executable)
--- a/README.sh
+++ b/README.sh
@@ -1,6 +1,6 @@
 #!/bin/bash
 
-IMAGE=juplo/spring-producer:1.0-SNAPSHOT
+IMAGE=juplo/spring-producer:1.0-fixedsharding-SNAPSHOT
 
 if [ "$1" = "cleanup" ]
 then

diff --git a/build.gradle b/build.gradle
index 1429c4dd737dcce22957e983bcb8f67635c7a0a5..d60423c55a3136ed31b969b899e11d3101ec9835 100644 (file)
--- a/build.gradle
+++ b/build.gradle
@@ -8,7 +8,7 @@ plugins {
 }
 
 group = 'de.juplo.kafka'
-version = '1.0-SNAPSHOT'
+version = '1.0-fixedsharding-SNAPSHOT'
 
 java {
        toolchain {

diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
index 00f68fcc477576d720a14eeff9dac33e44627817..c5b1b5ade1820de5c54fca72debe8b4c0336edc2 100644 (file)
--- a/docker/docker-compose.yml
+++ b/docker/docker-compose.yml
@@ -136,11 +136,12 @@ services:
       - kafka-3
 
   producer:
-    image: juplo/spring-producer:1.0-SNAPSHOT
+    image: juplo/spring-producer:1.0-fixedsharding-SNAPSHOT
     environment:
       juplo.bootstrap-server: kafka:9092
       juplo.client-id: producer
       juplo.producer.topic: test
+      juplo.producer.partition: 0
 
   consumer:
     image: juplo/simple-consumer:1.0-SNAPSHOT

diff --git a/pom.xml b/pom.xml
index f64266b4f648b0d4c915e289da8a3e4284399f90..7a47d40fcd5bbe0104c41d9a425528de71486b59 100644 (file)
--- a/pom.xml
+++ b/pom.xml
@@ -15,7 +15,7 @@
   <artifactId>spring-producer</artifactId>
   <name>Spring Producer</name>
   <description>A Simple Producer, based on Spring Boot, that sends messages via Kafka</description>
-  <version>1.0-SNAPSHOT</version>
+  <version>1.0-fixedsharding-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 0090ceeab45f9b9e4a262dada2ef52a37744ba7a..ab4892561044a75d9eca2714d84858c49a18b1ab 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -26,6 +26,7 @@ public class ApplicationConfiguration
       new ExampleProducer(
         properties.getClientId(),
         properties.getProducerProperties().getTopic(),
+        properties.getProducerProperties().getPartition(),
         properties.getProducerProperties().getThrottle() == null
           ? Duration.ofMillis(500)
           : properties.getProducerProperties().getThrottle(),

diff --git a/src/main/java/de/juplo/kafka/ApplicationProperties.java b/src/main/java/de/juplo/kafka/ApplicationProperties.java
index 43232628755882f32117ac1a72c5ade16dd13f1e..65b71f3a542a2106022de48946027a263b3f38bc 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationProperties.java
+++ b/src/main/java/de/juplo/kafka/ApplicationProperties.java
@@ -42,6 +42,8 @@ public class ApplicationProperties
     @NotEmpty
     private String topic;
     @NotNull
+    private Integer partition;
+    @NotNull
     @NotEmpty
     private String acks;
     @NotNull

diff --git a/src/main/java/de/juplo/kafka/ExampleProducer.java b/src/main/java/de/juplo/kafka/ExampleProducer.java
index c5a5a80d2d2a1798ac0c59873c5bd9ec833db5c4..ae13e131089fe4e48e778e2ba328b78be5604dd1 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleProducer.java
+++ b/src/main/java/de/juplo/kafka/ExampleProducer.java
@@ -12,6 +12,7 @@ public class ExampleProducer implements Runnable
 {
   private final String id;
   private final String topic;
+  private final int partition;
   private final Duration throttle;
   private final Producer<String, String> producer;
   private final Thread workerThread;
@@ -24,12 +25,14 @@ public class ExampleProducer implements Runnable
   public ExampleProducer(
     String id,
     String topic,
+    int partition,
     Duration throttle,
     Producer<String, String> producer,
     Runnable closeCallback)
   {
     this.id = id;
     this.topic = topic;
+    this.partition = partition;
     this.throttle = throttle;
     this.producer = producer;
 
@@ -84,6 +87,7 @@ public class ExampleProducer implements Runnable
 
     final ProducerRecord<String, String> record = new ProducerRecord<>(
       topic,  // Topic
+      partition, // Partition
       key,    // Key
       value   // Value
     );

diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
index 21fef284f03e091e3a9e7f87758e5fef9fb2fe42..332873fd5a7d02aaa59015c905a942425f53e830 100644 (file)
--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@ -3,6 +3,7 @@ juplo:
   client-id: DEV
   producer:
     topic: test
+    partition: 0
     acks: -1
     delivery-timeout: 10s
     max-block: 5s
@@ -30,6 +31,7 @@ info:
     client-id: ${juplo.client-id}
     producer:
       topic: ${juplo.producer.topic}
+      partition: ${juplo.producer.partition}
       acks: ${juplo.producer.acks}
       delivery-timeout: ${juplo.producer.delivery-timeout}
       max-block: ${juplo.producer.max-block}