Spring Kafka

https://github.com/spring-projects/spring-kafka
Admin
Spring for Apache Kafka provides Kafka integration for Spring applications, simplifying the...
Tokens:669,595
Snippets:3,815
Trust Score:9.7
License:Apache-2.0
Update:2 days ago
Show doc for...
Tokens:
Raw
### Seek Example Application

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/seek.adoc

A skeleton of a Spring Boot application demonstrating the setup for a seek-capable listener.

```java
@SpringBootApplication
public class SeekExampleApplication {
```

--------------------------------

### Programmatically start a container via KafkaListenerEndpointRegistry

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/kafkalistener-lifecycle.adoc

Example of auto-wiring the KafkaListenerEndpointRegistry and using it to manually start a specific container by its ID.

```java
@Autowired
private KafkaListenerEndpointRegistry registry;

...

    this.registry.getListenerContainer("myContainer").start();

...
```

--------------------------------

### Seek to Beginning Example

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/seek.adoc

An example of using onPartitionsAssigned to seek to the beginning of all assigned partitions, useful for compacted topics.

```java
public class MyListener implements ConsumerSeekAware {

    ...

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        callback.seekToBeginning(assignments.keySet());
    }

}
```

--------------------------------

### Example of Inbound-Only Mapper with Patterns

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/headers.adoc

Demonstrates creating an inbound mapper that excludes headers starting with 'abc' and includes all others.

```java
JsonKafkaHeaderMapper inboundMapper = JsonKafkaHeaderMapper.forInboundOnlyWithMatchers("!abc*", "*");
```

--------------------------------

### Create table for transaction example

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/tips.adoc

SQL statement to create a table used in the transaction synchronization example.

```sql
create table mytable (data varchar(20));
```

--------------------------------

### Install jars to local Maven cache

Source: https://github.com/spring-projects/spring-kafka/blob/main/README.md

Builds and installs the project jars into the local Maven cache.

```bash
./gradlew install
```

--------------------------------

### Manually Assigning All Partitions with SpEL and PartitionFinder

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/tips.adoc

Use this example to read all records from all partitions of a topic, such as for loading a distributed cache from a compacted topic. It uses a SpEL expression in @KafkaListener to dynamically create the partition list at startup, avoiding manual listing and recompilation when partitions change. Requires a PartitionFinder bean that uses ConsumerFactory to get partitions. Set ConsumerConfig.AUTO_OFFSET_RESET_CONFIG=earliest to load all records on startup, and set the container's AckMode to MANUAL to prevent offset commits for a null consumer group (automatically coerced since version 3.1).

```java
@KafkaListener(topicPartitions = @TopicPartition(topic = "compacted",
            partitions = "#{@finder.partitions('compacted')}",
            partitionOffsets = @PartitionOffset(partition = "*", initialOffset = "0")))
public void listen(@Header(KafkaHeaders.RECEIVED_KEY) String key, String payload) {
    ...
}

@Bean
public PartitionFinder finder(ConsumerFactory<String, String> consumerFactory) {
    return new PartitionFinder(consumerFactory);
}

public static class PartitionFinder {

    private final ConsumerFactory<String, String> consumerFactory;

    public PartitionFinder(ConsumerFactory<String, String> consumerFactory) {
        this.consumerFactory = consumerFactory;
    }

    public String[] partitions(String topic) {
        try (Consumer<String, String> consumer = consumerFactory.createConsumer()) {
            return consumer.partitionsFor(topic).stream()
                .map(pi -> "" + pi.partition())
                .toArray(String[]::new);
        }
    }

}
```

--------------------------------

### Usage of @SendTo with various routing strategies

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/annotation-send-to.adoc

Examples demonstrating runtime SpEL, config-time SpEL, static topic routing, and class-level @SendTo with method-level overrides.

```java
@KafkaListener(topics = "annotated21")
@SendTo("!{request.value()}") // runtime SpEL
public String replyingListener(String in) {
    ...
}

@KafkaListener(topics = "${some.property:annotated22}")
@SendTo("#{myBean.replyTopic}") // config time SpEL
public Collection<String> replyingBatchListener(List<String> in) {
    ...
}

@KafkaListener(topics = "annotated23", errorHandler = "replyErrorHandler")
@SendTo("annotated23reply") // static reply topic definition
public String replyingListenerWithErrorHandler(String in) {
    ...
}
...
@KafkaListener(topics = "annotated25")
@SendTo("annotated25reply1")
public class MultiListenerSendTo {

    @KafkaHandler
    public String foo(String in) {
        ...
    }

    @KafkaHandler
    @SendTo("!{'annotated25reply2'}")
    public String bar(@Payload(required = false) KafkaNull nul,
            @Header(KafkaHeaders.RECEIVED_KEY) int key) {
        ...
    }

}
```

--------------------------------

### Example of acceptable Git author format

Source: https://github.com/spring-projects/spring-kafka/blob/main/CONTRIBUTING.md

Please use your properly capitalized first and last name.

```text
Author: First Last <user@mail.com>
```

--------------------------------

### Non Blocking (Async) Send Example

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/sending-messages.adoc

Example of sending a message asynchronously using a ProducerRecord and handling success or failure in a whenComplete callback.

```java
public void sendToKafka(final MyOutputData data) {
    final ProducerRecord<String, String> record = createRecord(data);

    CompletableFuture<SendResult<String, String>> future = template.send(record);
    future.whenComplete((result, ex) -> {
        if (ex == null) {
            handleSuccess(data);
        }
        else {
            handleFailure(data, record, ex);
        }
    });
}
```

--------------------------------

### Listeners for RoutingKafkaTemplate example

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/listener-annotation.adoc

Corresponding listeners for the RoutingKafkaTemplate example. The first listens to topic 'one' with a String payload; the second listens to topic 'two' with a byte[] payload, overriding the value deserializer via annotation properties.

```java
@KafkaListener(id = "one", topics = "one")
public void listen1(String in) {
    System.out.println("1: " + in);
}

@KafkaListener(id = "two", topics = "two",
        properties = "value.deserializer:org.apache.kafka.common.serialization.ByteArrayDeserializer")
public void listen2(byte[] in) {
    System.out.println("2: " + new String(in));
}
```

--------------------------------

### Configure @KafkaListener with autoStartup

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/kafkalistener-lifecycle.adoc

Example of setting the autoStartup property to false on a @KafkaListener annotation to prevent automatic container startup.

```java
@KafkaListener(id = "myContainer", topics = "myTopic", autoStartup = "false")
public void listen(...) { ... }
```

--------------------------------

### Specify comma-delimited list of partitions or partition ranges

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/listener-annotation.adoc

Starting with version 2.6.4, you can specify a comma-delimited list of partitions or partition ranges. The range is inclusive; the example assigns partitions 0-5, 7, and 10-15.

```java
@KafkaListener(id = "pp", autoStartup = "false",
        topicPartitions = @TopicPartition(topic = "topic1",
                partitions = "0-5, 7, 10-15"))
public void process(String in) {
    ...
}
```

--------------------------------

### Receive list of headers in batch listener

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/listener-annotation.adoc

Example of a batch listener that receives lists of headers (key, partition, topic, offset) parallel to the payloads.

```java
@KafkaListener(id = "list", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List<String> list,
        @Header(KafkaHeaders.RECEIVED_KEY) List<Integer> keys,
        @Header(KafkaHeaders.RECEIVED_PARTITION) List<Integer> partitions,
        @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics,
        @Header(KafkaHeaders.OFFSET) List<Long> offsets) {
    ...
}
```

--------------------------------

### Configuring ContainerGroupSequencer with @KafkaListener groups

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/sequencing.adoc

This example demonstrates defining multiple @KafkaListener methods assigned to groups 'g1' and 'g2', and configuring a ContainerGroupSequencer bean to manage their startup sequence.

```java
@KafkaListener(id = "listen1", topics = "topic1", containerGroup = "g1", concurrency = "2")
public void listen1(String in) {
}

@KafkaListener(id = "listen2", topics = "topic2", containerGroup = "g1", concurrency = "2")
public void listen2(String in) {
}

@KafkaListener(id = "listen3", topics = "topic3", containerGroup = "g2", concurrency = "2")
public void listen3(String in) {
}

@KafkaListener(id = "listen4", topics = "topic4", containerGroup = "g2", concurrency = "2")
public void listen4(String in) {
}

@Bean
ContainerGroupSequencer sequencer(KafkaListenerEndpointRegistry registry) {
    return new ContainerGroupSequencer(registry, 5000, "g1", "g2");
}
```

--------------------------------

### Example output of linear commit history

Source: https://github.com/spring-projects/spring-kafka/blob/main/CONTRIBUTING.md

This output demonstrates a clean, linear commit history without intersecting lines.

```text
* e11821842f22b1a9461ab2209c0be7cd47f5c25e GH-123: Add feature X
* 59710724e5df4414139f67145152fc943d45baf7 GH-101: Fix NPE in KafkaTemplate
* d0f6e61308a020bdeb06992ece63fa3c4bad99d7 GH-99: Improve error handler backoff
```

--------------------------------

### Use the custom meta-annotation

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/listener-meta.adoc

This example shows how to apply the custom MyThreeConsumersListener annotation to a method, providing values for the aliased attributes.

```java
@MyThreeConsumersListener(id = "my.group", topics = "my.topic")
public void listen1(String in) {
    ...
}
```

--------------------------------

### Use AdminClient directly with KafkaAdmin

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/configuring-topics.adoc

Shows how to use the AdminClient directly for more advanced administrative features. The example autowires a KafkaAdmin and creates an AdminClient from its configuration properties, then closes it.

```java
@Autowired
private KafkaAdmin admin;

...

    AdminClient client = AdminClient.create(admin.getConfigurationProperties());
    ...
    client.close();
```

--------------------------------

### Configuration Properties for Chained Transactions Example

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/tips.adoc

These properties configure the Spring Boot application for the chained transactions example. They set up the MySQL datasource, Kafka consumer settings (auto-offset-reset=earliest, enable-auto-commit=false, isolation.level=read_committed), and producer transaction-id-prefix. Commented logging properties are included for tracing transactions and JDBC.

```properties
spring.datasource.url=jdbc:mysql://localhost/integration?serverTimezone=UTC
spring.datasource.username=root
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver

spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.properties.isolation.level=read_committed

spring.kafka.producer.transaction-id-prefix=tx-

#logging.level.org.springframework.transaction=trace
#logging.level.org.springframework.kafka.transaction=debug
#logging.level.org.springframework.jdbc=debug
```

--------------------------------

### Migrate @Backoff to @BackOff with duration and placeholder support

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/appendix/change-history.adoc

Migration example showing the rename from @Backoff to @BackOff, plus new duration string and property placeholder options.

```java
// Before
@RetryableTopic(backoff = @Backoff(delay = 2000, maxDelay = 10000, multiplier = 2))

// After
@RetryableTopic(backOff = @BackOff(delay = 2000, maxDelay = 10000, multiplier = 2))

// With new duration format support
@RetryableTopic(backOff = @BackOff(delayString = "2s", maxDelayString = "10s", multiplier = 2))

// With property placeholders
@RetryableTopic(backOff = @BackOff(delayString = "${retry.delay}", multiplierString = "${retry.multiplier}"))
```

--------------------------------

### Basic Kafka Streams API usage with StreamsBuilder and KafkaStreams

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/streams.adoc

Shows the standard Kafka Streams API pattern: build a topology with StreamsBuilder, configure with StreamsConfig, create a KafkaStreams instance, then start and close it. All streams defined by a single StreamsBuilder are started and stopped together, and a closed KafkaStreams cannot be restarted.

```java
// Use the builders to define the actual processing topology, e.g. to specify
// from which input topics to read, which stream operations (filter, map, etc.)
// should be called, and so on.

StreamsBuilder builder = ...;  // when using the Kafka Streams DSL

// Use the configuration to tell your application where the Kafka cluster is,
// which serializers/deserializers to use by default, to specify security settings,
// and so on.
StreamsConfig config = ...;

KafkaStreams streams = new KafkaStreams(builder, config);

// Start the Kafka Streams instance
streams.start();

// Stop the Kafka Streams instance
streams.close();
```

--------------------------------

### Manual Reply Message Header Configuration

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/annotation-send-to.adoc

Example of a listener method returning a Message object with manually configured headers for a ReplyingKafkaTemplate request.

```java
@KafkaListener(id = "messageReturned", topics = "someTopic")
public Message<?> listen(String in, @Header(KafkaHeaders.REPLY_TOPIC) byte[] replyTo,
        @Header(KafkaHeaders.CORRELATION_ID) byte[] correlation) {
    return MessageBuilder.withPayload(in.toUpperCase())
            .setHeader(KafkaHeaders.TOPIC, replyTo)
            .setHeader(KafkaHeaders.KEY, 42)
            .setHeader(KafkaHeaders.CORRELATION_ID, correlation)
            .setHeader("someOtherHeader", "someValue")
            .build();
}
```

--------------------------------

### KafkaMessageListenerContainer with explicit topics

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/message-listener-container.adoc

Example of creating a KafkaMessageListenerContainer with a ContainerProperties instance that specifies topics and a MessageListener.

```APIDOC
## KafkaMessageListenerContainer with explicit topics

### Description
Creates a KafkaMessageListenerContainer that listens to the specified topics. The container uses the provided ContainerProperties and ConsumerFactory.

### Method
Constructor

### Endpoint
`new KafkaMessageListenerContainer<>(ConsumerFactory<K, V> consumerFactory, ContainerProperties containerProperties)`

### Parameters
- **consumerFactory** (ConsumerFactory<K, V>) - Required - The consumer factory used to create Kafka consumers.
- **containerProperties** (ContainerProperties) - Required - Configuration properties for the container, including topics and message listener.

### Request Example
```java
ContainerProperties containerProps = new ContainerProperties("topic1", "topic2");
containerProps.setMessageListener(new MessageListener<Integer, String>() {
    // message listener implementation
});
DefaultKafkaConsumerFactory<Integer, String> cf =
                new DefaultKafkaConsumerFactory<>(consumerProps());
KafkaMessageListenerContainer<Integer, String> container =
                new KafkaMessageListenerContainer<>(cf, containerProps);
return container;
```

### Response
No direct response; the container is started asynchronously.
```

--------------------------------

### Apply initial offset to all assigned partitions with wildcard

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/listener-annotation.adoc

Starting with version 2.5.5, you can apply an initial offset to all assigned partitions using the '*' wildcard in the partitionOffsets attribute. There must only be one @PartitionOffset with the wildcard in each @TopicPartition.

```java
@KafkaListener(id = "thing3", topicPartitions =
        { @TopicPartition(topic = "topic1", partitions = { "0", "1" },
             partitionOffsets = @PartitionOffset(partition = "*", initialOffset = "0"))
        })
public void listen(ConsumerRecord<?, ?> record) {
    ...
}
```

--------------------------------

### Receive list of payloads in batch listener

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/listener-annotation.adoc

Example of a batch listener that receives a List<String> of payloads. The container factory must be configured with batchListener=true.

```java
@KafkaListener(id = "list", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List<String> list) {
    ...
}
```

--------------------------------

### Application console output

Source: https://github.com/spring-projects/spring-kafka/blob/main/samples/sample-04/README.adoc

Example console output showing the processing of a 'good' record and the sequential retry attempts for a 'fail' record across different retry topics and the DLT.

```text
2021-05-11 16:17:55.372  INFO 75520 --- [ fooGroup-0-C-1] com.example.Application : Received: fail from topic4 @ xx

2021-05-11 16:17:55.399  INFO 75520 --- [ fooGroup-0-C-1] com.example.Application : Received: good from topic4 @ xx

2021-05-11 16:17:57.384  INFO 75520 --- [etry-2000-0-C-1] com.example.Application : Received: fail from topic4-retry-2000 @ xx

2021-05-11 16:18:01.390  INFO 75520 --- [etry-4000-0-C-1] com.example.Application : Received: fail from topic4-retry-4000 @ xx

2021-05-11 16:18:09.399  INFO 75520 --- [etry-8000-0-C-1] com.example.Application : Received: fail from topic4-retry-8000 @ xx

2021-05-11 16:18:19.407  INFO 75520 --- [try-10000-0-C-1] com.example.Application : Received: fail from topic4-retry-10000 @ xx

2021-05-11 16:18:19.918  INFO 75520 --- [Group-dlt-0-C-1] com.example.Application : DLT Received: fail from topic4-dlt @ xx
```

--------------------------------

### Use JacksonJsonSerde fluent API

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/streams.adoc

Since version 2.3, the fluent API simplifies configuration of JacksonJsonSerde. This example uses forKeys() and noTypeInfo() for key and value serdes.

```java
stream.through(
    new JacksonJsonSerde<>(MyKeyType.class)
        .forKeys()
        .noTypeInfo(),
    new JacksonJsonSerde<>(MyValueType.class)
        .noTypeInfo(),
    "myTypes");

```

--------------------------------

### RoutingKafkaTemplate Example with ByteArraySerializer

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/sending-messages.adoc

Spring Boot application showing how to configure a RoutingKafkaTemplate with a map of patterns to ProducerFactory instances. The routing template does not support transactions, execute, flush, or metrics operations.

```java
@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @Bean
    public RoutingKafkaTemplate routingTemplate(GenericApplicationContext context,
            ProducerFactory<Object, Object> pf) {

        // Clone the PF with a different Serializer, register with Spring for shutdown
        Map<String, Object> configs = new HashMap<>(pf.getConfigurationProperties());
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
        DefaultKafkaProducerFactory<Object, Object> bytesPF = new DefaultKafkaProducerFactory<>(configs);
        context.registerBean("bytesPF", DefaultKafkaProducerFactory.class, () -> bytesPF);
```

--------------------------------

### Build and Test with Gradle

Source: https://github.com/spring-projects/spring-kafka/blob/main/CONTRIBUTING.md

Common Gradle wrapper commands for building the project, running tests, and generating documentation.

```bash
# Full build with tests
./gradlew build

# Run tests only
./gradlew test

# Run tests for a specific module
./gradlew :spring-kafka:test

# Run a single test class
./gradlew :spring-kafka:test --tests "org.springframework.kafka.listener.ConcurrentMessageListenerContainerTests"

# Publish to local Maven repository
./gradlew publishToMavenLocal

# Generate API Javadoc (output: build/api)
./gradlew api

# Generate reference documentation (output: spring-kafka-docs/build/site)
./gradlew antora

# Build distribution zips (output: build/distributions)
./gradlew dist
```

--------------------------------

### Blocking (Sync) Send Example

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/sending-messages.adoc

Example of sending a message synchronously by calling get() with a timeout. The cause of ExecutionException is KafkaProducerException with the producerRecord property.

```java
public void sendToKafka(final MyOutputData data) {
    final ProducerRecord<String, String> record = createRecord(data);

    try {
        template.send(record).get(10, TimeUnit.SECONDS);
        handleSuccess(data);
    }
    catch (ExecutionException e) {
        handleFailure(data, record, e.getCause());
    }
    catch (TimeoutException | InterruptedException e) {
        handleFailure(data, record, e);
    }
}
```

--------------------------------

### Configure DeadLetterPublishingRecoverer with Multiple KafkaTemplates

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/annotation-error-handling.adoc

Example of configuring the publisher with multiple KafkaTemplates that use different serializers, useful when publishing byte[] from a DeserializationException and values using a different serializer from successfully deserialized records.

```APIDOC
## DeadLetterPublishingRecoverer - Multiple KafkaTemplates Configuration

### Description
Configure the DeadLetterPublishingRecoverer with multiple KafkaTemplates to support different value types. The publisher uses the map keys to locate a template suitable for the value() about to be published. A LinkedHashMap is recommended so keys are examined in order.

### Method
Constructor

### Signature
`DeadLetterPublishingRecoverer(Map<Class<?>, KafkaOperations<?, ?>> templates)`

### Parameters
- **templates** (Map<Class<?>, KafkaOperations<?, ?>>) - Map of value types to KafkaOperations. Keys are examined in order.

### Example
```java
@Bean
public DeadLetterPublishingRecoverer publisher(KafkaTemplate<?, ?> stringTemplate,
        KafkaTemplate<?, ?> bytesTemplate) {
    Map<Class<?>, KafkaOperations<?, ?>> templates = new LinkedHashMap<>();
    templates.put(String.class, stringTemplate);
    templates.put(byte[].class, bytesTemplate);
    return new DeadLetterPublishingRecoverer(templates);
}
```

### Behavior
- When publishing null values with multiple templates, the recoverer looks for a template for the Void class; if none, the first template from values().iterator() is used.
- Since version 2.3, when used with ErrorHandlingDeserializer, the publisher restores the record value() to the original value that failed to be deserialized.
- Since version 2.7, if both key and value fail deserialization, the original values of both are populated in the record sent to the DLT.
```

--------------------------------

### Seek to Beginning with ConsumerSeekAware

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/seek.adoc

Example application that sends records to a topic and uses a listener implementing ConsumerSeekAware to seek to the beginning of all assigned partitions when the user presses Enter. The listener stores callbacks per partition and uses seekToBeginning on each callback.

```java
public static void main(String[] args) {
    SpringApplication.run(SeekExampleApplication.class, args);
}

@Bean
public ApplicationRunner runner(Listener listener, KafkaTemplate<String, String> template) {
    return args -> {
        IntStream.range(0, 10).forEach(i -> template.send(
            new ProducerRecord<>("seekExample", i % 3, "foo", "bar")));
        while (true) {
            System.in.read();
            listener.seekToStart();
        }
    };
}

@Bean
public NewTopic topic() {
    return new NewTopic("seekExample", 3, (short) 1);
}

}
```

```java
@Component
class Listener implements ConsumerSeekAware {

    private static final Logger logger = LoggerFactory.getLogger(Listener.class);

    private final ThreadLocal<ConsumerSeekCallback> callbackForThread = new ThreadLocal<>();

    private final Map<TopicPartition, ConsumerSeekCallback> callbacks = new ConcurrentHashMap<>();

    @Override
    public void registerSeekCallback(ConsumerSeekCallback callback) {
        this.callbackForThread.set(callback);
    }

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        assignments.keySet().forEach(tp -> this.callbacks.put(tp, this.callbackForThread.get()));
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        partitions.forEach(tp -> this.callbacks.remove(tp));
        this.callbackForThread.remove();
    }

    @Override
    public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
    }

    @KafkaListener(id = "seekExample", topics = "seekExample", concurrency = "3")
    public void listen(ConsumerRecord<String, String> in) {
        logger.info(in.toString());
    }

    public void seekToStart() {
        this.callbacks.forEach((tp, callback) -> callback.seekToBeginning(tp.topic(), tp.partition()));
    }

}
```

--------------------------------

### Build reference documentation

Source: https://github.com/spring-projects/spring-kafka/blob/main/README.md

Builds the reference documentation; results are placed in spring-kafka-docs/build/site.

```bash
./gradlew antora
```

--------------------------------

### Configure batch listener container factory

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/listener-annotation.adoc

Example of configuring a KafkaListenerContainerFactory bean with batchListener set to true. Starting with version 2.8, you can override this property using the batch property on @KafkaListener. Starting with version 2.9.6, separate setters for recordMessageConverter and batchMessageConverter are available.

```java
@Bean
public KafkaListenerContainerFactory<?> batchFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true);
   return factory;
}
```

--------------------------------

### Add ContainerCustomizer to factory

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/container-factory.adoc

Starting with version 2.3.4, a ContainerCustomizer can be added to the factory to further configure each container after creation. The example shows setting a customizer that customizes the container.

```java
@Bean
public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    ...
    factory.setContainerCustomizer(container -> { /* customize the container */ });
    return factory;
}
```

--------------------------------

### Use broker defaults for partitions and replicas

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/configuring-topics.adoc

Shows how to omit partitions() and/or replicas() so broker defaults are applied. Requires broker version 2.4.0 or higher (KIP-464).

```java
include::{java-examples}/topics/Config.java[tag=brokerProps]
```

```kotlin
include::{kotlin-examples}/topics/Config.kt[tag=brokerProps]
```

--------------------------------

### Define ContainerPostProcessor bean and use in @KafkaListener

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/container-factory.adoc

Starting with version 3.1, a ContainerPostProcessor can be applied to a single listener by specifying its bean name in the @KafkaListener annotation. The example shows defining the bean and using the containerPostProcessor attribute.

```java
@Bean
public ContainerPostProcessor<String, String, AbstractMessageListenerContainer<String, String>> customContainerPostProcessor() {
    return container -> { /* customize the container */ };
}

...

@KafkaListener(..., containerPostProcessor="customContainerPostProcessor", ...)
```

--------------------------------

### Create a meta-annotation with @RetryableTopic

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/retrytopic/retry-config.adoc

Starting with version 3.0, @RetryableTopic can be used as a meta-annotation on custom annotations. This example defines a custom annotation that aliases the concurrency attribute to a new attribute named 'parallelism' with a default value of '3'.

```java
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@RetryableTopic
static @interface MetaAnnotatedRetryableTopic {

    @AliasFor(attribute = "concurrency", annotation = RetryableTopic.class)
    String parallelism() default "3";

}
```

--------------------------------

### Configure Exception-Based DLT Routing with Annotations

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/retrytopic/features.adoc

Starting with version 3.2.0, route messages to custom DLTs based on exception types. The suffix is placed before the general dltTopicSuffix in the custom DLT name. For example, a DeserializationException routes to my-annotated-topic-deserialization-dlt instead of my-annotated-topic-dlt.

```java
@RetryableTopic(exceptionBasedDltRouting = {
    @ExceptionBasedDltDestination(
        suffix = "-deserialization", exceptions = {DeserializationException.class}
    )}
)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    // ... message processing
}
```

--------------------------------

### Topic configuration options for @KafkaListener

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/listener-annotation.adoc

Shows three ways to configure topics: by name, by pattern, and by explicit partitions. Only one of these must be configured.

```java
@KafkaListener(id = "myListener", topics = "myTopic")
public void listen(String data) {
    ...
}

@KafkaListener(id = "myListener", topicPattern = "my.*")
public void listen(String data) {
    ...
}

@KafkaListener(id = "myListener", topicPartitions = { @TopicPartition(topic = "myTopic", partitions = { "0", "1" })})
public void listen(String data) {
    ...
}
```

--------------------------------

### Chaining Database and Kafka Transactions with @Transactional

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/tips.adoc

This Spring Boot application demonstrates chaining database and Kafka transactions. The listener container starts the Kafka transaction and the @Transactional annotation starts the DB transaction; the DB transaction commits first, and if the Kafka transaction fails, the record is redelivered, so the DB update should be idempotent. The example includes a runner that sends a message, a listener that processes and sends to another topic, and topic definitions.

```java
@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @Bean
    public ApplicationRunner runner(KafkaTemplate<String, String> template) {
        return args -> template.executeInTransaction(t -> t.send("topic1", "test"));
    }

    @Bean
    public DataSourceTransactionManager dstm(DataSource dataSource) {
        return new DataSourceTransactionManager(dataSource);
    }

    @Component
    public static class Listener {

        private final JdbcTemplate jdbcTemplate;

        private final KafkaTemplate<String, String> kafkaTemplate;

        public Listener(JdbcTemplate jdbcTemplate, KafkaTemplate<String, String> kafkaTemplate) {
            this.jdbcTemplate = jdbcTemplate;
            this.kafkaTemplate = kafkaTemplate;
        }

        @KafkaListener(id = "group1", topics = "topic1")
        @Transactional("dstm")
        public void listen1(String in) {
            this.kafkaTemplate.send("topic2", in.toUpperCase());
            this.jdbcTemplate.execute("insert into mytable (data) values ('" + in + "')");
        }

        @KafkaListener(id = "group2", topics = "topic2")
        public void listen2(String in) {
            System.out.println(in);
        }

    }

    @Bean
    public NewTopic topic1() {
        return TopicBuilder.name("topic1").build();
    }

    @Bean
    public NewTopic topic2() {
        return TopicBuilder.name("topic2").build();
    }

}
```

--------------------------------

### Build complete distribution

Source: https://github.com/spring-projects/spring-kafka/blob/main/README.md

Builds the complete distribution including -dist, -docs, and -schema zip files; results are placed in build/distributions.

```bash
./gradlew dist
```

--------------------------------

### @KafkaListener returning Message<?> with automatic headers (2.5+)

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/sending-messages.adoc

Starting with version 2.5, the framework detects missing headers and populates them with the topic from @SendTo or the incoming KafkaHeaders.REPLY_TOPIC header, and echoes the incoming CORRELATION_ID and REPLY_PARTITION headers. This example relies on the default REPLY_TOPIC header and only sets the key.

```java
@KafkaListener(id = "requestor", topics = "request")
@SendTo  // default REPLY_TOPIC header
public Message<?> messageReturn(String in) {
    return MessageBuilder.withPayload(in.toUpperCase())
            .setHeader(KafkaHeaders.KEY, 42)
            .build();
}
```

--------------------------------

### Combine @KafkaListener and @EventListener for Idle Events

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/events.adoc

Example listener class that uses @KafkaListener to consume messages and @EventListener with a condition to handle ListenerContainerIdleEvent. The condition narrows events to containers whose listener ID starts with 'qux-', because concurrent containers are named 'id-n'. Event listeners receive events for all containers, so filtering by listener ID is necessary. Avoid calling container.stop() on the listener thread; hand off to another thread instead.

```java
public class Listener {

    @KafkaListener(id = "qux", topics = "annotated")
    public void listen4(@Payload String foo, Acknowledgment ack) {
        ...
    }

    @EventListener(condition = "event.listenerId.startsWith('qux-')")
    public void eventHandler(ListenerContainerIdleEvent event) {
        ...
    }

}
```

--------------------------------

### Seek to Timestamp with AbstractConsumerSeekAware Convenience Methods

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/seek.adoc

Example showing a listener extending AbstractConsumerSeekAware and a separate bean calling seekToTimestamp to seek all assigned partitions to the offset represented by a timestamp. Version 2.6 added seekToBeginning(), seekToEnd(), and seekToTimestamp(long).

```java
public class MyListener extends AbstractConsumerSeekAware {

    @KafkaListener(...)
    void listen(...) {
        ...
    }
}
```

```java
public class SomeOtherBean {

    MyListener listener;

    ...

    void someMethod() {
        this.listener.seekToTimestamp(System.currentTimeMillis() - 60_000);
    }

}
```

--------------------------------

### Configure BatchToRecordAdapter with TestListener

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/transactions.adoc

This example shows a test listener and configuration that uses a BatchToRecordAdapter to handle batch processing one record at a time. The listener throws an exception for 'bar' to simulate a partial failure. The configuration sets batchListener to true and provides a DefaultBatchToRecordAdapter that captures the failed record.

```java
public static class TestListener {

    final List<String> values = new ArrayList<>();

    @KafkaListener(id = "batchRecordAdapter", topics = "test")
    public void listen(String data) {
        values.add(data);
        if ("bar".equals(data)) {
            throw new RuntimeException("reject partial");
        }
    }

}

@Configuration
@EnableKafka
public static class Config {

    ConsumerRecord<?, ?> failed;

    @Bean
    public TestListener test() {
        return new TestListener();
    }

    @Bean
    public ConsumerFactory<?, ?> consumerFactory() {
        return mock(ConsumerFactory.class);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(consumerFactory());
        factory.setBatchListener(true);
        factory.setBatchToRecordAdapter(new DefaultBatchToRecordAdapter<>((record, ex) ->  {
            this.failed = record;
        }));
        return factory;
    }

}
```

--------------------------------

### Build API Javadoc

Source: https://github.com/spring-projects/spring-kafka/blob/main/README.md

Builds the API Javadoc; results are placed in build/api.

```bash
./gradlew api
```

--------------------------------

### Configure KafkaTemplate with producer factory

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/sending-messages.adoc

Shows how to define a ProducerFactory, producer configs, and KafkaTemplate beans. The producer configs set bootstrap servers and serializers. Avoid using @PostConstruct for sending when relying on automatic topic creation; use ApplicationListener<ContextRefreshedEvent> or SmartLifecycle instead.

```java
@Bean
public ProducerFactory<Integer, String> producerFactory() {
    return new DefaultKafkaProducerFactory<>(producerConfigs());
}

@Bean
public Map<String, Object> producerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    // See {kafka-url}/documentation/#producerconfigs for more properties
    return props;
}

@Bean
public KafkaTemplate<Integer, String> kafkaTemplate() {
    return new KafkaTemplate<Integer, String>(producerFactory());
}
```

--------------------------------

### IgnoreEmptyBatchRecordFilterStrategy example

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/filtering.adoc

This example shows a RecordFilterStrategy that always filters out all records and returns true from ignoreEmptyBatch(), so the @KafkaListener method is never invoked. The comment indicates that 'ignoreEmptyBatchRecordFilterStrategy' is the bean name of the strategy instance.

```java
public class IgnoreEmptyBatchRecordFilterStrategy implements RecordFilterStrategy {
    ...
    @Override
    public List<ConsumerRecord<String, String>> filterBatch(
            List<ConsumerRecord<String, String>> consumerRecords) {
        return List.of();
    }

    @Override
    public boolean ignoreEmptyBatch() {
        return true;
    }
}

// NOTE: ignoreEmptyBatchRecordFilterStrategy is bean name of IgnoreEmptyBatchRecordFilterStrategy instance.
@KafkaListener(id = "filtered", topics = "topic", filter = "ignoreEmptyBatchRecordFilterStrategy")
public void listen(List<Thing> things) {
    ...
}
```

--------------------------------

### @KafkaListener with explicit topic partitions and offsets

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/listener-annotation.adoc

Configures a listener with explicit topics, partitions, and initial offsets using @TopicPartition and @PartitionOffset. Partitions and partitionOffsets cannot both be specified for the same topic.

```java
@KafkaListener(id = "thing2", topicPartitions =
        { @TopicPartition(topic = "topic1", partitions = { "0", "1" }),
          @TopicPartition(topic = "topic2", partitions = "0",
             partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
        })
public void listen(ConsumerRecord<?, ?> record) {
    ...
}
```

--------------------------------

### NotIgnoreEmptyBatchRecordFilterStrategy example

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/filtering.adoc

This example shows a RecordFilterStrategy that always filters out all records but returns false from ignoreEmptyBatch(), so the @KafkaListener method is always invoked even when all records are filtered out. The comment indicates that 'notIgnoreEmptyBatchRecordFilterStrategy' is the bean name of the strategy instance.

```java
public class NotIgnoreEmptyBatchRecordFilterStrategy implements RecordFilterStrategy {
    ...
    @Override
    public List<ConsumerRecord<String, String>> filterBatch(
            List<ConsumerRecord<String, String>> consumerRecords) {
        return List.of();
    }

    @Override
    public boolean ignoreEmptyBatch() {
        return false;
    }
}

// NOTE: notIgnoreEmptyBatchRecordFilterStrategy is bean name of NotIgnoreEmptyBatchRecordFilterStrategy instance.
@KafkaListener(id = "filtered", topics = "topic", filter = "notIgnoreEmptyBatchRecordFilterStrategy")
public void listen(List<Thing> things) {
    ...
}
```

--------------------------------

### Global Embedded Kafka Broker Log Messages

Source: https://github.com/spring-projects/spring-kafka/blob/main/samples/sample-05/README.adoc

Log output from the GlobalEmbeddedKafkaTestExecutionListener showing the broker start and stop messages. The start message appears at the beginning of the test plan and the stop message at the end before reporting failures.

```text
11:03:44.383 [main] INFO  o.s.k.t.j.GlobalEmbeddedKafkaTestExecutionListener - Started global Embedded Kafka on: 127.0.0.1:53671
...
11:03:48.439 [main] INFO  o.s.k.t.j.GlobalEmbeddedKafkaTestExecutionListener - Stopped global Embedded Kafka.
```

--------------------------------

### Use EmbeddedKafkaHolder in test class to add topics and get broker

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/testing.adoc

In each test class, add topics via the shared broker and obtain the broker instance. If not using Spring Boot, get bootstrap servers with broker.getBrokersAsString().

```java
static {
    EmbeddedKafkaHolder.getEmbeddedKafka().addTopics("topic1", "topic2");
}

private static final EmbeddedKafkaBroker broker = EmbeddedKafkaHolder.getEmbeddedKafka();
```

--------------------------------

### Define ValidatedClass with Constraints

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/validation.adoc

Example of a POJO using JSR-303/JSR-380 validation annotations.

```java
public static class ValidatedClass {

  @Max(10)
  private int bar;

  public int getBar() {
    return this.bar;
  }

  public void setBar(int bar) {
    this.bar = bar;
  }

}
```

--------------------------------

### Create ToStringSerializer and ParseStringDeserializer

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/serdes.adoc

Shows how to instantiate ToStringSerializer and ParseStringDeserializer with a parser method reference.

```java
ToStringSerializer<Thing> thingSerializer = new ToStringSerializer<>();
//...
ParseStringDeserializer<Thing> deserializer = new ParseStringDeserializer<>(Thing::parse);
```

--------------------------------

### Example of unacceptable Git author format

Source: https://github.com/spring-projects/spring-kafka/blob/main/CONTRIBUTING.md

This format is not acceptable for pull requests.

```text
Author: Nickname <user@mail.com>
```

--------------------------------

### Using KafkaTestUtils.getSingleRecord

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/testing.adoc

Example of sending a record and then retrieving it with getSingleRecord. The ellipsis indicates omitted code.

```java
...
template.sendDefault(0, 2, "bar");
ConsumerRecord<Integer, String> received = KafkaTestUtils.getSingleRecord(consumer, "topic");
...
```

--------------------------------

### Spring Boot application using waitForAssignment with ReplyingKafkaTemplate

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/sending-messages.adoc

Full Spring Boot application showing how to use waitForAssignment to wait for the reply container to initialize before sending a request. The wait duration must be greater than the container's pollTimeout when using manual partition assignment. Uses Boot's auto-configured container factory to create the reply container.

```java
@SpringBootApplication
public class KRequestingApplication {

    public static void main(String[] args) {
        SpringApplication.run(KRequestingApplication.class, args).close();
    }

    @Bean
    public ApplicationRunner runner(ReplyingKafkaTemplate<String, String, String> template) {
        return args -> {
            if (!template.waitForAssignment(Duration.ofSeconds(10))) {
                throw new IllegalStateException("Reply container did not initialize");
            }
            ProducerRecord<String, String> record = new ProducerRecord<>("kRequests", "foo");
            RequestReplyFuture<String, String, String> replyFuture = template.sendAndReceive(record);
            SendResult<String, String> sendResult = replyFuture.getSendFuture().get(10, TimeUnit.SECONDS);
            System.out.println("Sent ok: " + sendResult.getRecordMetadata());
            ConsumerRecord<String, String> consumerRecord = replyFuture.get(10, TimeUnit.SECONDS);
            System.out.println("Return value: " + consumerRecord.value());
        };
    }

    @Bean
    public ReplyingKafkaTemplate<String, String, String> replyingTemplate(
            ProducerFactory<String, String> pf,
            ConcurrentMessageListenerContainer<String, String> repliesContainer) {

        return new ReplyingKafkaTemplate<>(pf, repliesContainer);
    }

    @Bean
    public ConcurrentMessageListenerContainer<String, String> repliesContainer(
            ConcurrentKafkaListenerContainerFactory<String, String> containerFactory) {

        ConcurrentMessageListenerContainer<String, String> repliesContainer =
                containerFactory.createContainer("kReplies");
        repliesContainer.getContainerProperties().setGroupId("repliesGroup");
        repliesContainer.setAutoStartup(false);
        return repliesContainer;
    }

    @Bean
    public NewTopic kRequests() {
        return TopicBuilder.name("kRequests")
            .partitions(10)
            .replicas(2)
            .build();
    }

    @Bean
    public NewTopic kReplies() {
        return TopicBuilder.name("kReplies")
            .partitions(10)
            .replicas(2)
            .build();
    }

}
```

--------------------------------

### Filter MessageListenerContainers dynamically

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/kafkalistener-lifecycle.adoc

Examples of using getListenerContainersMatching with Predicate or BiPredicate to filter containers by ID or state.

```java
// Prefix matching (Predicate<String>)
Collection<MessageListenerContainer> filteredContainers =
    registry.getListenerContainersMatching(id -> id.startsWith("productListener-retry-"));

// Regex matching (Predicate<String>)
Collection<MessageListenerContainer> regexFilteredContainers =
    registry.getListenerContainersMatching(myPattern::matches);

// Pre-built Set of IDs (Predicate<String>)
Collection<MessageListenerContainer> setFilteredContainers =
    registry.getListenerContainersMatching(myIdSet::contains);

// Advanced Filtering: ID prefix and running state (BiPredicate<String, MessageListenerContainer>)
Collection<MessageListenerContainer> advancedFilteredContainers =
    registry.getListenerContainersMatching(
        (id, container) -> id.startsWith("specificPrefix-") && container.isRunning()
    );
```

--------------------------------

### Customizing KafkaTemplate for reply partitioning or keys

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/annotation-send-to.adoc

Shows how to create a custom KafkaTemplate subclass to handle specific partitioning or key generation logic for replies.

```java
@Bean
public KafkaTemplate<String, String> myReplyingTemplate() {
    return new KafkaTemplate<String, String>(producerFactory()) {
```

--------------------------------

### Generate IntelliJ IDEA metadata with ./gradlew idea

Source: https://github.com/spring-projects/spring-kafka/blob/main/README.md

Run this Gradle command from the project root to generate IntelliJ IDEA metadata files (.iml and .ipr).

```bash
./gradlew idea
```

--------------------------------

### Add Not-Retryable Exception to DefaultAfterRollbackProcessor

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/annotation-error-handling.adoc

Example bean that adds IllegalArgumentException to the not-retryable exceptions. The recoverer is a BiConsumer<ConsumerRecord<?, ?>, Exception>.

```java
@Bean
public DefaultAfterRollbackProcessor errorHandler(BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer) {
    DefaultAfterRollbackProcessor processor = new DefaultAfterRollbackProcessor(recoverer);
    processor.addNotRetryableException(IllegalArgumentException.class);
    return processor;
}
```

--------------------------------

### Apply initial offset to a partition range

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/listener-annotation.adoc

The same technique can be used when specifying initial offsets. The initial offset will be applied to all partitions in the range (e.g., 0-5 applies to all 6 partitions).

```java
@KafkaListener(id = "thing3", topicPartitions =
        { @TopicPartition(topic = "topic1",
             partitionOffsets = @PartitionOffset(partition = "0-5", initialOffset = "0"))
        })
public void listen(ConsumerRecord<?, ?> record) {
    ...
}
```

--------------------------------

### Clone and build Spring Kafka

Source: https://github.com/spring-projects/spring-kafka/blob/main/README.md

Clones the Spring for Apache Kafka repository and builds it from source. Java 17 or later is recommended. If out-of-memory errors occur, adjust the org.gradle.jvmargs property in gradle.properties.

```bash
git clone git://github.com/spring-projects/spring-kafka.git
cd spring-kafka
./gradlew build
```

--------------------------------

### AggregatingReplyingKafkaTemplate sendAndReceive Example

Source: https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/sending-messages.adoc

Shows how to create an AggregatingReplyingKafkaTemplate with a release strategy and send a request, waiting for the send and the aggregated reply. The returned ConsumerRecord has a collection of reply records; the outer record is synthesized by the template. Requires the listener container to use AckMode.MANUAL or MANUAL_IMMEDIATE with enable.auto.commit=false.

```java
AggregatingReplyingKafkaTemplate<Integer, String, String> template =
        new AggregatingReplyingKafkaTemplate<>(producerFactory, container,
                        coll -> coll.size() == releaseSize);
...
RequestReplyFuture<Integer, String, Collection<ConsumerRecord<Integer, String>>> future =
        template.sendAndReceive(record);
future.getSendFuture().get(10, TimeUnit.SECONDS); // send ok
ConsumerRecord<Integer, Collection<ConsumerRecord<Integer, String>>> consumerRecord =
        future.get(30, TimeUnit.SECONDS);
```
Tokens:
Raw