WIP:haproxy

diff --git a/src/main/java/de/juplo/kafka/chat/backend/domain/ShardingPublisherStrategy.java b/src/main/java/de/juplo/kafka/chat/backend/domain/ShardingPublisherStrategy.java
index 59fd80c..9a1c725 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/domain/ShardingPublisherStrategy.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/domain/ShardingPublisherStrategy.java
@@ -5,5 +5,5 @@ import reactor.core.publisher.Mono;
 
 public interface ShardingPublisherStrategy
 {
-  Mono<String[]> publishOwnership(int shard);
+  Mono<String> publishOwnership(int shard);
 }

diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/DataChannel.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/DataChannel.java
index e0d42d4..114619e 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/DataChannel.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/DataChannel.java
@@ -142,7 +142,14 @@ public class DataChannel implements Runnable, ConsumerRebalanceListener
 
       consumer.seek(topicPartition, nextOffset[partition]);
       infoChannel.sendShardAssignedEvent(partition);
-      shardingPublisherStrategy.publishOwnership(partition);
+      shardingPublisherStrategy
+          .publishOwnership(partition)
+          .map(instanceId -> "Instance "
+              + instanceId
+              + " was published as owner of shard "
+              + partition)
+          .doOnError(e -> log.error("Could not publish ownershit for shard " + partition + ": " + e))
+          .subscribe();
     });
 
     consumer.resume(partitions);