oscerd opened a new issue, #1806:
URL: https://github.com/apache/camel-kafka-connector/issues/1806
## Description
`CamelKafkaConnectMain.Builder.build()` assembles the route in this order
(`CamelKafkaConnectMain.java:366-381`):
```java
ProcessorDefinition<?> rd = from(from);
if (!ObjectHelper.isEmpty(marshallDataFormat)) { rd =
rd.kamelet("ckcMarshal"); }
if (!ObjectHelper.isEmpty(unmarshallDataFormat)) { rd =
rd.kamelet("ckcUnMarshal"); }
if (getContext().getRegistry().lookupByName("aggregate") != null) { rd =
rd.kamelet("ckcAggregator"); }
if (idempotencyEnabled) { rd = rd.kamelet("ckcIdempotent"); }
rd = rd.kamelet("ckcRemoveHeader");
rd.toD(to);
```
`ckcRemoveHeader` — the stage that applies `camel.remove.headers.pattern` —
is appended **last**, so
it runs immediately before the exchange leaves the route.
That is the right position for the **source** direction, where headers are
mapped onto the outgoing
record at the end. The same builder is reused for the **sink** direction,
where headers enter at the
*front*: `CamelSinkTask.put()` maps `CamelHeader.`-prefixed record headers
into the exchange before
the exchange is sent into the route. On the sink path, therefore, the
header-removal stage runs after
every stage that consumes those headers:
- with `camel.idempotency.expression.type=header`, the idempotency key is
`simple("${headers.<X>}")`, evaluated in `ckcIdempotent` — before
`ckcRemoveHeader`;
- a configured `AggregationStrategy` bean sees the un-stripped headers on
both exchanges;
- any `ckcMarshal` / `ckcUnMarshal` data format that consults headers sees
them too.
So on the sink side the documented `camel.remove.headers.pattern` option
cannot actually keep those
headers away from the stages that are most likely to care about them.
## Expected Behavior
On the sink path, `camel.remove.headers.pattern` is applied before the
stages that consume exchange
headers, so that configuring it has the effect its documentation describes
(`docs/modules/ROOT/pages/user-guide/remove-headers.adoc`).
## Actual Behavior
It is applied last, after marshalling, aggregation and idempotency.
## Additional Context
The fix is presumably to insert `ckcRemoveHeader` right after `from(from)`
for the sink direction and
keep it last for the source direction, which needs the builder to know which
direction it is building
(it currently does not).
There is a related gap with no mechanism at all: `removeHeaders` only acts
on message headers, while
`CamelSinkTask.put()` also maps `CamelProperty.`-prefixed record headers
into **exchange properties**
(`camel.map.properties`, default `true`). There is no
`camel.remove.properties.pattern` counterpart,
so the only control over that channel is turning it off wholesale.
--
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]
