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
* When composed functions are used on component functions in the Kafka Streams binder,
there is an issue in which the first function in the composition is invoked twice.
Fixing this issue by ensuring that the function execution path is only invoked once.
Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2941
* There seems to be a race condition in ReactorKafkaBinderTests that causes
tests to fail on CI occasionaly. Trying to address this by single dedicated
topic per test.
* Currently, EmbeddedKafka is initialized as part of the class initialization in a few tests
in the Kafka binder. This is preventing individual JUnit tests from being executed from an IDE (IntelliJ, for example).
If we move this initialization to the JUnit method, then that seems to be working.
Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2943
* Currently, EmbeddedKafka is initialized as part of the class initialization.
This is preventing individual JUnit tests from being executed from an IDE (IntelliJ, for example).
If we move this initialization to the JUnit `BeforeAll` method, then that seems to be working.
Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2942
Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2932
Spring Boot provides `SanitizingFunction` to allow the applicaitons to clear out
sensitive data when using certain actuator endpoints. This feature can be
extended to custom endpoints as well. Enable the bindings actuator endpoint
to sanitze sensitive data based on user-provided logic in `SantizingFuction`
beans in the application.
Spring Pulsar 1.1.x introduced a new method `getPulsarClient`
on `PulsarProducerFactory` and the PulsarBinderIntegrationTests
has a custom factory that must be updated in order to run
on both Pulsar 1.0.x and 1.1.x (i.e. Spring Boot 3.2.x and 3.3.x).
