[jira] [Updated] (CAMEL-24778) camel-kafka - consumer with batching=true throws NullPointerException when maxPollRecords is not set

["Claus Ibsen (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-24778?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Claus Ibsen updated CAMEL-24778:
--------------------------------
    Fix Version/s: 4.23.0

> camel-kafka - consumer with batching=true throws NullPointerException when maxPollRecords is not set
> ----------------------------------------------------------------------------------------------------
>
>                 Key: CAMEL-24778
>                 URL: https://issues.apache.org/jira/browse/CAMEL-24778
>             Project: Camel
>          Issue Type: Bug
>          Components: camel-kafka
>            Reporter: Andrea Cosentino
>            Assignee: Andrea Cosentino
>            Priority: Major
>             Fix For: 4.23.0
>
>
> When the Kafka consumer is configured with {{batching=true}} but {{maxPollRecords}} is not explicitly set, the consumer thread fails with a NullPointerException as soon as the batching processor is constructed (on the first poll / reconnect).
> h3. Root cause
> {{KafkaConfiguration.maxPollRecords}} is an {{Integer}} field with *no initializer* (null by default). The {{@UriParam(defaultValue = "500")}} on it is only catalog/documentation metadata and is NOT applied to the field at runtime.
> * In *streaming* mode this is harmless: the value is only pushed to Kafka via {{addPropertyIfNotEmpty(props, MAX_POLL_RECORDS_CONFIG, getMaxPollRecords())}} (KafkaConfiguration.java:583), which skips null and lets the Kafka client fall back to its own default of 500.
> * In *batching* mode the Camel code reads the value directly:
> ** {{KafkaRecordBatchingProcessor.java:119}} -> {{new ArrayBlockingQueue<>(configuration.getMaxPollRecords())}} unboxes null -> NPE at construction.
> ** {{KafkaRecordBatchingProcessor.java:173}} -> {{exchangeList.size() >= configuration.getMaxPollRecords()}} would also NPE.
> h3. Impact
> Enabling the documented, supported {{batching}} feature out of the box crashes the consumer thread. The docs/catalog also mislead by advertising {{defaultValue=500}} while the field is actually null at runtime.
> h3. Proposed fix
> Initialize the field to 500 ({{private Integer maxPollRecords = 500;}}), matching the sibling {{sessionTimeoutMs}}/{{pollTimeoutMs}} fields which are initialized. This makes the documented default real and does not change streaming behaviour (Kafka's own default is also 500). Add a regression test that starts a batching consumer without setting {{maxPollRecords}}.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)