oscerd opened a new pull request, #3032: URL: https://github.com/apache/camel-kamelets/pull/3032
Closes the one part of #2871 that was still outstanding. ## Why this is the only part left Most of #2871 was overtaken by #2868, which landed after it was filed: | #2871 proposed | state on `main` | |---|---| | `timestamp-router-action` should take a property for the topic rather than read a Camel internal header | done — `topicHeaderName` / `timestampHeaderName` (622edd686) | | Callers should not have to set `CamelKafka*` on sinks | done — `kafka-sink` maps `key` / `ce-key` → `CamelKafkaKey`, `partition-key` / `ce-partitionkey` → `CamelKafkaPartitionKey` | | The `kafka.TOPIC` through HTTP leak in tests | done (2e373d5d6, 49d30bc75) | | **Sources should map `CamelKafka*` to non-internal output headers** | **nothing done — this PR** | I did not implement the issue's other suggestion, exposing `kafkaTopic` / `kafkaKey` / `kafkaPartition` as Kamelet **properties**. A Kamelet property binds once when the template is instantiated, so it cannot carry a per-message topic, key or partition, which is exactly what those values are. The friendly-header indirection `kafka-sink` already uses is the abstraction that works for per-message values, and it is already in place. Reasoning posted on the issue. ## What this adds ``` kafka-topic ce-kafkatopic from CamelKafkaTopic kafka-key ce-kafkakey from CamelKafkaKey kafka-partition ce-kafkapartition from CamelKafkaPartition kafka-offset ce-kafkaoffset from CamelKafkaOffset kafka-timestamp ce-kafkatimestamp from CamelKafkaTimestamp ``` Plain name plus a CloudEvents one, which is how this catalog names source output headers — `infinispan-source` and `ftp-source` are the pattern followed here. The names were taken from the `camel-kafka` catalog in 4.23.0-SNAPSHOT rather than assumed. Only `consumer` group headers are surfaced; `CamelKafkaHeaders`, `CamelKafkaManualCommit`, `CamelKafkaLastPollRecord` and `CamelKafkaLastRecordBeforeCommit` are flow control rather than record metadata, so they are left alone. **Purely additive.** The `CamelKafka*` headers stay on the exchange, so anything already reading them keeps working. ## Verified `KafkaIT` run locally against a real Redpanda broker, with the new assertions in place: ``` ✔ SUCCESS (52038ms) kafka-sink-pipe-test ✔ SUCCESS (31308ms) kafka-source-pipe-test ✔ SUCCESS (24270ms) kafka-router-pipe-test TOTAL: 3 PASSED: 3 (100.0%) ``` `kafka-source-pipe-test` now asserts the new headers arrive at the HTTP sink — exact values for `kafka-topic` and `ce-kafkatopic`, `@notEmpty()@` for key, partition, offset and timestamp, since those are not fixed. Also probed the missing-header case directly: a record with no key sets `kafka-key` empty rather than failing the exchange. `script/validator` reports no errors, the doc generator produces no diff beyond the partial, and `KameletsCatalogTest` passes (18 tests). ## Note for a separate PR While editing `kafka-source-description.adoc` I found its Authentication and Configuration sections are wrong, independently of this change: they document **User** and **Password** as required properties and claim the Kamelet "requires SASL/PLAIN authentication". The actual properties are `saslUsername` / `saslPassword`, both optional, with `saslAuthType` defaulting to `NONE` — and both usage examples in that file pass `user:` / `password:`, which are not properties at all. I corrected only the Output Format section here, since that is what this change affects. The rest deserves its own fix. --- _Claude Code on behalf of Andrea Cosentino_ -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
