@EnableKafkaStreams activates Spring support for Kafka Streams applications. Applied to a configuration class, it adds a Spring-managed wrapper around KafkaStreams and controls application startup and shutdown. section 12.3
A working configuration typically includes:
- Place @EnableKafkaStreams on the application configuration alongside the relevant Spring annotations. section 12.3
- Provide a bean named defaultKafkaStreamsConfig, using KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME, to supply stream properties. section 12.3
- Configure values such as the application ID and bootstrap servers through KafkaStreamsConfiguration. section 12.3
- Build the topology with a StreamsBuilder; Spring creates the StreamsBuilderFactoryBean and manages the Kafka Streams lifecycle. section 12.3
Loan-processing application
A loan application can replace listener and KafkaTemplate-based processing with a Kafka Streams topology. The topology reads loan applications, calculates debt ratios, and supports aggregation of approvals and rejections in one stream-processing application. section 12.3
Listing 12.17 Adding annotation to the configuration class
@SpringBootApplication(scanBasePackages = {"bbejeck.spring.datagen",
"bbejeck.spring.streams.boot"}) #1
@EnableKafka #2
@EnableKafkaStreams #3
@Configuration #4
public class KafkaStreamsLoanApplicationApplication {
//Configuration items left out for clarity
}
Use the configuration-class listing to see @EnableKafkaStreams positioned with @SpringBootApplication, @EnableKafka, and @Configuration. section 12.3
For deeper control, register a KafkaStreamsCustomizer or StreamsBuilderFactoryBeanCustomizer to configure the underlying KafkaStreams instance, such as adding a StateListener. Alternatively, use Spring only for dependency injection and construct, start, and close KafkaStreams yourself. section 12.3 section 12.4
