KafkaStreamsCustomizer is a Spring Kafka callback that exposes the KafkaStreams instance before startup, letting an application apply settings such as a StateListener. section 12.3
Use it when the managed Kafka Streams instance must provide runtime visibility:
- Define a KafkaStreamsCustomizer bean and configure the KafkaStreams instance through a Java lambda. section 12.3
- Attach that customizer to the StreamsBuilderFactoryBean with a StreamsBuilderFactoryBeanCustomizer bean. section 12.3
- Register a StateListener that logs active task topic-partition assignments when the instance reaches RUNNING. section 12.3
Observing a running Kafka Streams instance
A KafkaStreamsCustomizer calls setStateListener and checks for KafkaStreams.State.RUNNING; metadataForLocalThreads() then supplies thread names and active tasks for logging. section 12.3
Listing 12.19 Kafka Streams customizer for setting a StateListener
@Bean
KafkaStreamsCustomizer getKafkaStreamsCustomizer() {
return kafkaStreams ->
kafkaStreams.setStateListener((newState, oldState) -> { #1
if (newState == KafkaStreams.State.RUNNING) {
LOG.info("Streams now in running state");
kafkaStreams.metadataForLocalThreads()
.forEach(tm -> LOG.info("{} active task info: {}",
tm.threadName(), tm.activeTasks())); #2
}
});
}
Inspect the complete lambda-based customizer that installs the StateListener. section 12.3
Listing 12.20 Creating a customizer to apply KafkaStreams object settings
@Bean
StreamsBuilderFactoryBeanCustomizer kafkaStreamsCustomizer() { #1
return streamsFactoryBean ->
streamsFactoryBean
.setKafkaStreamsCustomizer(getKafkaStreamsCustomizer()); #2
}
Follow the factory-bean bridge that applies the KafkaStreamsCustomizer. section 12.3
