TopicNameExtractor is a Kafka Streams interface for choosing a destination topic separately for each record. Pass it to KStream.to() when routing depends on record data instead of a fixed topic. section 6.6.4
Implement the single extract method around the information that controls routing:
- Inspect the key, value, or both; the example reads a purchase department and routes coffee and electronics to matching topics. section 6.6.4
- Return a pre-created topic name for matching records and a fallback such as purchases for everything else. section 6.6.4
- Use RecordContext when routing requires metadata such as the original topic, partition, offset, timestamp, or headers. section 6.6.4
Header-driven routing
A header-based extractor checks RecordContext.headers(), reads the first routing header when present, converts its bytes to a UTF-8 topic name, and otherwise returns a default topic. section 6.6.4
Listing 6.24 Implementing the extract method to determine the topic name
@Override
public String extract(String key,
Purchase value,
RecordContext recordContext) {
String department = value.getDepartment();
if (department.equals("coffee")
|| department.equals("electronics")) { #1
return department;
} else {
return "purchases"; #2
}
}
See a minimal extractor that routes selected departments to their own topics and sends other purchases to purchases. section 6.6.4
Listing 6.25 Using information in a Header to determine the topic name
public String extract(String key,
Purchase value,
RecordContext recordContext) {
Headers headers = recordContext.headers(); #1
if (headers != null) {
Iterator<Header> routingHeaderIterator =
headers.headers("routing").iterator();
if (routingHeaderIterator.hasNext()) {
Header routing = routingHeaderIterator.next(); #2
return new String(routing.value(),
StandardCharsets.UTF_8); #3
}
}
return defaultTopicName; #4
}
See header extraction with null checking, routing-header lookup, UTF-8 conversion, and default-topic fallback. section 6.6.4
