Fixes: https://github.com/spring-projects/spring-kafka/issues/3542
* Providing the ability to add record interceptors instead of overriding them
Change `RecordInterceptor` to `List<RecordInterceptor>` in
`MessageListenerContainer` will allow the addition of multiple `RecordInterceptor`
instances instead of overriding the existing one
Currently, only a single `RecordInterceptor` is supported.
Users may want to register multiple `RecordInterceptors`.
There are some workarounds, but they are not clean or ideal solutions.
By supporting `List<RecordInterceptor`>, users can add their own
interceptors via `setRecordInterceptor(...)`.
* Adding new API for addRecordInterceptor.
* Addressing PR review
Signed-off-by: Sanghyeok An <ojt90902@naver.com>
Fixes: https://github.com/spring-projects/spring-kafka/pull/3950
Fix trace context loss in async Kafka error handling
Problem
When async returns are enabled and a consumer failure occurs, the trace context from the original message is not propagated. This leads to each step of the retry/DLT flow starting a new trace instead of continuing the original one.
Example (current behavior):
• Producer → trace 1
• Consumer → trace 1, fails → message goes to retry topic
• Retry listener → trace 2, fails → message goes to DLT topic
• DLT listener → trace 3
This breaks end-to-end traceability, as each listener receives a new trace ID.
Root cause
The issue stems from the `handleAsyncFailure` method, which runs in a different thread but does not propagate the original Observation (trace) context associated with the failed record.
Fix
Ensure that the observation context is correctly propagated when handling async failures. This preserves the trace ID across retry and DLT flows.
**Auto-cherry-pick to `3.3.x`**
Signed-off-by: Igor Macedo Quintanilha <igor.quintanilha@teya.com>
Co-authored-with: Artem Bilan <artem.bilan@broadcom.com>
- New AbstractShareKafkaMessageListenerContainer base class with lifecycle management
- ShareKafkaMessageListenerContainer implementation for share consumer protocol
- Integration tests for end-to-end message delivery validation
Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>
* Remove reactor-kafka based components - ReactiveKafkaConsumerTemplate and
ReactiveKafkaProducerTemplate - and the associated test classes and packages,
in 4.0.0 version.
* Remove the optional dependency on reactor-kafka from the build
See this blog for more details:
https://spring.io/blog/2025/05/20/reactor-kafka-discontinued
Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>
* Deprecate `KafkaTestUtils.consumerProps` methods based on the `String` for `autoCommit`
* Introduce `KafkaTestUtils.consumerProps` based on the `boolean` for `autoCommit`
* Fix respective deprecation warnings
Signed-off-by: Mikhail Polivakha <mikhailpolivakha@gmail.com>
* Add and refactor integration tests in DefaultShareConsumerFactoryTests
- Added tests to check that multiple shared consumers each get records and all data is consumed.
- Refactored shared test logic into a helper method.
- Other cleanup in the test
Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>
Fixes: #3935
Issue link: https://github.com/spring-projects/spring-kafka/issues/3935
Replaces the Error Prone and NullAway Gradle configuration and dependencies
with the `io.spring.nullability` plugin to align null-safety checks with
Spring Framework's nullability standards.
* Removed `net.ltgt.errorprone` plugin
* Removed ErrorProne and NullAway dependencies from the build configuration
* Applied `io.spring.nullability` plugin
Signed-off-by: Hyunsang Han <gustkd3@gmail.com>
These tests were just verifying Kafka’s own rebalance protocol, not anything specific
in Spring for Apache Kafka framework. Since the new protocol is handled entirely by the
Kafka server, there’s no need to test it here. Users can enable the protocol via the
config - group.protocol. No extra framework logic involved.
Testing the protocol in the framework test suite is unnecessary and leads to long-running
tests without added value.
See the ref docs here:
https://docs.spring.io/spring-kafka/reference/4.0/kafka/receiving-messages/rebalance-listeners.html#new-rebalalcne-protocol
Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>
* Introduce share consumer factories for Kafka Queues (Early Access)
- Preliminary set of changes to support Kafka queueus introduced via KIP-932 (Kafka Queue)
for early access in Apache Kafka 4.0.0.
See: https://cwiki.apache.org/confluence/display/KAFKA/KIP-932%3A+Queues+for+Kafka
- Add ShareConsumerFactory interface and DefaultShareConsumerFactory implementation as the
initial building blocks for supporting Kafka Queues (KIP-932) in Spring for Apache Kafka 4.0.x.
This factory and the implementation provide a flexible API for creating share consumers,
and are designed as the foundation for further queue integration.
- Tests to verify the share consumer behavior
Related to #3875https://github.com/spring-projects/spring-kafka/issues/3875
Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>
Integration tests with an EmbeddedKafka instance are sometimes run in parallel in Junit.
The current example is not thread safe, so I think the oficial documentation should account for that.
**Auto-cherry-pick to `3.3.x` & `3.2.x`**
Signed-off-by: mipo256 <mikhailpolivakha@gmail.com>
* Fix docs on `RetryableTopic.replicationFactor()`
* Rephrase replicationFactor description about broker version
Signed-off-by: Seonghyeon Cho <seonghyeoncho96@gmail.com>
**Auto-cherry-pick to `3.3.x` & `3.2.x`**
* Replace `FixedDelayStrategy` to `SameIntervalTopicReuseStrategy` in docs
* `useSingleTopicForFixedDelays` -> `useSingleTopicForSameIntervals`
Signed-off-by: Seonghyeon Cho <seonghyeoncho96@gmail.com>
**Auto-cherry-pick to `3.3.x` & `3.2.x`**
Fixes: #3880
Add an documentation on how to correctly inject headers in default `@KafkaHandler`.
Because of arguments resolution limitations in Spring Messaging, the `@Header` cannot be used on the default `@KafkaHandler`.
Instead the more general `@Headers Map<String, Object>` has to be used.
To satisfy default `@KafkaHandler` method expectations (usually `Object payload`), the `HandlerAdapter` injects a payload into `providedArgs`.
The type of that payload may clash with expected type of the `@Header` argument.
See `InvocableHandlerMethod.getMethodArgumentValues()` for details how params are resolved.
Signed-off-by: Sanghyeok An <ojt90902@naver.com>
Co-authored-by: Artem Bilan <artem.bilan@broadcom.com>
Fixes: #3885
PR link: https://github.com/spring-projects/spring-kafka/pull/3885
The `ListenerUtils.conditionalSleep()` has a loop.
And if the whole interval is longer than consumer poll interval, the consumer is going to be dropped from the group.
* Add a new `ListenerUtils.conditionalSleepWithPoll()` with a `consumer.poll(Duration.ZERO)` in between sleep loop iterations.
Signed-off-by: Sanghyeok An <ojt90902@naver.com>
**Auto-cherry-pick to `3.3.x` & `3.2.x`**
Signed-off-by: Artem Bilan <artem.bilan@broadcom.com>
