[jira] [Created] (CAMEL-23997) camel-kafka: medium-severity findings from code review (config, consumer, producer batch mode, transforms)

["Federico Mariani (Jira)" <ji...@apache.org>]

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)