FE: fix missing previousCursorId in case of empty pollRange (kafbat#550)

hadisfr · Sep 17, 2024 · 50aa173 · 50aa173
1 parent 410826e
commit 50aa173
Show file tree

Hide file tree

Showing 4 changed files with 9 additions and 10 deletions.
diff --git a/api/src/main/java/io/kafbat/ui/emitter/AbstractEmitter.java b/api/src/main/java/io/kafbat/ui/emitter/AbstractEmitter.java
@@ -40,9 +40,9 @@ protected void sendConsuming(FluxSink<TopicMessageEventDTO> sink, PolledRecords
     messagesProcessing.sentConsumingInfo(sink, records);
   }
 
-  // cursor is null if target partitions were fully polled (no, need to do paging)
-  protected void sendFinishStatsAndCompleteSink(FluxSink<TopicMessageEventDTO> sink, @Nullable Cursor.Tracking cursor) {
-    messagesProcessing.sendFinishEvents(sink, cursor);
+  protected void sendFinishStatsAndCompleteSink(FluxSink<TopicMessageEventDTO> sink, Cursor.Tracking cursor,
+                                                boolean hasNext) {
+    messagesProcessing.sendFinishEvents(sink, cursor, hasNext);
     sink.complete();
   }
 }
diff --git a/api/src/main/java/io/kafbat/ui/emitter/ConsumingStats.java b/api/src/main/java/io/kafbat/ui/emitter/ConsumingStats.java
@@ -3,7 +3,6 @@
 import io.kafbat.ui.model.TopicMessageConsumingDTO;
 import io.kafbat.ui.model.TopicMessageEventDTO;
 import io.kafbat.ui.model.TopicMessagePageCursorDTO;
-import javax.annotation.Nullable;
 import reactor.core.publisher.FluxSink;
 
 class ConsumingStats {
@@ -28,8 +27,8 @@ void incFilterApplyError() {
     filterApplyErrors++;
   }
 
-  void sendFinishEvent(FluxSink<TopicMessageEventDTO> sink, @Nullable Cursor.Tracking cursor) {
-    String previousCursorId = cursor != null ? cursor.getPreviousCursorId() : null;
+  void sendFinishEvent(FluxSink<TopicMessageEventDTO> sink, Cursor.Tracking cursor, boolean hasNext) {
+    String previousCursorId = cursor.getPreviousCursorId();
     sink.next(
         new TopicMessageEventDTO()
             .type(TopicMessageEventDTO.TypeEnum.DONE)
@@ -39,7 +38,7 @@ void sendFinishEvent(FluxSink<TopicMessageEventDTO> sink, @Nullable Cursor.Track
                     : null
             )
             .nextCursor(
-                cursor != null
+                hasNext
                     ? new TopicMessagePageCursorDTO().id(cursor.registerCursor())
                     : null
             )

diff --git a/api/src/main/java/io/kafbat/ui/emitter/MessagesProcessing.java b/api/src/main/java/io/kafbat/ui/emitter/MessagesProcessing.java
@@ -72,9 +72,9 @@ void sentConsumingInfo(FluxSink<TopicMessageEventDTO> sink, PolledRecords polled
     }
   }
 
-  void sendFinishEvents(FluxSink<TopicMessageEventDTO> sink, @Nullable Cursor.Tracking cursor) {
+  void sendFinishEvents(FluxSink<TopicMessageEventDTO> sink, Cursor.Tracking cursor, boolean hasNext) {
     if (!sink.isCancelled()) {
-      consumingStats.sendFinishEvent(sink, cursor);
+      consumingStats.sendFinishEvent(sink, cursor, hasNext);
     }
   }
 

diff --git a/api/src/main/java/io/kafbat/ui/emitter/RangePollingEmitter.java b/api/src/main/java/io/kafbat/ui/emitter/RangePollingEmitter.java
@@ -64,7 +64,7 @@ public void accept(FluxSink<TopicMessageEventDTO> sink) {
       if (sink.isCancelled()) {
         log.debug("Polling finished due to sink cancellation");
       }
-      sendFinishStatsAndCompleteSink(sink, pollRange.isEmpty() ? null : cursor);
+      sendFinishStatsAndCompleteSink(sink, cursor, !pollRange.isEmpty());
       log.debug("Polling finished");
     } catch (InterruptException kafkaInterruptException) {
       log.debug("Polling finished due to thread interruption");