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
The `FunctionConfiguration.sanitize()` use a `rawtypes` trying to hide
generics routines.
Spring Integration 6.4 has introduced a `BaseMessageBuilder` which does not work
with `rawtypes`.
* Fix `FunctionConfiguration.sanitize()` to expose a `<P>` generic argument
to satisfy a new `MessageBuilder` byte code, plus to meet all the expectations
of this method consumers
* In addition perform some code clean to optimize a usage of `MessageBuilder`:
- `setHeader()` does set the value overriding existing one
- `setHeader()` removes the entry if provided value is `null`
- This way we just can go ahead with `MessageBuilder` method chain avoiding extra conditions
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.
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.
