[
https://issues.apache.org/jira/browse/CAMEL-24778?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-24778.
---------------------------------
Resolution: Fixed
> 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)
