KafkaTopicProvisioner incompatible with SpringBoot v3.4.0

[lbsekr]

Describe the issue
The provisioning class KafkaTopicProvisioner in springframework.cloud:spring-cloud-stream-binder-kafka-core is incompatible with the current Springboot version 3.4.0.

The KafkaTopicProvisioner class needs to call the createAdminClientProperties method during construction. This calls a previously deprecated method KafkaProperties::(buildAdminProperties) with no arguments.

In Springboot 3.4.0, the arg-less method was removed and replaced with new method signatures that expect a single SslBundles argument. Because of this, KafkaTopicProvisioner is broken with Springboot 3.4.0.

To Reproduce
Steps to reproduce the behavior:

1. Create a Springboot project with version 3.4.0
2. Use org.springframework.cloud:spring-cloud-dependencies with version 2023.0.4 or just springframework.cloud:spring-cloud-stream-binder-kafka-core with version 4.1.4
3. Configure Kafka bindings
4. Run the project
5. 💥

Version of the framework
Springboot at version 3.4.0
springframework.cloud:spring-cloud-stream-binder-kafka-core at version 4.1.4

Expected behavior
The KafkaTopicProvisioner class should call the updated method signatures with default argument null and optionally pass a SslBundles value if configured.

Additional context
See related issues 43300 39144 from the springboot project.

[lbsekr]

Seems to be already implemented in b7f2f1b and included in 4.2.0-MC2. Can the rollout of this fix in particular be expedited please?

[desperateCoder]

Thanks for pointing me to the release candidate! As a big "THANKS", I'll point you to the available version:

https://mvnrepository.com/artifact/org.springframework.cloud/spring-cloud-stream/4.2.0-RC1

Although I use them via:

     <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-dependencies</artifactId>
        <version>2024.0.0-RC1</version>
        <type>pom</type>
        <scope>import</scope>
      </dependency>

in my <dependencyManagement>

Just tried it, I can confirm it's working!

[lbsekr]

Yes, it works for development purposes. But I will not be using release candidates in production. New versions of other libs in the Spring context already require Springboot 3.4.0, which means I basically have to roll back version updates and not update until this is fixed in a stable release.

[voodemsanthosh]

We do see same problem with Kafka, here is my kafka failure logs

"log_message":org.springframework.messaging.MessagingException: 'org.springframework.integration.support.MessageBuilder org.springframework.integration.support.MessageBuilder.removeHeader(java.lang.String)'
at org.springframework.integration.core.ErrorMessagePublisher.determinePayload(ErrorMessagePublisher.java:186)
at org.springframework.integration.core.ErrorMessagePublisher.publish(ErrorMessagePublisher.java:162)
at org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer.recover(ErrorMessageSendingRecoverer.java:83)
at org.springframework.retry.support.RetryTemplate.handleRetryExhausted(RetryTemplate.java:573)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:418)
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:246)
at org.springframework.integration.kafka.inbound.KafkaInboundEndpoint.doWithRetry(KafkaInboundEndpoint.java:70)
at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:457)
at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:422)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:2875)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:2853)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:2766)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:2604)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:2493)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:2144)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeIfHaveRecords(KafkaMessageListenerContainer.java:1520)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1458)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1327)
at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1804)
at java.base/java.lang.Thread.run(Thread.java:1583)
Caused by: java.lang.NoSuchMethodError: 'org.springframework.integration.support.MessageBuilder org.springframework.integration.support.MessageBuilder.removeHeader(java.lang.String)'
at org.springframework.cloud.stream.function.FunctionConfiguration.sanitize(FunctionConfiguration.java:395)
at org.springframework.cloud.stream.function.FunctionConfiguration$FunctionWrapper.apply(FunctionConfiguration.java:817)
at org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder$1.handleMessageInternal(FunctionConfiguration.java:657)
at org.springframework.integration.handler.AbstractMessageHandler.doHandleMessage(AbstractMessageHandler.java:105)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:73)
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:132)
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148)
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121)
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72)
at org.springframework.integration.channel.AbstractMessageChannel.sendInternal(AbstractMessageChannel.java:390)
at org.springframework.integration.channel.AbstractMessageChannel.sendWithMetrics(AbstractMessageChannel.java:361)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:331)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:304)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
at org.springframework.integration.endpoint.MessageProducerSupport.lambda$sendMessage$1(MessageProducerSupport.java:261)
at io.micrometer.observation.Observation.observe(Observation.java:499)
at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:261)
at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.sendMessageIfAny(KafkaMessageDrivenChannelAdapter.java:391)
at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.lambda$onMessage$0(KafkaMessageDrivenChannelAdapter.java:460)
at org.springframework.integration.kafka.inbound.KafkaInboundEndpoint.lambda$doWithRetry$0(KafkaInboundEndpoint.java:77)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:357)
... 15 more