X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fchat%2Fbackend%2Fimplementation%2Fkafka%2FDataChannel.java;h=f139b765068ed58136dbcf9c3517f81c2c5135c3;hb=d958fc6f355071a567cf2b1b048c53e124fb4f00;hp=2fa4998e72c406afad6705192ef2c535bcd483ac;hpb=bab48fac3824b1dbf660881e5c517dc98ab9cc8f;p=demos%2Fkafka%2Fchat

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 2fa4998e..f139b765 100644
--- 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
@@ -7,10 +7,7 @@ import de.juplo.kafka.chat.backend.implementation.kafka.messages.AbstractMessage
 import de.juplo.kafka.chat.backend.implementation.kafka.messages.data.EventChatMessageReceivedTo;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.*;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
@@ -147,11 +144,16 @@ public class DataChannel implements Runnable, ConsumerRebalanceListener
       infoChannel.sendShardAssignedEvent(partition);
       shardingPublisherStrategy
           .publishOwnership(partition)
-          .doOnNext(instanceId -> log.info(
-              "Instance {} was published as owner of shard {}",
+          .doOnSuccess(instanceId -> log.info(
+              "Successfully published instance {} as owner of shard {}",
               instanceId,
               partition))
-          .subscribe();
+          .doOnError(throwable -> log.error(
+              "Could not publish instance {} as owner of shard {}: {}",
+              instanceId,
+              partition,
+              throwable))
+          .block();
     });
 
     consumer.resume(partitions);
@@ -164,6 +166,7 @@ public class DataChannel implements Runnable, ConsumerRebalanceListener
     {
       int partition = topicPartition.partition();
       isShardOwned[partition] = false;
+      nextOffset[partition] = consumer.position(topicPartition);
       log.info("Partition revoked: {} - next={}", partition, nextOffset[partition]);
       infoChannel.sendShardRevokedEvent(partition);
     });
@@ -266,6 +269,11 @@ public class DataChannel implements Runnable, ConsumerRebalanceListener
     KafkaChatMessageService kafkaChatRoomService =
         (KafkaChatMessageService) chatRoomData.getChatRoomService();
 
+    log.debug(
+        "Loaded message from partition={} at offset={}: {}",
+        partition,
+        offset,
+        message);
     kafkaChatRoomService.persistMessage(message);
   }
 
@@ -274,7 +282,12 @@ public class DataChannel implements Runnable, ConsumerRebalanceListener
     return IntStream
         .range(0, numShards)
         .filter(shard -> isShardOwned[shard])
-        .allMatch(shard -> nextOffset[shard] >= currentOffset[shard]);
+        .allMatch(shard ->
+        {
+          TopicPartition partition = new TopicPartition(topic, shard);
+          long position = consumer.position(partition);
+          return position >= currentOffset[shard];
+        });
   }
 
   private void pauseAllOwnedPartions()
@@ -319,4 +332,9 @@ public class DataChannel implements Runnable, ConsumerRebalanceListener
     KafkaChatMessageService service = new KafkaChatMessageService(this, chatRoomId);
     return new ChatRoomData(clock, service, bufferSize);
   }
+
+  ConsumerGroupMetadata getConsumerGroupMetadata()
+  {
+    return consumer.groupMetadata();
+  }
 }