[BUG] Not working when using Spring Kafka and Kafka Binder via connection string auth in Spring Cloud Azure 5.19.0

[moarychan]

Describe the bug
KafkaAdminClient still using a default local address for connection.

Exception or Stack Trace
Connect to the Event hubs server.

To Reproduce
Use this sample.

Code Snippet
N/A

Expected behavior
Normally connect to Event Hubs via Kafka API.

Screenshots
If applicable, add screenshots to help explain your problem.

Setup (please complete the following information):

• OS: Windows 11
• IDE: Intellij idea
• Java version: Java 21
• Frameworks: Spring Boot 3.3.4 + Spring Cloud 2023.0.3 + Spring Cloud Azure 5.19.0

Additional context
Works for 5.18.0.

Information Checklist
Kindly make sure that you have added all the following information above and checkoff the required fields otherwise we will treat the issuer as an incomplete report

• Bug Description Added
• Repro Steps Added
• Setup information Added

[github-actions]

@moarychan @Netyyyy @rujche @saragluna

[github-actions]

Thank you for your feedback. Tagging and routing to the team member best able to assist.

[moarychan]

The root cause is that the configuration AzureEventHubsKafkaAutoConfiguration is spliced to a another new class KafkaPropertiesConfiguration, the initialization of the new class KafkaPropertiesConfiguration takes precedence over the connection string provider, the workaround for this bug is:
Update configuration for Kafka Binder:

spring.cloud.stream.binders.kafka.environment.spring.main.sources=com.azure.spring.cloud.autoconfigure.implementation.eventhubs.kafka.AzureEventHubsKafkaAutoConfiguration,com.azure.spring.cloud.autoconfigure.implementation.eventhubs.kafka.KafkaPropertiesConfiguration

Re-register the bean KafkaPropertiesBeanPostProcessor for Spring Kafka in the parent application context, since is not public, you may should make a copy then register it as a bean and depends on bean eventHubsKafkaConnectionString; or you can make the eventHubsKafkaConnectionString bean is initialed first before the KafkaPropertiesBeanPostProcessor.