- Replace deprecated ValueTransformerWithKey with FixedKeyProcessor
- Update TimeWindows.of() to TimeWindows.ofSizeWithNoGrace()
- Migrate branch() to split().branch() with Named and Branched
- Replace deprecated KafkaTestUtils.consumerProps() signature
- Update StreamPartitioner return type to Optional<Set<Integer>>
- Modernize Processor API imports and method signatures
- Add EmbeddedKafkaBroker parameter injection for JUnit 5
- Fix Consumer.poll() to use Duration parameter
- Remove unused imports and deprecated API usage
Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>
Fixes: #3090
Issue: https://github.com/spring-cloud/spring-cloud-stream/issues/3090
Fixes issue where KafkaBinderConfiguration would add customizers after the
transaction manager was already created.
The KafkaMessageChannelBinder was initializing the transaction manager in the
constructor before client factory customizers were added, which meant the customizers
were never applied to the producer factory used by the transaction manager.
- Moves transaction manager initialization from the constructor to the onInit() method
- Leverages the existing InitializingBean lifecycle to ensure customizers are applied
before the transaction manager is created
- Adds comprehensive test coverage to verify the fix in the binder and via auto-configuration
Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>
Resolves#3091
- Producer that fetches partition info now initializes within a try-with-resources block
- If exceptions occur on calling producer.partitionsFor(topicName), it's now properly closed and resources are released
Signed-off-by: akenra <37288280+akenra@users.noreply.github.com>
Fixes: https://github.com/spring-cloud/spring-cloud-stream/issues/3062
The `KafkaBinderMetrics` creates `KafkaConsumer` instances and schedule the fix rate
task for them, but never closes them even when the `scheduler` is shut downed
* Implement a `Lifecycle` contract in the `KafkaBinderMetrics` and call `close()`
from the `stop()` to satisfy CRaC resource management expectations.
* Also close all the `KafkaConsumer` instances from the `metadataConsumers`
**Cherry-pick to `4.1.x`**
Fixes: #3040Resolves#3041
This change adds out-of-the-box `ListenerContainerCustomizer<AbstractMessageListenerContainer>`
to set `VirtualThreadTaskExecutor` into an `AbstractMessageListenerContainer` created by the binder
when `Threading.VIRTUAL` condition is met
