From: Kai Moritz <kai@juplo.de>
Date: Sat, 24 Feb 2024 11:56:24 +0000 (+0100)
Subject: fix: The number of chards is the number of partitions of the `data_channel`
X-Git-Tag: rebase--2024-02-27--11-49~22
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=e2933c8acd7ffd1c3a8530b0d34621c183ef09fa;p=demos%2Fkafka%2Fchat

fix: The number of chards is the number of partitions of the `data_channel`
---

diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/InfoChannel.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/InfoChannel.java
index 2df7573c..f5d5253c 100644
--- a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/InfoChannel.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/InfoChannel.java
@@ -48,6 +48,7 @@ public class InfoChannel implements Runnable
     String topic,
     Producer<String, AbstractMessageTo> producer,
     Consumer<String, AbstractMessageTo> infoChannelConsumer,
+    int numShards,
     URI instanceUri)
   {
     log.debug(
@@ -58,9 +59,7 @@ public class InfoChannel implements Runnable
     this.producer = producer;
     this.chatRoomInfo = new HashMap<>();
 
-    this.numShards = consumer
-        .partitionsFor(topic)
-        .size();
+    this.numShards = numShards;
     this.shardOwners = new String[numShards];
     this.currentOffset = new long[numShards];
     this.nextOffset = new long[numShards];
diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java
index b5a442f8..5a41ebcc 100644
--- a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java
@@ -130,6 +130,7 @@ public class KafkaServicesConfiguration
         properties.getKafka().getInfoChannelTopic(),
         producer,
         infoChannelConsumer,
+        properties.getKafka().getNumPartitions(),
         properties.getKafka().getInstanceUri());
   }