Revert "Fix NPE when only batchHandler is specified (#281)" (#282)

This reverts commit 32e3acb.
vert-x3 · Jan 3, 2025 · 4413c1e · 4413c1e
1 parent 32e3acb
commit 4413c1e
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 10 deletions.
diff --git a/src/main/java/io/vertx/kafka/client/consumer/impl/KafkaReadStreamImpl.java b/src/main/java/io/vertx/kafka/client/consumer/impl/KafkaReadStreamImpl.java
@@ -204,10 +204,7 @@ private void run(Handler<ConsumerRecord<K, V>> handler, Handler<ConsumerRecords<
       this.pollRecords(records -> {
 
         if (records != null && records.count() > 0) {
-          if (handler != null) {
-            // only set iterator if records are going to be consumed by individual record handler
-            this.current = records.iterator();
-          }
+          this.current = records.iterator();
           if (multiHandler != null) {
             multiHandler.handle(records);
           }

diff --git a/src/test/java/io/vertx/kafka/client/tests/ConsumerTestBase.java b/src/test/java/io/vertx/kafka/client/tests/ConsumerTestBase.java
@@ -1332,8 +1332,7 @@ public void testConsumerBatchHandler(TestContext ctx) throws Exception {
     String consumerId = topicName;
     Async batch1 = ctx.async();
     AtomicInteger index = new AtomicInteger();
-    int batchSize = 500;
-    int numMessages = 1000;
+    int numMessages = 500;
     kafkaCluster.useTo().produceStrings(numMessages, batch1::complete, () ->
       new ProducerRecord<>(topicName, 0, "key-" + index.get(), "value-" + index.getAndIncrement()));
     batch1.awaitSuccess(10000);
@@ -1347,7 +1346,7 @@ public void testConsumerBatchHandler(TestContext ctx) throws Exception {
     Async batchHandler = ctx.async();
     batchHandler.handler(ar -> wrappedConsumer.close());
     wrappedConsumer.batchHandler(records -> {
-      ctx.assertEquals(batchSize, records.size());
+      ctx.assertEquals(numMessages, records.size());
       for (int i = 0; i < records.size(); i++) {
         KafkaConsumerRecord<Object, Object> record = records.recordAt(i);
         int dec = count.decrementAndGet();
@@ -1356,10 +1355,8 @@ public void testConsumerBatchHandler(TestContext ctx) throws Exception {
         } else {
           ctx.assertEquals("key-" + (-1 - dec), record.key());
         }
-        if (dec == 0) {
-          batchHandler.complete();
-        }
       }
+      batchHandler.complete();
     });
     wrappedConsumer.subscribe(Collections.singleton(topicName));
   }