Federico Mariani created CAMEL-23997:
----------------------------------------
Summary: camel-kafka: medium-severity findings from code review
(config, consumer, producer batch mode, transforms)
Key: CAMEL-23997
URL: https://issues.apache.org/jira/browse/CAMEL-23997
Project: Camel
Issue Type: Bug
Components: camel-kafka
Reporter: Federico Mariani
Umbrella ticket for the remaining *medium-severity* findings from a full code
review of camel-kafka (main, 4.22.0-SNAPSHOT). The critical findings are
tracked separately in the linked issues (consumer message loss; saslAuthType;
idempotent/resume). Each item below can be fixed independently; sub-tasks can
be split out as needed.
h3. Configuration
*1. sslEndpointAlgorithm=none does NOT disable hostname verification*
({{KafkaConfiguration}} lines 551-554 producer, 639-642 consumer; from
CAMEL-18146)
When the value is {{none}}/{{false}} the property is simply omitted, but
kafka-clients resolves an absent {{ssl.endpoint.identification.algorithm}} to
its own default {{https}} (verified via {{ConsumerConfig}}). The documented way
to disable verification is an *empty string*. The option is silently
ineffective; users must work around via {{additionalProperties}}. Fix: put an
empty string when {{none}}/{{false}}; consider marking the option {{security =
"insecure:ssl"}} per design/security.adoc.
*2. offsetRepository no longer disables Kafka auto-commit* (line 581;
regression from CAMEL-13870)
The mapping uses {{getAutoCommitEnable()}} (checks only {{batching}}) instead
of the {{offsetRepository}}-aware {{isAutoCommitEnable()}}. With an
{{offsetRepository}} configured and {{autoCommitEnable}} at default true,
{{enable.auto.commit=true}} is passed to the client, whose background thread
commits to the broker in parallel with repository tracking — diverging offsets
on rebalance. The setter javadoc explicitly promises "Defining one will disable
the autocommit". Fix: make {{getAutoCommitEnable()}} also return false when
{{offsetRepository != null}}.
*3. sendBufferBytes only applied to consumers when SSL is enabled* (line ~650;
regression from CAMEL-17615 refactor)
{{ProducerConfig.SEND_BUFFER_CONFIG}} sits inside the SSL-only block of the
consumer path; on PLAINTEXT the option is silently ignored. Fix: move to the
general consumer section (and use the {{ConsumerConfig}} constant).
*4. queueBufferingMaxMessages is a dead option* (lines 183-184)
Declared {{@UriParam(label = "producer", defaultValue = "10000")}} but never
mapped into any Kafka property — it belongs to the old Scala producer removed
in CAMEL-9467. Deprecate/remove with an upgrade-guide entry pointing to
{{bufferMemorySize}}/{{maxBlockMs}}.
h3. Consumer
*5. reconnect flag never reset after a successful reconnect*
({{KafkaFetchRecords}} lines 191, 319-326, 396-402; from CAMEL-17244)
Once {{setReconnect(true)}} (breakOnFirstError path), nothing resets it;
{{initializeConsumer()}} carries the comment "set reconnect to false ..." but
calls {{setConnected(false)}} instead. Consequence: with
{{breakOnFirstError=true}} + {{pollOnError=STOP}} (or a later fatal
{{AuthenticationException}}), the outer loop keeps reconnecting forever because
{{isReconnect()}} stays true. Fix: {{setReconnect(false)}} after successful
re-connection.
*6. Auto-generated groupId is a new random UUID per fetcher task*
({{KafkaConsumer#getProps}}, line 112)
With {{consumersCount=3}} and no {{groupId}}, each task gets its own
{{UUID.randomUUID()}} group → 3 independent groups → every message processed 3
times. Fix: generate once in {{doStart()}} and share.
*7. Manual-commit mode with Sync/Async commit managers auto-commits offsets the
route never committed* ({{KafkaRecordStreamingProcessor}} lines 97-99 + facade
119-123)
{{processExchange}} records every offset (including failed exchanges) and the
facade calls {{commitManager.commit(partition)}} per partition — so with
{{allowManualCommit=true}} + a configured Default*CommitFactory, offsets are
committed even when the route deliberately did not call
{{KafkaManualCommit.commit()}}. Behavior silently differs from the default
({{NoopCommitManager}}) setup. Needs a decision: gate on
{{!isAllowManualCommit()}} or document.
h3. Producer (batch/iterator mode, {{KeyValueHolderIterator}})
*8. Plain list elements are not converted to the serializer type* (lines 78-81;
regression from CAMEL-14360, original behavior from CAMEL-10586)
The non-Exchange/Message branch passes the raw body into the {{ProducerRecord}}
despite the method's own comment. {{List.of(1, 2, 3)}} + default
StringSerializer → {{ClassCastException}} at send time, while the same Integer
as a single body works. Fix: {{tryConvertToSerializedType(exchange, body,
valueSerializer)}} in the plain branch.
*9. Keys of List<Message> elements are never converted* (lines 117-132)
{{getInnerKey}} passes {{innerExchange}} (null for {{Message}} elements) to
{{tryConvertToSerializedType}}, which returns the object unconverted when the
exchange is null — the value conversion three lines later correctly uses the
fallback exchange, the key conversion does not. {{List<Message>}} with Integer
{{CamelKafkaKey}} + StringSerializer → {{ClassCastException}}. Fix: pass the
fallback {{ex}}.
*10. Endpoint-configured key= ignored for batch elements lacking the
CamelKafkaKey header* (lines 117-132 vs 134-140)
{{kafkaConfiguration.getKey()}} is only consulted inside the {{if (innerKey !=
null)}} guard, so elements without the header get a null record key even when
{{key=fixedKey}} is configured — inconsistent with {{getInnerPartitionKey}} and
with single-message mode ({{getOverrideKey}}). Breaks ordering expectations
silently. (Behavior change — note in upgrade guide.)
*11. Per-element recomputation of propagated headers*
({{PropagatedHeadersProvider#getDefaultHeaders}}; perf regression from
CAMEL-18527)
The plain-element branch recomputes the filtered+serialized header list for
every list element even in the default {{batchWithIndividualHeaders=false}}
mode where it was already cached. 10k elements x 20 headers = 200k redundant
filter+serialize calls per exchange. Fix: return the cached list.
*12. DefaultKafkaHeaderSerializer's type-converter fallback (CAMEL-18380) never
activates*
The serializer is {{CamelContextAware}} but nothing in camel-kafka injects the
context into the default instance created in {{KafkaConfiguration}} — so
unsupported header types are silently dropped instead of converted (the unit
test passes only because it injects the context manually). Fix:
{{CamelContextAware.trySetCamelContext(configuration.getHeaderSerializer(),
context)}} in {{KafkaProducer.doStart()}}.
h3. Kamelet transforms
*13. ReplaceField: renames parsing guarded by the wrong variable*
({{ReplaceField.java:49-51}})
{{if (ObjectHelper.isNotEmpty(disabled)) \{ renameFields = renames.split(",")
\}}} — tests {{disabled}} instead of {{renames}} (copy-paste). Used as a plain
bean (the classes were made reusable in CAMEL-21843): {{disabled}} set +
{{renames}} null → NPE; {{renames}} set + {{disabled}} empty → renames (and the
whole enabled/disabled filter loop) silently skipped. The kamelet defaults mask
it.
*14. MessageTimestampRouter: takes the first configured key name instead of the
first field present* ({{MessageTimestampRouter.java:59-64}})
The loop breaks on the first non-empty key *name* without checking the payload
contains it; with {{timestampKeys=ts1,ts2}} and a payload containing only
{{ts2}}, the timestamp is null and the record silently goes to the default
topic instead of the time-bucketed one — contradicting the documented "first
found field". Also line 70 raw-casts the value to String
({{ClassCastException}} for numeric fields).
h3. Reference
The complete review (including ~20 further low-severity items and cleanups:
charset mismatch in header serde, InterruptException consumer leak, dead code,
per-exchange ObjectMapper allocations, mislabeled producer-only options, etc.)
is available as a markdown document attached to / referenced from this review;
low items can be fixed opportunistically or split out on demand.
----
_This issue was researched and written by Claude Code on behalf of Federico
Mariani (GitHub: Croway). Findings were verified against the source and git
history before filing._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
