First, thanks for the excellent StreamBridge feature! I've found it very useful.
However, I occasionally encounter the following exception during race conditions.
```
java.lang.NullPointerException: null
at java.base/java.util.TreeMap.rotateRight(TreeMap.java:2240)
at java.base/java.util.TreeMap.fixAfterInsertion(TreeMap.java:2272)
at java.base/java.util.TreeMap.put(TreeMap.java:580)
at org.springframework.cloud.stream.config.BindingServiceProperties.bindToDefault(BindingServiceProperties.java:397)
at org.springframework.cloud.stream.config.BindingServiceProperties.bindIfNecessary(BindingServiceProperties.java:381)
at org.springframework.cloud.stream.config.BindingServiceProperties.getBindingProperties(BindingServiceProperties.java:301)
at org.springframework.cloud.stream.function.StreamBridge.send(StreamBridge.java:149)
```
BindingServiceProperties.bindings property should be thread-safe.
- Integrate KafkaConnectionDetails, a Spring Boot component, in binder
- Update KafkaBinderConfigurationProperties to use KafkaConnectionDetails
- Modify KafkaTopicProvisioner to leverage KafkaConnectionDetails
- Adjust Kafka binder configurations to pass KafkaConnectionDetails
- Update tests to accommodate KafkaConnectionDetails changes
- Add KafkaConnectionDetails to shared.beans for auto-configuration
This change improves flexibility in configuring Kafka connections,
allowing for better support of externalized configuration management
and aligning with Spring Boot's connection abstraction model.
Related to: https://github.com/spring-projects/spring-integration/issues/9362
After the fix in Spring Integration: bdcb856a90
we ended up in a deadlock situation with a `beginPublishingTrigger` in the `FunctionConfiguration`
used for the `delaySubscription()` on an original `Publisher`.
The `FluxMessageChannel` uses its own `delaySubscription()` until the channel has its subscribers.
Apparently the logic before was with errors, so the `FluxMessageChannel` was marked as active
even if its subscriber is not ready yet, leading to famous `Dispatcher does not have subscribers` error.
So, looks like this `beginPublishingTrigger` was introduced back in days in Spring Cloud Stream
to mitigate that situation until we really emit a `BindingCreatedEvent`.
The deadlock (and the flaw, respectively) is with the `setupBindingTrigger()` method implementation
where `FluxMessageChannel` now "really" delays a subscription to the provided `Publisher`,
therefore not triggering that `Mono.create()` fulfilment immediately.
The `BindingCreatedEvent` arrives earlier, than we have a subscriber on the channel,
but `triggerRef.get()` is `null`, so we don't `success()` it and in the end don't subscribe
to an original `Publisher` since `delaySubscription()` on it is never completed.
Since `FunctionConfiguration` fully relies on `IntegrationFlow.from(Publisher)`,
which ends up with the mentioned `FluxMessageChannel.subscribeTo()` and its own `delaySubscription()`
(which, in turn, apparently fixed now), we don't need our own `delaySubscription()` any more.
Therefore the fix in this PR is to propose to remove `beginPublishingTrigger` logic altogether.
- When concurrent threads publish to a binding, PartitionAwareFunctionWrapper
resets to null between invocations. Addressing this issue by guarding this reset
from occurring if the partiton header on the producer is found.
Fixes https://github.com/spring-cloud/spring-cloud-stream/issues/2961
