From be27a128b7606623400e5ada1b23e6b19b4644ad 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 ec6079b8..c13f713e 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 @@ -284,7 +284,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