[
https://issues.apache.org/jira/browse/CAMEL-23997?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Federico Mariani updated CAMEL-23997:
-------------------------------------
Attachment: camel-kafka-review-2026-07.md
> 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
> Priority: Major
> Attachments: camel-kafka-review-2026-07.md
>
>
> 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)
