You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "1230fahid (via GitHub)" <gi...@apache.org> on 2026/07/19 03:14:51 UTC
[PR] KAFKA-20709: Add HeaderRouter SMT [kafka]
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20709: Add HeaderRouter SMT [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22870:
URL: https://github.com/apache/kafka/pull/22870#issuecomment-5081975035
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org