Binding name is null for commaSeparated destination

[skpandey91]

I have below exception while starting an application with spring cloud enabled. (Spring cloud-3.0.1 and sping boot 2.4.6)

":"ERROR","logger":"org.springframework.cloud.stream.binding.BindingService","applicationName":"","message":"Failed to create consumer binding; retrying in 30 seconds","exception":"\norg.springframework.cloud.stream.binder.BinderException: Exception thrown while starting consumer: \n\tat org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:462)\n\tat org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:91)\n\tat org.springframework.cloud.stream.binder.AbstractBinder.bindConsumer(AbstractBinder.java:143)\n\tat org.springframework.cloud.stream.binding.BindingService.lambda$rescheduleConsumerBinding$1(BindingService.java:202)\n\tat org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\tat java.base/java.lang.Thread.run(Thread.java:829)\nCaused by: java.lang.IllegalArgumentException: 'bindingName' cannot be null\n\tat org.springframework.util.Assert.notNull(Assert.java:201)\n\tat org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.setupRebalanceListener(KafkaMessageChannelBinder.java:821)\n\tat org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createConsumerEndpoint(KafkaMessageChannelBinder.java:663)\n\tat org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createConsumerEndpoint(KafkaMessageChannelBinder.java:158)\n\tat org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:408)\n\t... 10 common frames omitted\n\n"}

This happening in case :

1. I have comma separated destinations like below.
   MyChannel: destination: Topic1,Topic2 group: shopping-cart-golden-ms contentType: application/json partitionCount: 1

2. I have a bean implementing KafkaBindingRebalanceListener in the application.

This is happening because below code in KafkaMessageChannelBinder.setupRebalanceListener is removing the binding name after the first iteration from binding-targets. Then next topic in the next iteration from the commaseparated list finds binding null in the mentioned method.
public void setupRebalanceListener(final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties, final ContainerProperties containerProperties) { Assert.isTrue(!((KafkaConsumerProperties)extendedConsumerProperties.getExtension()).isResetOffsets(), "'resetOffsets' cannot be set when a KafkaBindingRebalanceListener is provided"); final String bindingName = (String)bindingNameHolder.get(); bindingNameHolder.remove();

 Please suggest some solution for this

[sobychacko]

@skpandey91 Could you please set up a minimal Spring Boot application with your configuration where we can reproduce the issue and triage it? Thank you!

[skpandey91]

https://github.com/skpandey91/spring-cloud-stream-demo
Please see above example . just run the spring boot application

[sobychacko]

@skpandey91 Thanks for the sample app. We confirmed that this is a bug. Here is a PR addressing this. We will backport to 3.2.x. I noticed that you are using SCSt version 3.1.x. You might want to upgrade to 3.2.x. Also, keep in mind that StreamListener is deprecated and using the functional model is the recommended approach now.