diff --git a/docs/en/spring-for-apache-kafka/spring-kafka.md b/docs/en/spring-for-apache-kafka/spring-kafka.md
index 0287046898d38834e59d36c000d406b279d0f6f8..f8f4ad8c9ac116cf3ab384413136953f735f872e 100644
--- a/docs/en/spring-for-apache-kafka/spring-kafka.md
+++ b/docs/en/spring-for-apache-kafka/spring-kafka.md
@@ -1,19 +1,19 @@
# Spring for Apache Kafka
-## [](#preface)1. Preface
+## 1. Preface
The Spring for Apache Kafka project applies core Spring concepts to the development of Kafka-based messaging solutions.
We provide a “template” as a high-level abstraction for sending messages.
We also provide support for Message-driven POJOs.
-## [](#whats-new-part)2. What’s new?
+## 2. What’s new?
-### [](#spring-kafka-intro-new)2.1. What’s New in 2.8 Since 2.7
+### 2.1. What’s New in 2.8 Since 2.7
This section covers the changes made from version 2.7 to version 2.8.
For changes in earlier version, see [[history]](#history).
-#### [](#x28-kafka-client)2.1.1. Kafka Client Version
+#### 2.1.1. Kafka Client Version
This version requires the 3.0.0 `kafka-clients`
@@ -22,7 +22,7 @@ This version requires the 3.0.0 `kafka-clients`
See [Exactly Once Semantics](#exactly-once) and [KIP-447](https://cwiki.apache.org/confluence/display/KAFKA/KIP-447%3A+Producer+scalability+for+exactly+once+semantics) for more information.
-#### [](#x28-packages)2.1.2. Package Changes
+#### 2.1.2. Package Changes
Classes and interfaces related to type mapping have been moved from `…support.converter` to `…support.mapping`.
@@ -34,13 +34,13 @@ Classes and interfaces related to type mapping have been moved from `…suppo
* `Jackson2JavaTypeMapper`
-#### [](#x28-ooo-commits)2.1.3. Out of Order Manual Commits
+#### 2.1.3. Out of Order Manual Commits
The listener container can now be configured to accept manual offset commits out of order (usually asynchronously).
The container will defer the commit until the missing offset is acknowledged.
See [Manually Committing Offsets](#ooo-commits) for more information.
-#### [](#x28-batch-overrude)2.1.4. `@KafkaListener` Changes
+#### 2.1.4. `@KafkaListener` Changes
It is now possible to specify whether the listener method is a batch listener on the method itself.
This allows the same container factory to be used for both record and batch listeners.
@@ -54,17 +54,17 @@ See [Conversion Errors with Batch Error Handlers](#batch-listener-conv-errors) f
`RecordFilterStrategy`, when used with batch listeners, can now filter the entire batch in one call.
See the note at the end of [Batch Listeners](#batch-listeners) for more information.
-#### [](#x28-template)2.1.5. `KafkaTemplate` Changes
+#### 2.1.5. `KafkaTemplate` Changes
You can now receive a single record, given the topic, partition and offset.
See [Using `KafkaTemplate` to Receive](#kafka-template-receive) for more information.
-#### [](#x28-eh)2.1.6. `CommonErrorHandler` Added
+#### 2.1.6. `CommonErrorHandler` Added
The legacy `GenericErrorHandler` and its sub-interface hierarchies for record an batch listeners have been replaced by a new single interface `CommonErrorHandler` with implementations corresponding to most legacy implementations of `GenericErrorHandler`.
See [Container Error Handlers](#error-handlers) for more information.
-#### [](#x28-lcc)2.1.7. Listener Container Changes
+#### 2.1.7. Listener Container Changes
The `interceptBeforeTx` container property is now `true` by default.
@@ -73,18 +73,18 @@ Both exceptions are considered fatal and the container will stop by default, unl
See [Using `KafkaMessageListenerContainer`](#kafka-container) and [Listener Container Properties](#container-props) for more information.
-#### [](#x28-serializers)2.1.8. Serializer/Deserializer Changes
+#### 2.1.8. Serializer/Deserializer Changes
The `DelegatingByTopicSerializer` and `DelegatingByTopicDeserializer` are now provided.
See [Delegating Serializer and Deserializer](#delegating-serialization) for more information.
-#### [](#x28-dlpr)2.1.9. `DeadLetterPublishingRecover` Changes
+#### 2.1.9. `DeadLetterPublishingRecover` Changes
The property `stripPreviousExceptionHeaders` is now `true` by default.
See [Managing Dead Letter Record Headers](#dlpr-headers) for more information.
-#### [](#x28-retryable-topics-changes)2.1.10. Retryable Topics Changes
+#### 2.1.10. Retryable Topics Changes
Now you can use the same factory for retryable and non-retryable topics.
See [Specifying a ListenerContainerFactory](#retry-topic-lcf) for more information.
@@ -95,11 +95,11 @@ Refer to [Exception Classifier](#retry-topic-ex-classifier) to see how to manage
The KafkaBackOffException thrown when using the retryable topics feature is now logged at DEBUG level.
See [[change-kboe-logging-level]](#change-kboe-logging-level) if you need to change the logging level back to WARN or set it to any other level.
-## [](#introduction)3. Introduction
+## 3. Introduction
This first part of the reference documentation is a high-level overview of Spring for Apache Kafka and the underlying concepts and some code snippets that can help you get up and running as quickly as possible.
-### [](#quick-tour)3.1. Quick Tour
+### 3.1. Quick Tour
Prerequisites: You must install and run Apache Kafka.
Then you must put the Spring for Apache Kafka (`spring-kafka`) JAR and all of its dependencies on your class path.
@@ -143,7 +143,7 @@ compile 'org.springframework.kafka:spring-kafka'
However, the quickest way to get started is to use [start.spring.io](https://start.spring.io) (or the wizards in Spring Tool Suits and Intellij IDEA) and create a project, selecting 'Spring for Apache Kafka' as a dependency.
-#### [](#compatibility)3.1.1. Compatibility
+#### 3.1.1. Compatibility
This quick tour works with the following versions:
@@ -153,14 +153,14 @@ This quick tour works with the following versions:
* Minimum Java version: 8
-#### [](#getting-started)3.1.2. Getting Started
+#### 3.1.2. Getting Started
The simplest way to get started is to use [start.spring.io](https://start.spring.io) (or the wizards in Spring Tool Suits and Intellij IDEA) and create a project, selecting 'Spring for Apache Kafka' as a dependency.
Refer to the [Spring Boot documentation](https://docs.spring.io/spring-boot/docs/current/reference/html/spring-boot-features.html#boot-features-kafka) for more information about its opinionated auto configuration of the infrastructure beans.
Here is a minimal consumer application.
-##### [](#spring-boot-consumer-app)Spring Boot Consumer App
+##### Spring Boot Consumer App
Example 1. Application
@@ -217,7 +217,7 @@ spring.kafka.consumer.auto-offset-reset=earliest
The `NewTopic` bean causes the topic to be created on the broker; it is not needed if the topic already exists.
-##### [](#spring-boot-producer-app)Spring Boot Producer App
+##### Spring Boot Producer App
Example 3. Application
@@ -270,7 +270,7 @@ class Application {
}
```
-##### [](#with-java-configuration-no-spring-boot)With Java Configuration (No Spring Boot)
+#####
| |Spring for Apache Kafka is designed to be used in a Spring Application Context. For example, if you create the listener container yourself outside of a Spring context, not all functions will work unless you satisfy all of the `…Aware` interfaces that the container implements.|
|---|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
@@ -435,17 +435,17 @@ class Config {
As you can see, you have to define several infrastructure beans when not using Spring Boot.
-## [](#reference)4. Reference
+## 4. Reference
This part of the reference documentation details the various components that comprise Spring for Apache Kafka.
The [main chapter](#kafka) covers the core classes to develop a Kafka application with Spring.
-### [](#kafka)4.1. Using Spring for Apache Kafka
+### 4.1. Using Spring for Apache Kafka
This section offers detailed explanations of the various concerns that impact using Spring for Apache Kafka.
For a quick but less detailed introduction, see [Quick Tour](#quick-tour).
-#### [](#connecting)4.1.1. Connecting to Kafka
+#### 4.1.1. Connecting to Kafka
* `KafkaAdmin` - see [Configuring Topics](#configuring-topics)
@@ -467,7 +467,7 @@ When using `@KafkaListener` s, `stop()` and `start()` the `KafkaListenerEndpoint
See the Javadocs for more information.
-##### [](#factory-listeners)Factory Listeners
+##### Factory Listeners
Starting with version 2.5, the `DefaultKafkaProducerFactory` and `DefaultKafkaConsumerFactory` can be configured with a `Listener` to receive notifications whenever a producer or consumer is created or closed.
@@ -505,7 +505,7 @@ These listeners can be used, for example, to create and bind a Micrometer `Kafka
The framework provides listeners that do exactly that; see [Micrometer Native Metrics](#micrometer-native).
-#### [](#configuring-topics)4.1.2. Configuring Topics
+#### 4.1.2. Configuring Topics
If you define a `KafkaAdmin` bean in your application context, it can automatically add topics to the broker.
To do so, you can add a `NewTopic` `@Bean` for each topic to the application context.
@@ -689,15 +689,15 @@ private KafkaAdmin admin;
client.close();
```
-#### [](#sending-messages)4.1.3. Sending Messages
+#### 4.1.3. Sending Messages
This section covers how to send messages.
-##### [](#kafka-template)Using `KafkaTemplate`
+##### Using `KafkaTemplate`
This section covers how to use `KafkaTemplate` to send messages.
-###### [](#overview)Overview
+###### Overview
The `KafkaTemplate` wraps a producer and provides convenience methods to send data to Kafka topics.
The following listing shows the relevant methods from `KafkaTemplate`:
@@ -890,7 +890,7 @@ If you wish to block the sending thread to await the result, you can invoke the
You may wish to invoke `flush()` before waiting or, for convenience, the template has a constructor with an `autoFlush` parameter that causes the template to `flush()` on each send.
Flushing is only needed if you have set the `linger.ms` producer property and want to immediately send a partial batch.
-###### [](#examples)Examples
+###### Examples
This section shows examples of sending messages to Kafka:
@@ -938,7 +938,7 @@ public void sendToKafka(final MyOutputData data) {
Note that the cause of the `ExecutionException` is `KafkaProducerException` with the `failedProducerRecord` property.
-##### [](#routing-template)Using `RoutingKafkaTemplate`
+##### Using `RoutingKafkaTemplate`
Starting with version 2.5, you can use a `RoutingKafkaTemplate` to select the producer at runtime, based on the destination `topic` name.
@@ -989,7 +989,7 @@ The corresponding `@KafkaListener` s for this example are shown in [Annotation P
For another technique to achieve similar results, but with the additional capability of sending different types to the same topic, see [Delegating Serializer and Deserializer](#delegating-serialization).
-##### [](#producer-factory)Using `DefaultKafkaProducerFactory`
+##### Using `DefaultKafkaProducerFactory`
As seen in [Using `KafkaTemplate`](#kafka-template), a `ProducerFactory` is used to create the producer.
@@ -1033,7 +1033,7 @@ void removeConfig(String configKey);
Starting with version 2.8, if you provide serializers as objects (in the constructor or via the setters), the factory will invoke the `configure()` method to configure them with the configuration properties.
-##### [](#replying-template)Using `ReplyingKafkaTemplate`
+##### Using `ReplyingKafkaTemplate`
Version 2.1.3 introduced a subclass of `KafkaTemplate` to provide request/reply semantics.
The class is named `ReplyingKafkaTemplate` and has two additional methods; the following shows the method signatures:
@@ -1248,7 +1248,7 @@ These header names are used by the `@KafkaListener` infrastructure to route the
Starting with version 2.3, you can customize the header names - the template has 3 properties `correlationHeaderName`, `replyTopicHeaderName`, and `replyPartitionHeaderName`.
This is useful if your server is not a Spring application (or does not use the `@KafkaListener`).
-###### [](#exchanging-messages)Request/Reply with `Message` s
+###### Request/Reply with `Message` s
Version 2.7 added methods to the `ReplyingKafkaTemplate` to send and receive `spring-messaging` 's `Message` abstraction:
@@ -1343,7 +1343,7 @@ val things = future2?.get(10, TimeUnit.SECONDS)?.payload
things?.forEach(Consumer { thing1: Thing? -> log.info(thing1.toString()) })
```
-##### [](#reply-message)Reply Type Message\
+##### Reply Type Message\
When the `@KafkaListener` returns a `Message`, with versions before 2.5, it was necessary to populate the reply topic and correlation id headers.
In this example, we use the reply topic header from the request:
@@ -1375,7 +1375,7 @@ public Message messageReturn(String in) {
}
```
-##### [](#aggregating-request-reply)Aggregating Multiple Replies
+##### Aggregating Multiple Replies
The template in [Using `ReplyingKafkaTemplate`](#replying-template) is strictly for a single request/reply scenario.
For cases where multiple receivers of a single message return a reply, you can use the `AggregatingReplyingKafkaTemplate`.
@@ -1429,11 +1429,11 @@ The real `ConsumerRecord` s in the `Collection` contain the actual topic(s) from
| |If you use an [`ErrorHandlingDeserializer`](#error-handling-deserializer) with this aggregating template, the framework will not automatically detect `DeserializationException` s. Instead, the record (with a `null` value) will be returned intact, with the deserialization exception(s) in headers. It is recommended that applications call the utility method `ReplyingKafkaTemplate.checkDeserialization()` method to determine if a deserialization exception occurred. See its javadocs for more information. The `replyErrorChecker` is also not called for this aggregating template; you should perform the checks on each element of the reply.|
|---|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-#### [](#receiving-messages)4.1.4. Receiving Messages
+#### 4.1.4. Receiving Messages
You can receive messages by configuring a `MessageListenerContainer` and providing a message listener or by using the `@KafkaListener` annotation.
-##### [](#message-listeners)Message Listeners
+##### Message Listeners
When you use a [message listener container](#message-listener-container), you must provide a listener to receive data.
There are currently eight supported interfaces for message listeners.
@@ -1505,7 +1505,7 @@ public interface BatchAcknowledgingConsumerAwareMessageListener extends Ba
| |You should not execute any `Consumer` methods that affect the consumer’s positions and or committed offsets in your listener; the container needs to manage such information.|
|---|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-##### [](#message-listener-container)Message Listener Containers
+##### Message Listener Containers
Two `MessageListenerContainer` implementations are provided:
@@ -1535,7 +1535,7 @@ Starting with versions 2.3.8, 2.4.6, the `ConcurrentMessageListenerContainer` no
The `group.instance.id` is suffixed with `-n` with `n` starting at `1`.
This, together with an increased `session.timeout.ms`, can be used to reduce rebalance events, for example, when application instances are restarted.
-###### [](#kafka-container)Using `KafkaMessageListenerContainer`
+###### Using `KafkaMessageListenerContainer`
The following constructor is available:
@@ -1615,7 +1615,7 @@ Defining `authExceptionRetryInterval` allows the container to recover when prope
Starting with version 2.8, when creating the consumer factory, if you provide deserializers as objects (in the constructor or via the setters), the factory will invoke the `configure()` method to configure them with the configuration properties.
-###### [](#using-ConcurrentMessageListenerContainer)Using `ConcurrentMessageListenerContainer`
+###### Using `ConcurrentMessageListenerContainer`
The single constructor is similar to the `KafkaListenerContainer` constructor.
The following listing shows the constructor’s signature:
@@ -1649,7 +1649,7 @@ The metrics are grouped into the `Map` by the `cli
Starting with version 2.3, the `ContainerProperties` provides an `idleBetweenPolls` option to let the main loop in the listener container to sleep between `KafkaConsumer.poll()` calls.
An actual sleep interval is selected as the minimum from the provided option and difference between the `max.poll.interval.ms` consumer config and the current records batch processing time.
-###### [](#committing-offsets)Committing Offsets
+###### Committing Offsets
Several options are provided for committing offsets.
If the `enable.auto.commit` consumer property is `true`, Kafka auto-commits the offsets according to its configuration.
@@ -1722,14 +1722,14 @@ See [Container Error Handlers](#error-handlers) for more information.
| |When using partition assignment via group management, it is important to ensure the `sleep` argument (plus the time spent processing records from the previous poll) is less than the consumer `max.poll.interval.ms` property.|
|---|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-###### [](#container-auto-startup)Listener Container Auto Startup
+###### Listener Container Auto Startup
The listener containers implement `SmartLifecycle`, and `autoStartup` is `true` by default.
The containers are started in a late phase (`Integer.MAX-VALUE - 100`).
Other components that implement `SmartLifecycle`, to handle data from listeners, should be started in an earlier phase.
The `- 100` leaves room for later phases to enable components to be auto-started after the containers.
-##### [](#ooo-commits)Manually Committing Offsets
+##### Manually Committing Offsets
Normally, when using `AckMode.MANUAL` or `AckMode.MANUAL_IMMEDIATE`, the acknowledgments must be acknowledged in order, because Kafka does not maintain state for each record, only a committed offset for each group/partition.
Starting with version 2.8, you can now set the container property `asyncAcks`, which allows the acknowledgments for records returned by the poll to be acknowledged in any order.
@@ -1739,7 +1739,7 @@ The consumer will be paused (no new records delivered) until all the offsets for
| |While this feature allows applications to process records asynchronously, it should be understood that it increases the possibility of duplicate deliveries after a failure.|
|---|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-##### [](#kafka-listener-annotation)`@KafkaListener` Annotation
+##### `@KafkaListener` Annotation
The `@KafkaListener` annotation is used to designate a bean method as a listener for a listener container.
The bean is wrapped in a `MessagingMessageListenerAdapter` configured with various features, such as converters to convert the data, if necessary, to match the method parameters.
@@ -1747,7 +1747,7 @@ The bean is wrapped in a `MessagingMessageListenerAdapter` configured with vario
You can configure most attributes on the annotation with SpEL by using `#{…}` or property placeholders (`${…}`).
See the [Javadoc](https://docs.spring.io/spring-kafka/api/org/springframework/kafka/annotation/KafkaListener.html) for more information.
-###### [](#record-listener)Record Listeners
+###### Record Listeners
The `@KafkaListener` annotation provides a mechanism for simple POJO listeners.
The following example shows how to use it:
@@ -1816,7 +1816,7 @@ public void listen(String data) {
}
```
-###### [](#manual-assignment)Explicit Partition Assignment
+###### Explicit Partition Assignment
You can also configure POJO listeners with explicit topics and partitions (and, optionally, their initial offsets).
The following example shows how to do so:
@@ -1881,7 +1881,7 @@ public void listen(ConsumerRecord record) {
The initial offset will be applied to all 6 partitions.
-###### [](#manual-acknowledgment)Manual Acknowledgment
+###### Manual Acknowledgment
When using manual `AckMode`, you can also provide the listener with the `Acknowledgment`.
The following example also shows how to use a different container factory.
@@ -1895,7 +1895,7 @@ public void listen(String data, Acknowledgment ack) {
}
```
-###### [](#consumer-record-metadata)Consumer Record Metadata
+###### Consumer Record Metadata
Finally, metadata about the record is available from message headers.
You can use the following header names to retrieve the headers of the message:
@@ -1940,7 +1940,7 @@ public void listen(String str, ConsumerRecordMetadata meta) {
This contains all the data from the `ConsumerRecord` except the key and value.
-###### [](#batch-listeners)Batch Listeners
+###### Batch Listeners
Starting with version 1.1, you can configure `@KafkaListener` methods to receive the entire batch of consumer records received from the consumer poll.
To configure the listener container factory to create batch listeners, you can set the `batchListener` property.
@@ -2037,7 +2037,7 @@ public void pollResults(ConsumerRecords records) {
| |If the container factory has a `RecordFilterStrategy` configured, it is ignored for `ConsumerRecords` listeners, with a `WARN` log message emitted. Records can only be filtered with a batch listener if the `>` form of listener is used. By default, records are filtered one-at-a-time; starting with version 2.8, you can override `filterBatch` to filter the entire batch in one call.|
|---|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-###### [](#annotation-properties)Annotation Properties
+###### Annotation Properties
Starting with version 2.0, the `id` property (if present) is used as the Kafka consumer `group.id` property, overriding the configured property in the consumer factory, if present.
You can also set `groupId` explicitly or set `idIsGroup` to false to restore the previous behavior of using the consumer factory `group.id`.
@@ -2126,7 +2126,7 @@ public void listen2(byte[] in) {
}
```
-##### [](#listener-group-id)Obtaining the Consumer `group.id`
+##### Obtaining the Consumer `group.id`
When running the same listener code in multiple containers, it may be useful to be able to determine which container (identified by its `group.id` consumer property) that a record came from.
@@ -2144,7 +2144,7 @@ public void listener(@Payload String foo,
| |This is available in record listeners and batch listeners that receive a `List` of records. It is **not** available in a batch listener that receives a `ConsumerRecords` argument. Use the `KafkaUtils` mechanism in that case.|
|---|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-##### [](#container-thread-naming)Container Thread Naming
+##### Container Thread Naming
Listener containers currently use two task executors, one to invoke the consumer and another that is used to invoke the listener when the kafka consumer property `enable.auto.commit` is `false`.
You can provide custom executors by setting the `consumerExecutor` and `listenerExecutor` properties of the container’s `ContainerProperties`.
@@ -2156,7 +2156,7 @@ This executor creates threads with names similar to `-C-1` (consumer t
For the `ConcurrentMessageListenerContainer`, the `` part of the thread name becomes `-m`, where `m` represents the consumer instance.`n` increments each time the container is started.
So, with a bean name of `container`, threads in this container will be named `container-0-C-1`, `container-1-C-1` etc., after the container is started the first time; `container-0-C-2`, `container-1-C-2` etc., after a stop and subsequent start.
-##### [](#kafka-listener-meta)`@KafkaListener` as a Meta Annotation
+##### `@KafkaListener` as a Meta Annotation
Starting with version 2.2, you can now use `@KafkaListener` as a meta annotation.
The following example shows how to do so:
@@ -2189,7 +2189,7 @@ public void listen1(String in) {
}
```
-##### [](#class-level-kafkalistener)`@KafkaListener` on a Class
+##### `@KafkaListener` on a Class
When you use `@KafkaListener` at the class-level, you must specify `@KafkaHandler` at the method level.
When messages are delivered, the converted message payload type is used to determine which method to call.
@@ -2247,7 +2247,7 @@ void listen(Object in, @Header(KafkaHeaders.RECORD_METADATA) ConsumerRecordMetad
}
```
-##### [](#kafkalistener-attrs)`@KafkaListener` Attribute Modification
+##### `@KafkaListener` Attribute Modification
Starting with version 2.7.2, you can now programmatically modify annotation attributes before the container is created.
To do so, add one or more `KafkaListenerAnnotationBeanPostProcessor.AnnotationEnhancer` to the application context.`AnnotationEnhancer` is a `BiFunction, AnnotatedElement, Map` and must return a map of attributes.
@@ -2272,7 +2272,7 @@ public static AnnotationEnhancer groupIdEnhancer() {
}
```
-##### [](#kafkalistener-lifecycle)`@KafkaListener` Lifecycle Management
+##### `@KafkaListener` Lifecycle Management
The listener containers created for `@KafkaListener` annotations are not beans in the application context.
Instead, they are registered with an infrastructure bean of type `KafkaListenerEndpointRegistry`.
@@ -2307,7 +2307,7 @@ A collection of managed containers can be obtained by calling the registry’s `
Version 2.2.5 added a convenience method `getAllListenerContainers()`, which returns a collection of all containers, including those managed by the registry and those declared as beans.
The collection returned will include any prototype beans that have been initialized, but it will not initialize any lazy bean declarations.
-##### [](#kafka-validation)`@KafkaListener` `@Payload` Validation
+##### `@KafkaListener` `@Payload` Validation
Starting with version 2.2, it is now easier to add a `Validator` to validate `@KafkaListener` `@Payload` arguments.
Previously, you had to configure a custom `DefaultMessageHandlerMethodFactory` and add it to the registrar.
@@ -2385,7 +2385,7 @@ public KafkaListenerErrorHandler validationErrorHandler() {
Starting with version 2.5.11, validation now works on payloads for `@KafkaHandler` methods in a class-level listener.
See [`@KafkaListener` on a Class](#class-level-kafkalistener).
-##### [](#rebalance-listeners)Rebalancing Listeners
+##### Rebalancing Listeners
`ContainerProperties` has a property called `consumerRebalanceListener`, which takes an implementation of the Kafka client’s `ConsumerRebalanceListener` interface.
If this property is not provided, the container configures a logging listener that logs rebalance events at the `INFO` level.
@@ -2438,7 +2438,7 @@ containerProperties.setConsumerRebalanceListener(new ConsumerAwareRebalanceListe
| |Starting with version 2.4, a new method `onPartitionsLost()` has been added (similar to a method with the same name in `ConsumerRebalanceLister`). The default implementation on `ConsumerRebalanceLister` simply calls `onPartionsRevoked`. The default implementation on `ConsumerAwareRebalanceListener` does nothing. When supplying the listener container with a custom listener (of either type), it is important that your implementation not call `onPartitionsRevoked` from `onPartitionsLost`. If you implement `ConsumerRebalanceListener` you should override the default method. This is because the listener container will call its own `onPartitionsRevoked` from its implementation of `onPartitionsLost` after calling the method on your implementation. If you implementation delegates to the default behavior, `onPartitionsRevoked` will be called twice each time the `Consumer` calls that method on the container’s listener.|
|---|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-##### [](#annotation-send-to)Forwarding Listener Results using `@SendTo`
+##### Forwarding Listener Results using `@SendTo`
Starting with version 2.0, if you also annotate a `@KafkaListener` with a `@SendTo` annotation and the method invocation returns a result, the result is forwarded to the topic specified by the `@SendTo`.
@@ -2580,7 +2580,7 @@ When using request/reply semantics, the target partition can be requested by the
| |If a listener method returns an `Iterable`, by default a record for each element as the value is sent. Starting with version 2.3.5, set the `splitIterables` property on `@KafkaListener` to `false` and the entire result will be sent as the value of a single `ProducerRecord`. This requires a suitable serializer in the reply template’s producer configuration. However, if the reply is `Iterable>` the property is ignored and each message is sent separately.|
|---|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-##### [](#filtering-messages)Filtering Messages
+##### Filtering Messages
In certain scenarios, such as rebalancing, a message that has already been processed may be redelivered.
The framework cannot know whether such a message has been processed or not.
@@ -2599,11 +2599,11 @@ In addition, a `FilteringBatchMessageListenerAdapter` is provided, for when you
| |The `FilteringBatchMessageListenerAdapter` is ignored if your `@KafkaListener` receives a `ConsumerRecords` instead of `List>`, because `ConsumerRecords` is immutable.|
|---|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-##### [](#retrying-deliveries)Retrying Deliveries
+##### Retrying Deliveries
See the `DefaultErrorHandler` in [Handling Exceptions](#annotation-error-handling).
-##### [](#sequencing)Starting `@KafkaListener` s in Sequence
+##### Starting `@KafkaListener` s in Sequence
A common use case is to start a listener after another listener has consumed all the records in a topic.
For example, you may want to load the contents of one or more compacted topics into memory before processing records from other topics.
@@ -2652,7 +2652,7 @@ As an aside; previously, containers in each group were added to a bean of type `
These collections are now deprecated in favor of beans of type `ContainerGroup` with a bean name that is the group name, suffixed with `.group`; in the example above, there would be 2 beans `g1.group` and `g2.group`.
The `Collection` beans will be removed in a future release.
-##### [](#kafka-template-receive)Using `KafkaTemplate` to Receive
+##### Using `KafkaTemplate` to Receive
This section covers how to use `KafkaTemplate` to receive messages.
@@ -2673,87 +2673,87 @@ As you can see, you need to know the partition and offset of the record(s) you n
With the last two methods, each record is retrieved individually and the results assembled into a `ConsumerRecords` object.
When creating the `TopicPartitionOffset` s for the request, only positive, absolute offsets are supported.
-#### [](#container-props)4.1.5. Listener Container Properties
+#### 4.1.5. Listener Container Properties
| Property | Default | Description |
|---------------------------------------------------------------|-------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| []()[`ackCount`](#ackCount) | 1 | The number of records before committing pending offsets when the `ackMode` is `COUNT` or `COUNT_TIME`. |
-| []()[`adviceChain`](#adviceChain) | `null` | A chain of `Advice` objects (e.g. `MethodInterceptor` around advice) wrapping the message listener, invoked in order. |
-| []()[`ackMode`](#ackMode) | BATCH | Controls how often offsets are committed - see [Committing Offsets](#committing-offsets). |
-| []()[`ackOnError`](#ackOnError) | `false` | [DEPRECATED in favor of `ErrorHandler.isAckAfterHandle()`] |
-| []()[`ackTime`](#ackTime) | 5000 | The time in milliseconds after which pending offsets are committed when the `ackMode` is `TIME` or `COUNT_TIME`. |
-| []()[`assignmentCommitOption`](#assignmentCommitOption) | LATEST\_ONLY \_NO\_TX | Whether or not to commit the initial position on assignment; by default, the initial offset will only be committed if the `ConsumerConfig.AUTO_OFFSET_RESET_CONFIG` is `latest` and it won’t run in a transaction even if there is a transaction manager present. See the javadocs for `ContainerProperties.AssignmentCommitOption` for more information about the available options. |
-|[]()[`authExceptionRetryInterval`](#authExceptionRetryInterval)| `null` | When not null, a `Duration` to sleep between polls when an `AuthenticationException` or `AuthorizationException` is thrown by the Kafka client. When null, such exceptions are considered fatal and the container will stop. |
-| []()[`clientId`](#clientId) | (empty string) | A prefix for the `client.id` consumer property. Overrides the consumer factory `client.id` property; in a concurrent container, `-n` is added as a suffix for each consumer instance. |
-| []()[`checkDeserExWhenKeyNull`](#checkDeserExWhenKeyNull) | false | Set to `true` to always check for a `DeserializationException` header when a `null` `key` is received. Useful when the consumer code cannot determine that an `ErrorHandlingDeserializer` has been configured, such as when using a delegating deserializer. |
-| []()[`checkDeserExWhenValueNull`](#checkDeserExWhenValueNull) | false | Set to `true` to always check for a `DeserializationException` header when a `null` `value` is received. Useful when the consumer code cannot determine that an `ErrorHandlingDeserializer` has been configured, such as when using a delegating deserializer. |
-| []()[`commitCallback`](#commitCallback) | `null` | When present and `syncCommits` is `false` a callback invoked after the commit completes. |
-| []()[`commitLogLevel`](#commitLogLevel) | DEBUG | The logging level for logs pertaining to committing offsets. |
-| []()[`consumerRebalanceListener`](#consumerRebalanceListener) | `null` | A rebalance listener; see [Rebalancing Listeners](#rebalance-listeners). |
-| []()[`consumerStartTimout`](#consumerStartTimout) | 30s | The time to wait for the consumer to start before logging an error; this might happen if, say, you use a task executor with insufficient threads. |
-| []()[`consumerTaskExecutor`](#consumerTaskExecutor) |`SimpleAsyncTaskExecutor`| A task executor to run the consumer threads. The default executor creates threads named `-C-n`; with the `KafkaMessageListenerContainer`, the name is the bean name; with the `ConcurrentMessageListenerContainer` the name is the bean name suffixed with `-n` where n is incremented for each child container. |
-| []()[`deliveryAttemptHeader`](#deliveryAttemptHeader) | `false` | See [Delivery Attempts Header](#delivery-header). |
-| []()[`eosMode`](#eosMode) | `V2` | Exactly Once Semantics mode; see [Exactly Once Semantics](#exactly-once). |
-| []()[`fixTxOffsets`](#fixTxOffsets) | `false` |When consuming records produced by a transactional producer, and the consumer is positioned at the end of a partition, the lag can incorrectly be reported as greater than zero, due to the pseudo record used to indicate transaction commit/rollback and, possibly, the presence of rolled-back records. This does not functionally affect the consumer but some users have expressed concern that the "lag" is non-zero. Set this property to `true` and the container will correct such mis-reported offsets. The check is performed before the next poll to avoid adding significant complexity to the commit processing. At the time of writing, the lag will only be corrected if the consumer is configured with `isolation.level=read_committed` and `max.poll.records` is greater than 1. See [KAFKA-10683](https://issues.apache.org/jira/browse/KAFKA-10683) for more information.|
-| []()[`groupId`](#groupId) | `null` | Overrides the consumer `group.id` property; automatically set by the `@KafkaListener` `id` or `groupId` property. |
-| []()[`idleBeforeDataMultiplier`](#idleBeforeDataMultiplier) | 5.0 | Multiplier for `idleEventInterval` that is applied before any records are received. After a record is received, the multiplier is no longer applied. Available since version 2.8. |
-| []()[`idleBetweenPolls`](#idleBetweenPolls) | 0 | Used to slow down deliveries by sleeping the thread between polls. The time to process a batch of records plus this value must be less than the `max.poll.interval.ms` consumer property. |
-| []()[`idleEventInterval`](#idleEventInterval) | `null` | When set, enables publication of `ListenerContainerIdleEvent` s, see [Application Events](#events) and [Detecting Idle and Non-Responsive Consumers](#idle-containers). Also see `idleBeforeDataMultiplier`. |
-|[]()[`idlePartitionEventInterval`](#idlePartitionEventInterval)| `null` | When set, enables publication of `ListenerContainerIdlePartitionEvent` s, see [Application Events](#events) and [Detecting Idle and Non-Responsive Consumers](#idle-containers). |
-| []()[`kafkaConsumerProperties`](#kafkaConsumerProperties) | None | Used to override any arbitrary consumer properties configured on the consumer factory. |
-| []()[`logContainerConfig`](#logContainerConfig) | `false` | Set to true to log at INFO level all container properties. |
-| []()[`messageListener`](#messageListener) | `null` | The message listener. |
-| []()[`micrometerEnabled`](#micrometerEnabled) | `true` | Whether or not to maintain Micrometer timers for the consumer threads. |
-| []()[`missingTopicsFatal`](#missingTopicsFatal) | `false` | When true prevents the container from starting if the confifgured topic(s) are not present on the broker. |
-| []()[`monitorInterval`](#monitorInterval) | 30s | How often to check the state of the consumer threads for `NonResponsiveConsumerEvent` s. See `noPollThreshold` and `pollTimeout`. |
-| []()[`noPollThreshold`](#noPollThreshold) | 3.0 | Multiplied by `pollTimeOut` to determine whether to publish a `NonResponsiveConsumerEvent`. See `monitorInterval`. |
-| []()[`onlyLogRecordMetadata`](#onlyLogRecordMetadata) | `false` | Set to false to log the complete consumer record (in error, debug logs etc) instead of just `[[email protected]](/cdn-cgi/l/email-protection)`. |
-| []()[`pollTimeout`](#pollTimeout) | 5000 | The timeout passed into `Consumer.poll()`. |
-| []()[`scheduler`](#scheduler) |`ThreadPoolTaskScheduler`| A scheduler on which to run the consumer monitor task. |
-| []()[`shutdownTimeout`](#shutdownTimeout) | 10000 | The maximum time in ms to block the `stop()` method until all consumers stop and before publishing the container stopped event. |
-| []()[`stopContainerWhenFenced`](#stopContainerWhenFenced) | `false` | Stop the listener container if a `ProducerFencedException` is thrown. See [After-rollback Processor](#after-rollback) for more information. |
-| []()[`stopImmediate`](#stopImmediate) | `false` | When the container is stopped, stop processing after the current record instead of after processing all the records from the previous poll. |
-| []()[`subBatchPerPartition`](#subBatchPerPartition) | See desc. | When using a batch listener, if this is `true`, the listener is called with the results of the poll split into sub batches, one per partition. Default `false` except when using transactions with `EOSMode.ALPHA` - see [Exactly Once Semantics](#exactly-once). |
-| []()[`syncCommitTimeout`](#syncCommitTimeout) | `null` | The timeout to use when `syncCommits` is `true`. When not set, the container will attempt to determine the `default.api.timeout.ms` consumer property and use that; otherwise it will use 60 seconds. |
-| []()[`syncCommits`](#syncCommits) | `true` | Whether to use sync or async commits for offsets; see `commitCallback`. |
-| []()[`topics` `topicPattern` `topicPartitions`](#topics) | n/a | The configured topics, topic pattern or explicitly assigned topics/partitions. Mutually exclusive; at least one must be provided; enforced by `ContainerProperties` constructors. |
-| []()[`transactionManager`](#transactionManager) | `null` | See [Transactions](#transactions). |
+| | 1 | The number of records before committing pending offsets when the `ackMode` is `COUNT` or `COUNT_TIME`. |
+| wrapping the message listener, invoked in order. |
+| . |
+| `] |
+| | 5000 | The time in milliseconds after which pending offsets are committed when the `ackMode` is `TIME` or `COUNT_TIME`. |
+| | LATEST\_ONLY \_NO\_TX | Whether or not to commit the initial position on assignment; by default, the initial offset will only be committed if the `ConsumerConfig.AUTO_OFFSET_RESET_CONFIG` is `latest` and it won’t run in a transaction even if there is a transaction manager present. See the javadocs for `ContainerProperties.AssignmentCommitOption` for more information about the available options. |
+|| `null` | When not null, a `Duration` to sleep between polls when an `AuthenticationException` or `AuthorizationException` is thrown by the Kafka client. When null, such exceptions are considered fatal and the container will stop. |
+| | A prefix for the `client.id` consumer property. Overrides the consumer factory `client.id` property; in a concurrent container, `-n` is added as a suffix for each consumer instance. |
+| | false | Set to `true` to always check for a `DeserializationException` header when a `null` `key` is received. Useful when the consumer code cannot determine that an `ErrorHandlingDeserializer` has been configured, such as when using a delegating deserializer. |
+| | false | Set to `true` to always check for a `DeserializationException` header when a `null` `value` is received. Useful when the consumer code cannot determine that an `ErrorHandlingDeserializer` has been configured, such as when using a delegating deserializer. |
+| | `null` | When present and `syncCommits` is `false` a callback invoked after the commit completes. |
+| | DEBUG | The logging level for logs pertaining to committing offsets. |
+| . |
+| | 30s | The time to wait for the consumer to start before logging an error; this might happen if, say, you use a task executor with insufficient threads. |
+| |`SimpleAsyncTaskExecutor`| A task executor to run the consumer threads. The default executor creates threads named `-C-n`; with the `KafkaMessageListenerContainer`, the name is the bean name; with the `ConcurrentMessageListenerContainer` the name is the bean name suffixed with `-n` where n is incremented for each child container. |
+| . |
+| . |
+| for more information.|
+| | `null` | Overrides the consumer `group.id` property; automatically set by the `@KafkaListener` `id` or `groupId` property. |
+| | 5.0 | Multiplier for `idleEventInterval` that is applied before any records are received. After a record is received, the multiplier is no longer applied. Available since version 2.8. |
+| | 0 | Used to slow down deliveries by sleeping the thread between polls. The time to process a batch of records plus this value must be less than the `max.poll.interval.ms` consumer property. |
+| . Also see `idleBeforeDataMultiplier`. |
+|. |
+| | None | Used to override any arbitrary consumer properties configured on the consumer factory. |
+| | `false` | Set to true to log at INFO level all container properties. |
+| | `null` | The message listener. |
+| | `true` | Whether or not to maintain Micrometer timers for the consumer threads. |
+| are not present on the broker. |
+| | 30s | How often to check the state of the consumer threads for `NonResponsiveConsumerEvent` s. See `noPollThreshold` and `pollTimeout`. |
+| | 3.0 | Multiplied by `pollTimeOut` to determine whether to publish a `NonResponsiveConsumerEvent`. See `monitorInterval`. |
+| `. |
+| `. |
+| |`ThreadPoolTaskScheduler`| A scheduler on which to run the consumer monitor task. |
+| ` method until all consumers stop and before publishing the container stopped event. |
+| for more information. |
+| | `false` | When the container is stopped, stop processing after the current record instead of after processing all the records from the previous poll. |
+| . |
+| | `null` | The timeout to use when `syncCommits` is `true`. When not set, the container will attempt to determine the `default.api.timeout.ms` consumer property and use that; otherwise it will use 60 seconds. |
+| | `true` | Whether to use sync or async commits for offsets; see `commitCallback`. |
+| | n/a | The configured topics, topic pattern or explicitly assigned topics/partitions. Mutually exclusive; at least one must be provided; enforced by `ContainerProperties` constructors. |
+| . |
| Property | Default | Description |
|-------------------------------------------------------------|-------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| []()[`afterRollbackProcessor`](#afterRollbackProcessor) |`DefaultAfterRollbackProcessor`| An `AfterRollbackProcessor` to invoke after a transaction is rolled back. |
-|[]()[`applicationEventPublisher`](#applicationEventPublisher)| application context | The event publisher. |
-| []()[`batchErrorHandler`](#batchErrorHandler) | See desc. | Deprecated - see `commonErrorHandler`. |
-| []()[`batchInterceptor`](#batchInterceptor) | `null` | Set a `BatchInterceptor` to call before invoking the batch listener; does not apply to record listeners. Also see `interceptBeforeTx`. |
-| []()[`beanName`](#beanName) | bean name | The bean name of the container; suffixed with `-n` for child containers. |
-| []()[`commonErrorHandler`](#commonErrorHandler) | See desc. |`DefaultErrorHandler` or `null` when a `transactionManager` is provided when a `DefaultAfterRollbackProcessor` is used. See [Container Error Handlers](#error-handlers).|
-| []()[`containerProperties`](#containerProperties) | `ContainerProperties` | The container properties instance. |
-| []()[`errorHandler`](#errorHandler) | See desc. | Deprecated - see `commonErrorHandler`. |
-| []()[`genericErrorHandler`](#genericErrorHandler) | See desc. | Deprecated - see `commonErrorHandler`. |
-| []()[`groupId`](#groupId) | See desc. | The `containerProperties.groupId`, if present, otherwise the `group.id` property from the consumer factory. |
-| []()[`interceptBeforeTx`](#interceptBeforeTx) | `true` | Determines whether the `recordInterceptor` is called before or after a transaction starts. |
-| []()[`listenerId`](#listenerId) | See desc. | The bean name for user-configured containers or the `id` attribute of `@KafkaListener` s. |
-| []()[`pauseRequested`](#pauseRequested) | (read only) | True if a consumer pause has been requested. |
-| []()[`recordInterceptor`](#recordInterceptor) | `null` | Set a `RecordInterceptor` to call before invoking the record listener; does not apply to batch listeners. Also see `interceptBeforeTx`. |
-| []()[`topicCheckTimeout`](#topicCheckTimeout) | 30s | When the `missingTopicsFatal` container property is `true`, how long to wait, in seconds, for the `describeTopics` operation to complete. |
+| |`DefaultAfterRollbackProcessor`| An `AfterRollbackProcessor` to invoke after a transaction is rolled back. |
+|| application context | The event publisher. |
+| | See desc. | Deprecated - see `commonErrorHandler`. |
+| | `null` | Set a `BatchInterceptor` to call before invoking the batch listener; does not apply to record listeners. Also see `interceptBeforeTx`. |
+| | bean name | The bean name of the container; suffixed with `-n` for child containers. |
+| .|
+| | `ContainerProperties` | The container properties instance. |
+| | See desc. | Deprecated - see `commonErrorHandler`. |
+| | See desc. | Deprecated - see `commonErrorHandler`. |
+| | See desc. | The `containerProperties.groupId`, if present, otherwise the `group.id` property from the consumer factory. |
+| | `true` | Determines whether the `recordInterceptor` is called before or after a transaction starts. |
+| | See desc. | The bean name for user-configured containers or the `id` attribute of `@KafkaListener` s. |
+| | True if a consumer pause has been requested. |
+| | `null` | Set a `RecordInterceptor` to call before invoking the record listener; does not apply to batch listeners. Also see `interceptBeforeTx`. |
+| | 30s | When the `missingTopicsFatal` container property is `true`, how long to wait, in seconds, for the `describeTopics` operation to complete. |
| Property | Default | Description |
|-------------------------------------------------------------------|-----------|----------------------------------------------------------------------------------------------|
-| []()[`assignedPartitions`](#assignedPartitions) |(read only)| The partitions currently assigned to this container (explicitly or not). |
-|[]()[`assignedPartitionsByClientId`](#assignedPartitionsByClientId)|(read only)| The partitions currently assigned to this container (explicitly or not). |
-| []()[`clientIdSuffix`](#clientIdSuffix) | `null` |Used by the concurrent container to give each child container’s consumer a unique `client.id`.|
-| []()[`containerPaused`](#containerPaused) | n/a | True if pause has been requested and the consumer has actually paused. |
+| . |
+|. |
+| | `null` |Used by the concurrent container to give each child container’s consumer a unique `client.id`.|
+| | n/a | True if pause has been requested and the consumer has actually paused. |
| Property | Default | Description |
|-------------------------------------------------------------------|-----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| []()[`alwaysClientIdSuffix`](#alwaysClientIdSuffix) | `true` | Set to false to suppress adding a suffix to the `client.id` consumer property, when the `concurrency` is only 1. |
-| []()[`assignedPartitions`](#assignedPartitions) |(read only)| The aggregate of partitions currently assigned to this container’s child `KafkaMessageListenerContainer` s (explicitly or not). |
-|[]()[`assignedPartitionsByClientId`](#assignedPartitionsByClientId)|(read only)|The partitions currently assigned to this container’s child `KafkaMessageListenerContainer` s (explicitly or not), keyed by the child container’s consumer’s `client.id` property.|
-| []()[`concurrency`](#concurrency) | 1 | The number of child `KafkaMessageListenerContainer` s to manage. |
-| []()[`containerPaused`](#containerPaused) | n/a | True if pause has been requested and all child containers' consumer has actually paused. |
-| []()[`containers`](#containers) | n/a | A reference to all child `KafkaMessageListenerContainer` s. |
+| | `true` | Set to false to suppress adding a suffix to the `client.id` consumer property, when the `concurrency` is only 1. |
+| . |
+|, keyed by the child container’s consumer’s `client.id` property.|
+| | 1 | The number of child `KafkaMessageListenerContainer` s to manage. |
+| | n/a | True if pause has been requested and all child containers' consumer has actually paused. |
+| | n/a | A reference to all child `KafkaMessageListenerContainer` s. |
-#### [](#events)4.1.6. Application Events
+#### 4.1.6. Application Events
The following Spring application events are published by listener containers and their consumers:
@@ -2898,7 +2898,7 @@ if (event.getReason.equals(Reason.FENCED)) {
}
```
-##### [](#idle-containers)Detecting Idle and Non-Responsive Consumers
+##### Detecting Idle and Non-Responsive Consumers
While efficient, one problem with asynchronous consumers is detecting when they are idle.
You might want to take some action if no messages arrive for some period of time.
@@ -2946,7 +2946,7 @@ Receiving such an event lets you stop the containers, thus waking the consumer s
Starting with version 2.6.2, if a container has published a `ListenerContainerIdleEvent`, it will publish a `ListenerContainerNoLongerIdleEvent` when a record is subsequently received.
-##### [](#event-consumption)Event Consumption
+##### Event Consumption
You can capture these events by implementing `ApplicationListener` — either a general listener or one narrowed to only receive this specific event.
You can also use `@EventListener`, introduced in Spring Framework 4.2.
@@ -2983,12 +2983,12 @@ public class Listener {
| |If you wish to use the idle event to stop the lister container, you should not call `container.stop()` on the thread that calls the listener. Doing so causes delays and unnecessary log messages. Instead, you should hand off the event to a different thread that can then stop the container. Also, you should not `stop()` the container instance if it is a child container. You should stop the concurrent container instead.|
|---|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-###### [](#current-positions-when-idle)Current Positions when Idle
+###### Current Positions when Idle
Note that you can obtain the current positions when idle is detected by implementing `ConsumerSeekAware` in your listener.
See `onIdleContainer()` in [Seeking to a Specific Offset](#seek).
-#### [](#topicpartition-initial-offset)4.1.7. Topic/Partition Initial Offset
+#### 4.1.7. Topic/Partition Initial Offset
There are several ways to set the initial offset for a partition.
@@ -3002,7 +3002,7 @@ When you use group management where the broker assigns partitions:
* For an existing group ID, the initial offset is the current offset for that group ID.
You can, however, seek to a specific offset during initialization (or at any time thereafter).
-#### [](#seek)4.1.8. Seeking to a Specific Offset
+#### 4.1.8. Seeking to a Specific Offset
In order to seek, your listener must implement `ConsumerSeekAware`, which has the following methods:
@@ -3227,7 +3227,7 @@ public class SomeOtherBean {
}
```
-#### [](#container-factory)4.1.9. Container factory
+#### 4.1.9. Container factory
As discussed in [`@KafkaListener` Annotation](#kafka-listener-annotation), a `ConcurrentKafkaListenerContainerFactory` is used to create containers for annotated methods.
@@ -3264,7 +3264,7 @@ public KafkaListenerContainerFactory kafkaListenerContainerFactory() {
}
```
-#### [](#thread-safety)4.1.10. Thread Safety
+#### 4.1.10. Thread Safety
When using a concurrent message listener container, a single listener instance is invoked on all consumer threads.
Listeners, therefore, need to be thread-safe, and it is preferable to use stateless listeners.
@@ -3283,9 +3283,9 @@ Note that `SimpleThreadScope` does not destroy beans that have a destruction int
| |By default, the application context’s event multicaster invokes event listeners on the calling thread. If you change the multicaster to use an async executor, thread cleanup is not effective.|
|---|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-#### [](#micrometer)4.1.11. Monitoring
+#### 4.1.11. Monitoring
-##### [](#monitoring-listener-performance)Monitoring Listener Performance
+##### Monitoring Listener Performance
Starting with version 2.3, the listener container will automatically create and update Micrometer `Timer` s for the listener, if `Micrometer` is detected on the class path, and a single `MeterRegistry` is present in the application context.
The timers can be disabled by setting the `ContainerProperty` `micrometerEnabled` to `false`.
@@ -3305,7 +3305,7 @@ You can add additional tags using the `ContainerProperties` `micrometerTags` pro
| |With the concurrent container, timers are created for each thread and the `name` tag is suffixed with `-n` where n is `0` to `concurrency-1`.|
|---|---------------------------------------------------------------------------------------------------------------------------------------------|
-##### [](#monitoring-kafkatemplate-performance)Monitoring KafkaTemplate Performance
+##### Monitoring KafkaTemplate Performance
Starting with version 2.5, the template will automatically create and update Micrometer `Timer` s for send operations, if `Micrometer` is detected on the class path, and a single `MeterRegistry` is present in the application context.
The timers can be disabled by setting the template’s `micrometerEnabled` property to `false`.
@@ -3322,7 +3322,7 @@ The timers are named `spring.kafka.template` and have the following tags:
You can add additional tags using the template’s `micrometerTags` property.
-##### [](#micrometer-native)Micrometer Native Metrics
+##### Micrometer Native Metrics
Starting with version 2.5, the framework provides [Factory Listeners](#factory-listeners) to manage a Micrometer `KafkaClientMetrics` instance whenever producers and consumers are created and closed.
@@ -3369,11 +3369,11 @@ double count = this.meterRegistry.get("kafka.producer.node.incoming.byte.total")
A similar listener is provided for the `StreamsBuilderFactoryBean` - see [KafkaStreams Micrometer Support](#streams-micrometer).
-#### [](#transactions)4.1.12. Transactions
+#### 4.1.12. Transactions
This section describes how Spring for Apache Kafka supports transactions.
-##### [](#overview-2)Overview
+##### Overview
The 0.11.0.0 client library added support for transactions.
Spring for Apache Kafka adds support in the following ways:
@@ -3405,7 +3405,7 @@ With Spring Boot, it is only necessary to set the `spring.kafka.producer.transac
| |Starting with version 2.5.8, you can now configure the `maxAge` property on the producer factory. This is useful when using transactional producers that might lay idle for the broker’s `transactional.id.expiration.ms`. With current `kafka-clients`, this can cause a `ProducerFencedException` without a rebalance. By setting the `maxAge` to less than `transactional.id.expiration.ms`, the factory will refresh the producer if it is past it’s max age.|
|---|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-##### [](#using-kafkatransactionmanager)Using `KafkaTransactionManager`
+##### Using `KafkaTransactionManager`
The `KafkaTransactionManager` is an implementation of Spring Framework’s `PlatformTransactionManager`.
It is provided with a reference to the producer factory in its constructor.
@@ -3417,7 +3417,7 @@ If a transaction is active, any `KafkaTemplate` operations performed within the
The manager commits or rolls back the transaction, depending on success or failure.
You must configure the `KafkaTemplate` to use the same `ProducerFactory` as the transaction manager.
-##### [](#transaction-synchronization)Transaction Synchronization
+##### Transaction Synchronization
This section refers to producer-only transactions (transactions not started by a listener container); see [Using Consumer-Initiated Transactions](#container-transaction-manager) for information about chaining transactions when the container starts the transaction.
@@ -3440,14 +3440,14 @@ See [[ex-jdbc-sync]](#ex-jdbc-sync) for examples of an application that synchron
| |Starting with versions 2.5.17, 2.6.12, 2.7.9 and 2.8.0, if the commit fails on the synchronized transaction (after the primary transaction has committed), the exception will be thrown to the caller. Previously, this was silently ignored (logged at debug). Applications should take remedial action, if necessary, to compensate for the committed primary transaction.|
|---|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-##### [](#container-transaction-manager)Using Consumer-Initiated Transactions
+##### Using Consumer-Initiated Transactions
The `ChainedKafkaTransactionManager` is now deprecated, since version 2.7; see the javadocs for its super class `ChainedTransactionManager` for more information.
Instead, use a `KafkaTransactionManager` in the container to start the Kafka transaction and annotate the listener method with `@Transactional` to start the other transaction.
See [[ex-jdbc-sync]](#ex-jdbc-sync) for an example application that chains JDBC and Kafka transactions.
-##### [](#kafkatemplate-local-transactions)`KafkaTemplate` Local Transactions
+##### `KafkaTemplate` Local Transactions
You can use the `KafkaTemplate` to execute a series of operations within a local transaction.
The following example shows how to do so:
@@ -3467,7 +3467,7 @@ If an exception is thrown, the transaction is rolled back.
| |If there is a `KafkaTransactionManager` (or synchronized) transaction in process, it is not used. Instead, a new "nested" transaction is used.|
|---|--------------------------------------------------------------------------------------------------------------------------------------------------|
-##### [](#transaction-id-prefix)`transactionIdPrefix`
+##### `transactionIdPrefix`
As mentioned in [the overview](#transactions), the producer factory is configured with this property to build the producer `transactional.id` property.
There is a dichotomy when specifying this property in that, when running multiple instances of the application with `EOSMode.ALPHA`, it must be the same on all instances to satisfy fencing zombies (also mentioned in the overview) when producing records on a listener container thread.
@@ -3485,7 +3485,7 @@ This property must have a different value on each application instance.
This problem (different rules for `transactional.id`) has been eliminated when `EOSMode.BETA` is being used (with broker versions \>= 2.5); see [Exactly Once Semantics](#exactly-once).
-##### [](#tx-template-mixed)`KafkaTemplate` Transactional and non-Transactional Publishing
+##### `KafkaTemplate` Transactional and non-Transactional Publishing
Normally, when a `KafkaTemplate` is transactional (configured with a transaction-capable producer factory), transactions are required.
The transaction can be started by a `TransactionTemplate`, a `@Transactional` method, calling `executeInTransaction`, or by a listener container, when configured with a `KafkaTransactionManager`.
@@ -3494,7 +3494,7 @@ Starting with version 2.4.3, you can set the template’s `allowNonTransactional
In that case, the template will allow the operation to run without a transaction, by calling the `ProducerFactory` 's `createNonTransactionalProducer()` method; the producer will be cached, or thread-bound, as normal for reuse.
See [Using `DefaultKafkaProducerFactory`](#producer-factory).
-##### [](#transactions-batch)Transactions with Batch Listeners
+##### Transactions with Batch Listeners
When a listener fails while transactions are being used, the `AfterRollbackProcessor` is invoked to take some action after the rollback occurs.
When using the default `AfterRollbackProcessor` with a record listener, seeks are performed so that the failed record will be redelivered.
@@ -3552,7 +3552,7 @@ public static class Config {
}
```
-#### [](#exactly-once)4.1.13. Exactly Once Semantics
+#### 4.1.13. Exactly Once Semantics
You can provide a listener container with a `KafkaAwareTransactionManager` instance.
When so configured, the container starts a transaction before invoking the listener.
@@ -3602,7 +3602,7 @@ Refer to [KIP-447](https://cwiki.apache.org/confluence/display/KAFKA/KIP-447%3A+
`V1` and `V2` were previously `ALPHA` and `BETA`; they have been changed to align the framework with [KIP-732](https://cwiki.apache.org/confluence/display/KAFKA/KIP-732%3A+Deprecate+eos-alpha+and+replace+eos-beta+with+eos-v2).
-#### [](#interceptors)4.1.14. Wiring Spring Beans into Producer/Consumer Interceptors
+#### 4.1.14. Wiring Spring Beans into Producer/Consumer Interceptors
Apache Kafka provides a mechanism to add interceptors to producers and consumers.
These objects are managed by Kafka, not Spring, and so normal Spring dependency injection won’t work for wiring in dependent Spring Beans.
@@ -3737,7 +3737,7 @@ consumer interceptor in my foo bean
Received test
```
-#### [](#pause-resume)4.1.15. Pausing and Resuming Listener Containers
+#### 4.1.15. Pausing and Resuming Listener Containers
Version 2.1.3 added `pause()` and `resume()` methods to listener containers.
Previously, you could pause a consumer within a `ConsumerAwareMessageListener` and resume it by listening for a `ListenerContainerIdleEvent`, which provides access to the `Consumer` object.
@@ -3812,7 +3812,7 @@ ConsumerResumedEvent [partitions=[pause.resume.topic-1, pause.resume.topic-0]]
thing2
```
-#### [](#pause-resume-partitions)4.1.16. Pausing and Resuming Partitions on Listener Containers
+#### 4.1.16. Pausing and Resuming Partitions on Listener Containers
Since version 2.7 you can pause and resume the consumption of specific partitions assigned to that consumer by using the `pausePartition(TopicPartition topicPartition)` and `resumePartition(TopicPartition topicPartition)` methods in the listener containers.
The pausing and resuming takes place respectively before and after the `poll()` similar to the `pause()` and `resume()` methods.
@@ -3821,9 +3821,9 @@ The `isPartitionPaused()` method returns true if that partition has effectively
Also since version 2.7 `ConsumerPartitionPausedEvent` and `ConsumerPartitionResumedEvent` instances are published with the container as the `source` property and the `TopicPartition` instance.
-#### [](#serdes)4.1.17. Serialization, Deserialization, and Message Conversion
+#### 4.1.17. Serialization, Deserialization, and Message Conversion
-##### [](#overview-3)Overview
+##### Overview
Apache Kafka provides a high-level API for serializing and deserializing record values as well as their keys.
It is present with the `org.apache.kafka.common.serialization.Serializer` and`org.apache.kafka.common.serialization.Deserializer` abstractions with some built-in implementations.
@@ -3844,7 +3844,7 @@ constructors to accept `Serializer` and `Deserializer` instances for `keys` and
When you use this API, the `DefaultKafkaProducerFactory` and `DefaultKafkaConsumerFactory` also provide properties (through constructors or setter methods) to inject custom `Serializer` and `Deserializer` instances into the target `Producer` or `Consumer`.
Also, you can pass in `Supplier` or `Supplier` instances through constructors - these `Supplier` s are called on creation of each `Producer` or `Consumer`.
-##### [](#string-serde)String serialization
+##### String serialization
Since version 2.5, Spring for Apache Kafka provides `ToStringSerializer` and `ParseStringDeserializer` classes that use String representation of entities.
They rely on methods `toString` and some `Function` or `BiFunction` to parse the String and populate properties of an instance.
@@ -3889,7 +3889,7 @@ The method must be static and have a signature of either `(String, Headers)` or
A `ToFromStringSerde` is also provided, for use with Kafka Streams.
-##### [](#json-serde)JSON
+##### JSON
Spring for Apache Kafka also provides `JsonSerializer` and `JsonDeserializer` implementations that are based on the
Jackson JSON object mapper.
@@ -3916,7 +3916,7 @@ Starting with version 2.1, you can convey type information in record `Headers`,
In addition, you can configure the serializer and deserializer by using the following Kafka properties.
They have no effect if you have provided `Serializer` and `Deserializer` instances for `KafkaConsumer` and `KafkaProducer`, respectively.
-###### [](#serdes-json-config)Configuration Properties
+###### Configuration Properties
* `JsonSerializer.ADD_TYPE_INFO_HEADERS` (default `true`): You can set it to `false` to disable this feature on the `JsonSerializer` (sets the `addTypeInfo` property).
@@ -3946,7 +3946,7 @@ See also [[tip-json]](#tip-json).
| |Starting with version 2.8, if you construct the serializer or deserializer programmatically as shown in [Programmatic Construction](#prog-json), the above properties will be applied by the factories, as long as you have not set any properties explicitly (using `set*()` methods or using the fluent API). Previously, when creating programmatically, the configuration properties were never applied; this is still the case if you explicitly set properties on the object directly.|
|---|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-###### [](#serdes-mapping-types)Mapping Types
+###### Mapping Types
Starting with version 2.2, when using JSON, you can now provide type mappings by using the properties in the preceding list.
Previously, you had to customize the type mapper within the serializer and deserializer.
@@ -3986,7 +3986,7 @@ DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<
new IntegerDeserializer(), new JsonDeserializer<>(Cat1.class, false));
```
-###### [](#serdes-type-methods)Using Methods to Determine Types
+###### Using Methods to Determine Types
Starting with version 2.5, you can now configure the deserializer, via properties, to invoke a method to determine the target type.
If present, this will override any of the other techniques discussed above.
@@ -4034,7 +4034,7 @@ public static JavaType thing1Thing2JavaTypeForTopic(String topic, byte[] data, H
}
```
-###### [](#prog-json)Programmatic Construction
+###### Programmatic Construction
When constructing the serializer/deserializer programmatically for use in the producer/consumer factory, since version 2.3, you can use the fluent API, which simplifies configuration.
@@ -4080,9 +4080,9 @@ JsonDeserializer