(camel) branch main updated: CAMEL-24778: camel-kafka - fix NullPointerException in batching consumer when maxPollRecords is not set

[davsclaus]

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/main by this push:
     new e3cd428670ce CAMEL-24778: camel-kafka - fix NullPointerException in 
batching consumer when maxPollRecords is not set
e3cd428670ce is described below

commit e3cd428670cea9fe15a297068c91383b2daef594
Author: Andrea Cosentino <anco...@gmail.com>
AuthorDate: Fri Sep 18 09:01:25 2026 +0200

    CAMEL-24778: camel-kafka - fix NullPointerException in batching consumer 
when maxPollRecords is not set
    
    KafkaConfiguration.maxPollRecords had no initializer, so the
    @UriParam default of 500 was documentation only. Streaming mode was
    unaffected because addPropertyIfNotEmpty skipped the null and the
    Kafka client applied its own default of 500, but the batching
    consumer reads the value directly: KafkaRecordBatchingProcessor sizes
    its ArrayBlockingQueue from getMaxPollRecords() and unboxed null,
    crashing the consumer thread on the first poll whenever batching=true
    was enabled without an explicit maxPollRecords.
    
    Initialize the field to 500 so the documented default is real. The
    @UriParam metadata is unchanged. A test constructs the batching
    processor with a default configuration, which threw before the fix.
    
    Closes #26521
    
    Co-authored-by: Claude <nore...@anthropic.com>
---
 .../org/apache/camel/component/kafka/KafkaConfiguration.java   |  2 +-
 .../KafkaRecordBatchingProcessorCommonHeadersTest.java         | 10 ++++++++++
 2 files changed, 11 insertions(+), 1 deletion(-)

diff --git 
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
 
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
index de386e795497..15d979ef6455 100755
--- 
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
+++ 
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
@@ -117,7 +117,7 @@ public class KafkaConfiguration implements Cloneable, 
HeaderFilterStrategyAware
     @UriParam(label = "consumer", defaultValue = "45000")
     private Integer sessionTimeoutMs = 45000;
     @UriParam(label = "consumer", defaultValue = "500")
-    private Integer maxPollRecords;
+    private Integer maxPollRecords = 500;
     @UriParam(label = "consumer", defaultValue = "5000", javaType = 
"java.time.Duration")
     private Long pollTimeoutMs = 5000L;
     @UriParam(label = "consumer", javaType = "java.time.Duration")
diff --git 
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/batching/KafkaRecordBatchingProcessorCommonHeadersTest.java
 
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/batching/KafkaRecordBatchingProcessorCommonHeadersTest.java
index 88f0d6553327..b509b7c500db 100644
--- 
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/batching/KafkaRecordBatchingProcessorCommonHeadersTest.java
+++ 
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/batching/KafkaRecordBatchingProcessorCommonHeadersTest.java
@@ -21,6 +21,7 @@ import java.util.List;
 import org.apache.camel.CamelContext;
 import org.apache.camel.Exchange;
 import org.apache.camel.Message;
+import org.apache.camel.component.kafka.KafkaConfiguration;
 import org.apache.camel.component.kafka.KafkaConstants;
 import org.apache.camel.impl.DefaultCamelContext;
 import org.apache.camel.support.DefaultExchange;
@@ -28,6 +29,7 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
 
@@ -113,6 +115,14 @@ class KafkaRecordBatchingProcessorCommonHeadersTest {
         assertNull(batch.getHeader(KafkaConstants.PARTITION));
     }
 
+    @Test
+    void processorConstructsWithDefaultMaxPollRecords() {
+        // CAMEL-24778: with maxPollRecords left unset, the batch buffer was 
sized from a null Integer,
+        // throwing NullPointerException as the processor was constructed. The 
constructor only reads the
+        // configuration, so the processor and commit manager are irrelevant 
to this regression.
+        assertDoesNotThrow(() -> new KafkaRecordBatchingProcessor(new 
KafkaConfiguration(), null, null));
+    }
+
     @Test
     void emptyBatchSetsNoHeaders() {
         Message batch = batchMessageFor(List.of());