GH-333: Support currentLag() in doOnConsumer

Resolves #333
reactor · Oct 18, 2023 · 1a44d5c · 1a44d5c
1 parent ab9f851
commit 1a44d5c
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 2 deletions.
diff --git a/src/main/java/reactor/kafka/receiver/internals/ConsumerHandler.java b/src/main/java/reactor/kafka/receiver/internals/ConsumerHandler.java
@@ -65,7 +65,8 @@ class ConsumerHandler<K, V> {
         "resume",
         "offsetsForTimes",
         "beginningOffsets",
-        "endOffsets"
+        "endOffsets",
+        "currentLag"
     ));
 
     private static final AtomicInteger COUNTER = new AtomicInteger();

diff --git a/src/test/java/reactor/kafka/receiver/KafkaReceiverTest.java b/src/test/java/reactor/kafka/receiver/KafkaReceiverTest.java
@@ -59,6 +59,7 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.OptionalLong;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
@@ -288,12 +289,17 @@ public void manualAssignmentWithCommit() throws Exception {
         receiverOptions = receiverOptions.commitInterval(Duration.ZERO)
                 .commitBatchSize(0)
                 .assignment(getTopicPartitions());
+        KafkaReceiver<Integer, String> receiver = KafkaReceiver.create(receiverOptions);
         Flux<? extends ConsumerRecord<Integer, String>> kafkaFlux =
-                KafkaReceiver.create(receiverOptions)
+                receiver
                         .receive()
                         .delayUntil(r -> r.receiverOffset().commit())
                         .doOnSubscribe(s -> assignSemaphore.release());
         sendReceiveWithSendDelay(kafkaFlux, Duration.ofMillis(1000), 0, 10);
+        OptionalLong lag = receiver
+                .doOnConsumer(consumer -> consumer.currentLag(getTopicPartitions().iterator().next()))
+                .block(Duration.ofSeconds(10));
+        assertThat(lag.isPresent()).isTrue();
     }
 
     @Test