[direct-io-kafka] fix bounded read

datadrivencz · Mar 10, 2025 · d61ed51 · d61ed51
1 parent 4f06f05
commit d61ed51
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/direct/io-kafka/src/main/java/cz/o2/proxima/direct/io/kafka/KafkaLogReader.java b/direct/io-kafka/src/main/java/cz/o2/proxima/direct/io/kafka/KafkaLogReader.java
@@ -495,6 +495,7 @@ private void submitConsumerWithObserver(
                 handleRebalanceInOffsetCommit(kafka, listener);
               }
               rethrowErrorIfPresent(name, error);
+              log.debug("Current endOffsets {}, polledOffsets {}", endOffsets, polledOffsets);
               terminateIfConsumed(stopAtCurrent, kafka, endOffsets, polledOffsets, completed);
 
               progressWatermarkOnEmptyPartitions(
@@ -581,6 +582,7 @@ private static void notifyAssignedPartitions(
       KafkaConsumer<?, ?> kafka, ConsumerRebalanceListener listener) {
 
     Set<TopicPartition> assignment = kafka.assignment();
+    log.debug("Assignment before notification is {}", assignment);
     if (!assignment.isEmpty()) {
       listener.onPartitionsRevoked(assignment);
       listener.onPartitionsAssigned(assignment);
@@ -937,7 +939,7 @@ public void onPartitionsAssigned(Collection<TopicPartition> parts) {
                           ? getCommittedTopicOffsets(currentlyAssigned, c)
                           : getCurrentTopicOffsets(currentlyAssigned, c);
                   newOffsets.stream()
-                      .filter(o -> o.getOffset() > 0)
+                      .filter(o -> o.getOffset() >= 0)
                       .forEach(
                           o ->
                               polledOffsets.put(