Dependencies are not wired into custom KafkaMessageChannelBinder with non-empty properties

[ilia1243]

Describe the issue

According to documentation https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream-binder-kafka.html#rebalance-listener

if you provide a single KafkaBindingRebalanceListener bean in the application context, it will be wired into all Kafka consumer bindings.

However, it is not wired into custom binder with non-empty properties.

To Reproduce

application.yml:

spring:
  cloud:
    function:
      definition: testConsumer
    stream:
      kafka:
        binder:
          brokers: localhost:9092
      binders:
        test-binder:
          type: 'kafka'
          environment:
            spring.cloud.stream.kafka.binder:
              consumer-properties:
                max.poll.records: 1            # < at least one property here 
      bindings:
        testConsumer-in-0:
          binder: test-binder
          group: test-group
          destination: test-topic

Kotlin reproducer:

@SpringBootApplication
class Application {
    @Bean
    fun testConsumer() = java.util.function.Consumer<GenericMessage<ByteArray>> {
    }
}

@Component
class RebalanceListener : KafkaBindingRebalanceListener {
    override fun onPartitionsAssigned(bindingName: String, consumer: Consumer<*, *>, partitions: Collection<TopicPartition?>?, initial: Boolean) {
        // In successful case this should be printed
        println("RebalanceListener is wired and called")
    }
}

fun main() {
    runApplication<Application>()
}

Dependencies:

plugins {
    kotlin("jvm") version "2.2.20"
    kotlin("plugin.spring") version "2.2.20"
}

repositories {
    mavenCentral()
}

dependencies {
    implementation(platform("org.springframework.boot:spring-boot-dependencies:4.0.0-M3"))
    implementation(platform("org.springframework.cloud:spring-cloud-dependencies:2025.1.0-M3"))
    implementation("org.springframework.cloud:spring-cloud-starter-stream-kafka")
}

If replace in application.yml

consumer-properties:
  max.poll.records: 1

to

consumer-properties: {}

the problem does not reproduce, and KafkaBindingRebalanceListener is wired and called.

This may be related to the way the child application context is created https://github.com/spring-cloud/spring-cloud-stream/blob/v5.0.0-M3/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java#L477

    ConfigurableApplicationContext initializeBinderContextSimple(String configurationName, Map<String, Object> binderProperties,
            BinderType binderType, BinderConfiguration binderConfiguration, boolean refresh) {
        ...
        boolean useApplicationContextAsParent = binderProperties.isEmpty()
                && this.context != null;
        ...
        if (useApplicationContextAsParent) {
            binderProducingContext.setParent(this.context);
        }

If properties are not empty, the child context has no reference to the parent one, while KafkaBindingRebalanceListener is defined in the parent context.

Version of the framework: 2025.1.0-M3
Expected behavior: KafkaBindingRebalanceListener should be wired to all binders.

[olegz]

I was able to reproduce, so that is a good thing. I am also trying to figure out why we have that conditional in the first place. On the surface it doesn't make any sense.
Let me sleep on it and see what's going on, but for now I am making it a bug

[olegz]

Ok, i got to the bottom of it. Here is the explanation and the fix is coming shortly.

The empty configuration effectively implies that there is nothing different between the binder and the parent application context. In other words we can inherit whatever is in gthe parent application context into the binder context, hence parent/child relationship,

Now imaging you have two Kafka binders (multi-binder scenario). One for input and one for output. Setting up parent/child with different environment and configuration will cause chaos, so we're not doing that. However, in this case we still recognise that certain beans can be shared and we provide a special hook for that in the way of declaring shared.beans - https://github.com/spring-cloud/spring-cloud-stream/blob/main/core/spring-cloud-stream/src/main/resources/META-INF/shared.beans, which identifies which components from the parent context must be made visible by the binder's context.
You can declare that file yourself in your application'sMETA-INF/shared.beans file and simply add org.springframework.cloud.stream.binder.kafka.KafkaBindingRebalanceListener. But i will add it permanently as part of the fix so you can eventually remove it once the next release is out. I'll back-port it as well

[ilia1243]

Thank you. Is this bean the only missed one, or there are other beans that could be shared? https://github.com/spring-cloud/spring-cloud-stream/blob/v5.0.0-M3/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java#L132

[olegz]

As I mentioned, you can do it yourself, but simply declaring META-INF/shared.beans in your application and add what you need. You can also shared it with us where we can determine if such component has to be shared at the framework level, at which point we'll add it to the framework's shared.beans file.