1230fahid opened a new pull request, #22870: URL: https://github.com/apache/kafka/pull/22870
## Summary
This PR adds a new built-in Single Message Transformation, `HeaderRouter`,
that routes records to a destination topic based on the value of a Kafka record
header. This addresses a gap in Kafka Connect's built-in transforms where no
native mechanism existed to route records using header metadata.
The transform accepts an ordered list of header names (`header.names`). On
each record, headers are checked in priority order and the value of the first
matching header is used as the destination topic name.
An optional `fallback.topic` can be configured for records where none of the
specified headers are present; if omitted, such records pass through with their
original topic unchanged.
## Changes
- **`connect/transforms/.../HeaderRouter.java`**: New SMT implementation.
Iterates `header.names` in priority order, resolves header values as `String`,
`byte[]` (decoded as UTF-8), or any other type via `toString()`. Skips headers
with `null` or empty string values. Normalizes a blank `fallback.topic` to
absent at `configure()` time to prevent routing records to an invalid empty
topic name.
-
**`connect/transforms/.../META-INF/services/org.apache.kafka.connect.transforms.Transformation`**:
Registers `HeaderRouter` for automatic plugin discovery by the Connect
runtime.
- **`connect/runtime/.../tools/TransformationDoc.java`**: Adds
`HeaderRouter` to the doc generation tool so it appears in the generated HTML
reference documentation (`docs/generated/connect_transforms.html`).
## Testing
Unit tests added in `HeaderRouterTest.java` covering:
- Single header match routes record to header value as topic
- Priority ordering: first header in list wins when multiple are present
- Priority ordering: falls through to second header when first is absent
- `fallback.topic` used when no configured header is present on the record
- Pass-through when no header matches and no fallback is configured
- `null` header value is skipped; next header in list is tried
- Empty string header value is skipped; next header in list is tried
- Blank `fallback.topic` config is treated as absent (pass-through)
- `byte[]` header values are decoded as UTF-8 for the topic name
- Duplicate header keys: last value wins, per the Connect API contract
- Record headers are preserved unchanged after rerouting
- `ConfigException` thrown when `header.names` is missing or empty
--
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]
