- 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
Fixes: https://github.com/spring-cloud/spring-cloud-stream/issues/2939
The RabbitMQ 4.0 does not deal with client side `x-*` headers.
Therefore, an `x-death.count` is not incremented anymore when
message is re-published from client back to the broker.
* Spring AMQP 3.2 has introduced an `AmqpHeaders.RETRY_COUNT` custom header.
Use `messageProperties.incrementRetryCount()` in the `RabbitMessageChannelBinder`
when we re-published message back to the broker for server-side retries
* Fix docs respectively
Resolves#3019
Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2985
This commit introduces new customization options for Kafka listener containers
in Spring Cloud Stream, along with comprehensive documentation:
- Add KafkaListenerContainerCustomizer interface for Kafka-specific customization
with access to extended consumer properties
- Extend ListenerContainerWithDlqAndRetryCustomizer to include access to
extended consumer properties
- Update KafkaMessageChannelBinder to support the new customizer interfaces
- Implement KafkaListenerContainerCustomizerTests for integration testing
- Add detailed AsciiDoc reference documentation explaining the purpose,
usage, and hierarchy of these customizer interfaces:
* ListenerContainerCustomizer (existing)
* KafkaListenerContainerCustomizer (new)
* ListenerContainerWithDlqAndRetryCustomizer (extended)
- Update navigation to include the new documentation
These changes enhance the flexibility and configurability of Kafka consumer
endpoints in Spring Cloud Stream applications, allowing users to fine-tune
their listener containers based on specific requirements and scenarios,
with improved access to Kafka-specific properties.
- 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.
