How to pass BinderCustomizer to Spring cloud stream's BinderConfig?

Question

I am trying to use the BinderCustomizer as suggested on spring cloud stream documentation. I have multiple binders, both of which are KafkaMessageChannelBinder.

My code for BinderCustomizer :

  public BinderCustomizer binderCustomizer() {
    return (binder, binderName) -> {
      if (binder instanceof KafkaMessageChannelBinder) {
        ((KafkaMessageChannelBinder) binder).setRebalanceListener();
      }
    };
  }

But failing with a compatibility error.

Inconvertible types; cannot cast 'org.springframework.cloud.stream.binder.Binder<capture<?>,org.springframework.cloud.stream.binder.ConsumerProperties,org.springframework.cloud.stream.binder.ProducerProperties>' to 'org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder'

I am using spring-cloud-stream:3.2.1.

Is there any alternate way to pass the "BinderCustomizer". Please advise

Answer 1

It looks like it is unable to cast the Binder<? ,? ?> to KafkaMessageChannelBinder. We need to fix it. In the meantime, can you try the following workaround? Basically, we need to cast Binder to AbstractBinder first, then to KafkaMessageChannelBinder. I created an issue to fix this: https://github.com/spring-cloud/spring-cloud-stream/issues/2863

public BinderCustomizer binderCustomizer() {
    return (binder, binderName) -> {
      AbstractBinder aBinder = (AbstractBinder)binder;
      if (aBinder instanceof KafkaMessageChannelBinder) {
        ((KafkaMessageChannelBinder) binder).setRebalanceListener();
      }
    };
  }