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.
The getBinder method in DefaultBinderFactory was made thread-safe using ReentrantLock. This commit ensures that the method is friendly for virtual threads to avoid blocking and pinning. The lock is acquired at the beginning of the method and released in a finally block to ensure it is always released, even if an exception occurs. Fixes gh-3004
Fixes https://github.com/spring-cloud/spring-cloud-stream/issues/2650
* Enable native observability support for output binding in the reactive Kafka binder
* Adding test to verify this support with downstream consumers
* Adding ref docs
* Addressing PR review
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
