[Feature Request] Support loading customzied configuration for kafka binder configuration to kafka binder's context

[yiliuTo]

Describe the issue This issue is to track the question in stack overflow.

I am having a request which needs to implement a BeanPostProcessor for KafkaBinderConfigurationProperties. While if I define this BeanPostProcessor as a static bean in my configuration class, I can see my BPP being loaded only to the application's application context instead of the kafka binder's context, so the BPP cannot take effects. Here my application uses only a single kafka binder.

The way I currently find to make it loaded to the binder's context is to add my configuration class for my BPP as the binder's sources through the property of spring.main.sources. But I am wondering is there any other solutions to make my BPP (or my configuration class for BPP) loaded to the binder's context without setting the configuration sources for the single binder case?

I read parts of the DefaultBinderFactory class, and find that it loads the outer application context as the parent of the binder's context, but for the BeanFactory, it cannot be inherited to the child during the context refreshment.

Here are the links for the source code of my BPP and Configuration. Due to the above reason I choose to make my configuration not a auto-configuration so it currently cannot be loaded into the normal application context as well.

To Reproduce Steps to reproduce the behavior:

1. Create a kafka binder project, e.g., this one.

2. Create a custom BPP class like below:

   public class CustomKafkaBinderConfigurationPropertiesBeanPostProcessor implements BeanPostProcessor {
   	@Override
   	public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
   		if (bean instanceof KafkaBinderConfigurationProperties) {
   			KafkaBinderConfigurationProperties binderConfigurationProperties = (KafkaBinderConfigurationProperties) bean;
   			binderConfigurationProperties.getConfiguration().put("test", "test");
   		}
   		return bean;
   	}
   }
   

3. Create a Configuration for the BPP like below and make sure it can be scanned/loaded to the application context.

   @Configuration
   public class CustomBeanPostProcessorConfiguration {
   
   	@Bean
   	static CustomKafkaBinderConfigurationPropertiesBeanPostProcessor customBeanPostProcessor() {
   		return new CustomKafkaBinderConfigurationPropertiesBeanPostProcessor();
   	}
   }
   

4. The configuration cannot be loaded to the binder's context and thus the BPP cannot take effects unless similar properties being configured.

   spring.cloud.stream.binders.<binder-name>.environment.spring.main.sources
   

Version of the framework: org.springframework.cloud:spring-cloud-starter-stream-kafka:3.2.4 Expected behavior: the configuration can be loaded to the binder's context in the single binder case.