Skip to content

KAFKA-20709: Add HeaderRouter SMT#22870

Open
1230fahid wants to merge 1 commit into
apache:trunkfrom
1230fahid:fa-kafka-20709-smt
Open

KAFKA-20709: Add HeaderRouter SMT#22870
1230fahid wants to merge 1 commit into
apache:trunkfrom
1230fahid:fa-kafka-20709-smt

Conversation

@1230fahid

Copy link
Copy Markdown

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 commit adds a built-in SMT that routes records to a
destination topic based on the value of a Kafka record header.
Header names are specified in priority order.
The first header found on the record determines the destination
topic. An optional fallback.topic routes unmatched records.
If unset, records pass through with their original topic unchanged.
@github-actions github-actions Bot added triage PRs from the community connect labels Jul 19, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

connect triage PRs from the community

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants