Andrea Cosentino created CAMEL-24778:
----------------------------------------
Summary: 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
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)
