SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2601: Kafka Streams Binder Docs Cleanup

Browse Source 
Remove docs references to the deprecated/removed StreamsBuilderFactoryBeanCustomizer
in Spring for Apache Kafka in favor of StreamsBuilderFactoryBeanConfigurer.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2601
This commit is contained in:
Soby Chacko
2023-09-11 17:18:42 -04:00
parent 5f266ce580
commit ca58e13fc7
1 changed files with 12 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
docs/src/main/asciidoc/kafka/kafka-streams.adoc
View File

@@ -1257,35 +1257,33 @@ spring.cloud.stream.kafka.streams.bindings.process-out-0.producer.streamPartitio
Each output topic in the application needs to be configured separately like this.
=== StreamsBuilderFactoryBean customizer
=== StreamsBuilderFactoryBean Additional Customizations
It is often required to customize the `StreamsBuilderFactoryBean` that creates the `KafkaStreams` objects.
Based on the underlying support provided by Spring Kafka, the binder allows you to customize the `StreamsBuilderFactoryBean`.
You can use the `StreamsBuilderFactoryBeanCustomizer` to customize the `StreamsBuilderFactoryBean` itself.
Then, once you get access to the `StreamsBuilderFactoryBean` through this customizer, you can customize the corresponding `KafkaStreams` using `KafkaStreamsCustomzier`.
Both of these customizers are part of the Spring for Apache Kafka project.
You can use the `org.springframework.kafka.config.StreamsBuilderFactoryBeanConfigurer` from Spring for Apache Kafka project to customize/configure the `StreamsBuilderFactoryBean` itself.
Here is an example of using the `StreamsBuilderFactoryBeanCustomizer`.
Here is an example of using the `StreamsBuilderFactoryBeanConfigurer`.
```
@Bean
public StreamsBuilderFactoryBeanCustomizer streamsBuilderFactoryBeanCustomizer() {
public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() {
return sfb -> sfb.setStateListener((newState, oldState) -> {
//Do some action here!
});
}
```
The above is shown as an illustration of the things you can do to customize the `StreamsBuilderFactoryBean`.
You can essentially call any available mutation operations from `StreamsBuilderFactoryBean` to customize it.
This customizer will be invoked by the binder right before the factory bean is started.
The above is shown as an illustration of the things you can do to configure the `StreamsBuilderFactoryBean`.
You can essentially call any available mutation operations from `StreamsBuilderFactoryBean` to configure it.
This configurer will be invoked by the binder right before the factory bean is started.
Once you get access to the `StreamsBuilderFactoryBean`, you can also customize the underlying `KafkaStreams` object.
Once you get access to the `StreamsBuilderFactoryBean`, you can also customize the underlying `KafkaStreams` object via the `KafkaStreamsCustomizer`.
Here is a blueprint for doing so.
```
@Bean
public StreamsBuilderFactoryBeanCustomizer streamsBuilderFactoryBeanCustomizer() {
public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() {
return factoryBean -> {
factoryBean.setKafkaStreamsCustomizer(new KafkaStreamsCustomizer() {
@Override
@@ -1299,9 +1297,9 @@ public StreamsBuilderFactoryBeanCustomizer streamsBuilderFactoryBeanCustomizer()
}
```
`KafkaStreamsCustomizer` will be called by the `StreamsBuilderFactoryBeabn` right before the underlying `KafkaStreams` gets started.
`KafkaStreamsCustomizer` will be called by the `StreamsBuilderFactoryBean` right before the underlying `KafkaStreams` gets started.
There can only be one `StreamsBuilderFactoryBeanCustomizer` in the entire application.
There can only be one `StreamsBuilderFactoryBeanConfigurer` in the entire application.
Then how do we account for multiple Kafka Streams processors as each of them are backed up by individual `StreamsBuilderFactoryBean` objects?
In that case, if the customization needs to be different for those processors, then the application needs to apply some filter based on the application ID.
@@ -1309,7 +1307,7 @@ For e.g,
```
@Bean
public StreamsBuilderFactoryBeanCustomizer streamsBuilderFactoryBeanCustomizer() {
public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() {
return factoryBean -> {
if (factoryBean.getStreamsConfiguration().getProperty(StreamsConfig.APPLICATION_ID_CONFIG)