apache / apache/camel-kamelets

Kafka kamelets should use kamelet properties instead of leaking Camel internal headers

Closed
#2,871 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
71
Forks
88
Avg merge
1d 13h
Merged PRs (30d)
70

Description

## Summary

The Kafka kamelets currently rely on Camel-internal headers (`CamelKafkaTopic`, `CamelKafkaKey`, `CamelKafkaPartition`, `CamelKafkaTimestamp`) being forwarded across transport boundaries (e.g. HTTP → Kafka). This is a design issue:

1. **Header leaking**: Previously, old-style `kafka.TOPIC` / `kafka.KEY` headers could leak through HTTP because they weren't filtered by the HTTP component's `DefaultHeaderFilterStrategy`. With the migration to `CamelKafka*` names, the HTTP component now correctly filters them — but this broke tests that relied on the leak.

2. **Wrong abstraction level**: If a kamelet needs to let users dynamically control the Kafka topic, key, or partition, that should be exposed as a kamelet-level property (e.g. `kafkaTopic`) declared in `spec.definition.properties`, not by requiring the caller to set a Camel-internal header directly. The kamelet template should map the property to the internal header.

## Current behavior

- `kafka-sink` expects the caller to set `CamelKafkaKey` / `CamelKafkaPartitionKey` headers directly (or via `key` / `ce-key` CloudEvents headers that the template maps)
- `kafka-source` exposes `CamelKafkaTopic` / `CamelKafkaKey` / `CamelKafkaPartition` as outgoing headers that leak to downstream consumers
- `timestamp-router-action` and `message-timestamp-router-action` read `CamelKafkaTopic` / `CamelKafkaTimestamp` from the exchange — the caller must set these as Camel-internal headers
- Tests relied on `kafka.TOPIC` headers passing through HTTP, which was a leak, not a feature

## Proposed fix

- Add kamelet-level properties (e.g. `kafkaTopic`, `kafkaKey`, `kafkaPartition`) to the relevant kamelets
- The kamelet template maps these properties to the internal `CamelKafka*` headers
- For source kamelets, map `CamelKafka*` headers to non-internal output headers (e.g. `kafka-topic`, `kafka-key`) so downstream consumers can access them without depending on Camel internals
- Update tests to use the kamelet properties / non-internal headers
- Update the `timestamp-router-action` to accept a kamelet property for the topic instead of reading a Camel-internal header directly

This aligns with the kamelet design principle that the template is the trust boundary — callers should interact through declared properties, not by manipulating Camel-internal headers.

## Related

- #2869 — optional parameter endpoint creation failure
- #2868 — integration test failures (header migration)

Contributor guide

No contributing guide indexed for this repository

Research direction

Start by inspecting the kafka-sink, kafka-source, timestamp-router-action, and message-timestamp-router-action kamelet templates, especially their spec.definition.properties and existing tests. Trace how CamelKafka* headers are currently read or emitted, then update the tests to cover declared kamelet properties and non-internal source headers. Done means callers no longer need Camel-internal headers and the related HTTP-based tests no longer depend on header leakage.

Written by the indexing model from the issue text.

Assessment

Tech stack
kafka
Domain
backend, distributed-systems
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
45/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.