From 462962ac56c18fd652ac035bde08bf89866543b8 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sun, 24 Sep 2023 21:39:01 +0200 Subject: [PATCH] fix: The actual position has to be requested from the consumer * If the last seen offset and the current offset differ, although the partition did not contain any messages between this offsets, the loading process got stuck, because the position never advanced. * Therefore, the actual position, that is compared against the read end-offset, has to be requested from the consumer. --- .../chat/backend/implementation/kafka/DataChannel.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) 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 c2908461..5680360f 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 @@ -279,7 +279,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() -- 2.20.1