Der Producer schreibt gezielt in eine fixe (konfigurierbare) Partition

diff --git a/README.sh b/README.sh
index 499780a90952ea229e7f044e46eaf2e92aa79fe1..668e756fa06dfba5d50f023c6a82a3974125357e 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/docker/docker-compose.yml b/docker/docker-compose.yml
index c417a7fd14b2571b0773850162fdc1e34d83cbd4..d482e488da36d036e822ea2be0707fc20d848a5c 100644 (file)
--- a/docker/docker-compose.yml
+++ b/docker/docker-compose.yml
@@ -190,11 +190,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-1:
     image: juplo/simple-consumer:1.0-SNAPSHOT

diff --git a/pom.xml b/pom.xml
index 841299b7ed04518f1158efbb435b93b10c00fdcb..c3f745f4a7743b1f49bec90d5b1f12de9489ba72 100644 (file)
--- a/pom.xml
+++ b/pom.xml
@@ -15,7 +15,7 @@
   <artifactId>spring-producer</artifactId>
   <name>Spring Producer</name>
   <description>A Simple Spring-Boot-Producer, that takes messages via POST and confirms successs</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 2be61daf55edbb23936ddbb06adc73d03c5d9ed8..4d45c04724d19787dd83275b2ff879e5620f1220 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -23,6 +23,7 @@ public class ApplicationConfiguration
         new ExampleProducer(
             properties.getClientId(),
             properties.getProducerProperties().getTopic(),
+            properties.getProducerProperties().getPartition(),
             properties.getProducerProperties().getThrottleMs() == null
               ? 500
               : properties.getProducerProperties().getThrottleMs(),

diff --git a/src/main/java/de/juplo/kafka/ApplicationProperties.java b/src/main/java/de/juplo/kafka/ApplicationProperties.java
index 5cb9aa0517c37979275bca2a2b6f9499bdb879ea..8eebd093a88da2f75183b4ff7d29a5e266a37e97 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationProperties.java
+++ b/src/main/java/de/juplo/kafka/ApplicationProperties.java
@@ -40,6 +40,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 6f7c093c13ef3b9357e9f78426ec5b7225877d61..2d043c2c3e624f26804149677cd5ccd27bfb430d 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleProducer.java
+++ b/src/main/java/de/juplo/kafka/ExampleProducer.java
@@ -10,6 +10,7 @@ public class ExampleProducer implements Runnable
 {
   private final String id;
   private final String topic;
+  private final int partition;
   private final int throttleMs;
   private final Producer<String, String> producer;
   private final Thread workerThread;
@@ -21,11 +22,13 @@ public class ExampleProducer implements Runnable
   public ExampleProducer(
     String id,
     String topic,
+    int partition,
     int throttleMs,
     Producer<String, String> producer)
   {
     this.id = id;
     this.topic = topic;
+    this.partition = partition;
     this.throttleMs = throttleMs;
     this.producer = producer;
 
@@ -74,6 +77,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 85aee9d5c661e762de8dcffda20e6fff9af29713..79c5c98de1b4b79d1eaa8162b99e7044f0b7aa50 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
     batch-size: 16384
     linger-ms: 0
@@ -27,6 +28,7 @@ info:
     client-id: ${juplo.client-id}
     producer:
       topic: ${juplo.producer.topic}
+      partition: ${juplo.producer.partition}
       acks: ${juplo.producer.acks}
       batch-size: ${juplo.producer.batch-size}
       linger-ms: ${juplo.producer.linger-ms}