### Kafka Listener with Seek to Beginning

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

This example demonstrates a Kafka listener that can seek to the beginning of partitions. It uses a ThreadLocal to manage seek callbacks and a map to store callbacks per topic partition. External calls can trigger seeking to the start.

```java
@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);
    }

}

@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()));
    }

}
```

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

### Install Jars to Local Maven Cache

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

Build and install Spring Kafka jars into your local Maven cache.

```bash
./gradlew install
```

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

### Configure RoutingKafkaTemplate

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 configuring a RoutingKafkaTemplate to select producers at runtime based on topic patterns.

```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);
```

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

### Example Linear Commit History Output

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

An example of the output from `git log --graph --pretty=oneline`, showing a clean, linear commit history.

```text
* c129a02e6c752b49bacd4a445092a44f66c2a1e9 INT-2721 Increase Timers on JDBC Delayer Tests
* 14e556ce23d49229c420632cef608630b1d82e7d INT-2620 Fix Debug Log
* 6140aa7b2cfb6ae309c55a157e94b44e5d0bea4f INT-3037 Fix JDBC MS Discard After Completion
* 077f2b24ea871a3937c513e08241d1c6cb9c9179 Update Spring Social Twitter to 1.0.5
* 6d4f2b46d859c903881a561c35aa28df68f8faf3 INT-3053 Allow task-executor on <reply-listener/>
* 56f9581b85a8a40bbcf2461ffc0753212669a68d Update Spring Social Twitter version to 1.0.4
```

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

### Spring Boot Application for Seek Example

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

A basic Spring Boot application demonstrating the usage of `ConsumerSeekCallback` for seeking operations. This example allows seeking to the beginning by pressing Enter.

```java
@SpringBootApplication
public class SeekExampleApplication {

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

}
```

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

### Seeking to the Beginning of Partitions on Assignment

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

Example of using `ConsumerSeekAware` to seek to the beginning of all assigned partitions when they are assigned. This is useful for compacted topics.

```java
public class MyListener implements ConsumerSeekAware {

    ...

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

}
```

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

### SQL Table Creation

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

Example schema for a database table used in transaction synchronization examples.

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

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

### Programmatically start a KafkaListener container

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

Uses the KafkaListenerEndpointRegistry to manually start a container identified by its ID.

```java
@Autowired
private KafkaListenerEndpointRegistry registry;

...

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

...
```

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

### Embedded Kafka and KafkaTemplate Example

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

Demonstrates setting up an embedded Kafka broker, configuring a KafkaTemplate, sending messages, and verifying received messages using AssertJ conditions.

```java
@EmbeddedKafka(topics = KafkaTemplateTests.TEMPLATE_TOPIC)
public class KafkaTemplateTests {

    public static final String TEMPLATE_TOPIC = "templateTopic";
    public static EmbeddedKafkaBroker embeddedKafka;

    @BeforeAll
	public static void setUp() {
		embeddedKafka = EmbeddedKafkaCondition.getBroker();
	}

    @Test
    public void testTemplate() throws Exception {
        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testT", "false",
            embeddedKafka);
        DefaultKafkaConsumerFactory<Integer, String> cf =
                            new DefaultKafkaConsumerFactory<>(consumerProps);
        ContainerProperties containerProperties = new ContainerProperties(TEMPLATE_TOPIC);
        KafkaMessageListenerContainer<Integer, String> container =
                            new KafkaMessageListenerContainer<>(cf, containerProperties);
        final BlockingQueue<ConsumerRecord<Integer, String>> records = new LinkedBlockingQueue<>();
        container.setupMessageListener(new MessageListener<Integer, String>() {

            @Override
            public void onMessage(ConsumerRecord<Integer, String> record) {
                System.out.println(record);
                records.add(record);
            }

        });
        container.setBeanName("templateTests");
        container.start();
        ContainerTestUtils.waitForAssignment(container,
                            embeddedKafka.getPartitionsPerTopic());
        Map<String, Object> producerProps =
                            KafkaTestUtils.producerProps(embeddedKafka);
        ProducerFactory<Integer, String> pf =
                            new DefaultKafkaProducerFactory<>(producerProps);
        KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
        template.setDefaultTopic(TEMPLATE_TOPIC);
        template.sendDefault("foo");
        assertThat(records.poll(10, TimeUnit.SECONDS), hasValue("foo"));
        template.sendDefault(0, 2, "bar");
        ConsumerRecord<Integer, String> received = records.poll(10, TimeUnit.SECONDS);
        assertThat(received, hasKey(2));
        assertThat(received, hasPartition(0));
        assertThat(received, hasValue("bar"));
        template.send(TEMPLATE_TOPIC, 0, 2, "baz");
        received = records.poll(10, TimeUnit.SECONDS);
        assertThat(received, hasKey(2));
        assertThat(received, hasPartition(0));
        assertThat(received, hasValue("baz"));
    }
}
```

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

### Implement Kafka Application without Spring Boot

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

Example of manual configuration for producer and consumer beans without Spring Boot auto-configuration.

```java
include::{java-examples}/started/noboot/Sender.java[tag=startedNoBootSender]

include::{java-examples}/started/noboot/Listener.java[tag=startedNoBootListener]

include::{java-examples}/started/noboot/Config.java[tag=startedNoBootConfig]
```

```kotlin
include::{kotlin-examples}/started/noboot/Sender.kt[tag=startedNoBootSender]

include::{kotlin-examples}/started/noboot/Listener.kt[tag=startedNoBootListener]

include::{kotlin-examples}/started/noboot/Config.kt[tag=startedNoBootConfig]
```

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

### Example Header Enricher with SpEL

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

Demonstrates adding a literal header and a dynamic header using SpEL expressions with HeaderEnricher. Ensure a new instance of the processor is used for each record.

```java
Map<String, Expression> headers = new HashMap<>();
headers.put("header1", new LiteralExpression("value1"));
SpelExpressionParser parser = new SpelExpressionParser();
headers.put("header2", parser.parseExpression("record.timestamp() + ' @' + record.offset()"));
ProcessorSupplier supplier = () -> new HeaderEnricher<String, String>(headers);
KStream<String, String> stream = builder.stream(INPUT);
stream
        .process(() -> supplier)
        .to(OUTPUT);
```

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

### Configure KafkaListenerContainerFactory

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

Setup required for the listener container, including consumer factory and concurrency settings.

```java
@Configuration
@EnableKafka
public class KafkaConfig {

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
                        kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        ...
        return props;
    }
}
```

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

### Kafka Consumer Log Output

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

Example console output showing message reception and error handling behavior.

```text
2018-11-05 10:03:40.216  INFO 39766 --- [ fooGroup-0-C-1] com.example.Application                  : Received: Foo2 [foo=bar]
```

```text
2018-11-05 10:12:32.552  INFO 41635 --- [ fooGroup-0-C-1] com.example.Application                  : Received: Foo2 [foo=fail]
2018-11-05 10:12:32.561 ERROR 41635 --- [ fooGroup-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception
...
2018-11-05 10:12:33.033  INFO 41635 --- [ fooGroup-0-C-1] com.example.Application                  : Received: Foo2 [foo=fail]
2018-11-05 10:12:33.033 ERROR 41635 --- [ fooGroup-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception
...
2018-11-05 10:12:33.537  INFO 41635 --- [ fooGroup-0-C-1] com.example.Application                  : Received: Foo2 [foo=fail]
2018-11-05 10:12:43.359  INFO 41635 --- [ dltGroup-0-C-1] com.example.Application                  : Received from DLT: {"foo":"fail"}
```

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

### Implementing a Custom ConsumerRebalanceListener

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

An example of implementing `ConsumerAwareRebalanceListener` to store offsets in an external repository after partition revocations and to seek to specific offsets upon partition assignment. This demonstrates how to manage state during rebalancing events.

```java
containerProperties.setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() {

    @Override
    public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        // acknowledge any pending Acknowledgments (if using manual acks)
    }

    @Override
    public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        // ...
        store(consumer.position(partition));
        // ...
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // ...
        consumer.seek(partition, offsetTracker.getOffset() + 1);
        // ...
    }
});
```

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

### Mock Consumer Factory Setup

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

Configure a `MockConsumerFactory` for testing listener containers. It requires a `MockConsumer` instance, initial offsets, and optionally scheduled poll tasks to simulate message reception.

```java
@Bean
ConsumerFactory<String, String> consumerFactory() {
    MockConsumer<String, String> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
    TopicPartition topicPartition0 = new TopicPartition("topic", 0);
    List<TopicPartition> topicPartitions = Collections.singletonList(topicPartition0);
    Map<TopicPartition, Long> beginningOffsets = topicPartitions.stream().collect(
            Collectors.toMap(Function.identity(), tp -> 0L));
    consumer.updateBeginningOffsets(beginningOffsets);
    consumer.schedulePollTask(() -> {
        consumer.addRecord(
                new ConsumerRecord<>("topic", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "test1",
                        new RecordHeaders(), Optional.empty()));
        consumer.addRecord(
                new ConsumerRecord<>("topic", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "test2",
                        new RecordHeaders(), Optional.empty()));
    });
    return new MockConsumerFactory(() -> consumer);
}
```

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

### Configure DefaultAfterRollbackProcessor with Custom Recoverer

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

Configure the DefaultAfterRollbackProcessor to recover records after a specified number of failures using a custom BiConsumer. This example sets up recovery after three tries with no back off.

```java
AfterRollbackProcessor<String, String> processor = 
    new DefaultAfterRollbackProcessor((record, exception) -> {
        // recover after 3 failures, with no back off - e.g. send to a dead-letter topic
    }, new FixedBackOff(0L, 2L));
```

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

### Kafka Listener with Manual Share Container Factory

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

Example of a Kafka listener annotated for the 'order-processing' topic, utilizing a manually configured share Kafka listener container factory.

```java
@KafkaListener(topics = "order-processing", containerFactory = "manualShareKafkaListenerContainerFactory")
```

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

### KafkaTemplate with Different Serializers

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 creating KafkaTemplate instances with different producer configurations, specifically overriding the value serializer from a shared ProducerFactory.

```java
@Bean
public KafkaTemplate<String, String> stringTemplate(ProducerFactory<String, String> pf) {
    return new KafkaTemplate<>(pf);
}

@Bean
public KafkaTemplate<String, byte[]> bytesTemplate(ProducerFactory<String, byte[]> pf) {
    return new KafkaTemplate<>(pf,
            Collections.singletonMap(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class));
}
```

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

### Spring Boot Application for Request-Reply

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

Demonstrates a complete Spring Boot application setup for sending requests and receiving replies using ReplyingKafkaTemplate. Ensure the reply container is initialized before sending requests, especially with auto.offset.reset=latest.

```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();
    }

}

```

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

### Example Output of Pausing and Resuming Kafka Listeners

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

This output shows the sequence of events when pausing and resuming Kafka listener consumers, including partition assignments, messages received, and the generated `ConsumerPausedEvent` and `ConsumerResumedEvent`.

```text
partitions assigned: [pause.resume.topic-1, pause.resume.topic-0]
thing1
pausing
ConsumerPausedEvent [partitions=[pause.resume.topic-1, pause.resume.topic-0]]
resuming
ConsumerResumedEvent [partitions=[pause.resume.topic-1, pause.resume.topic-0]]
thing2
```

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

### Example Inbound-Only Mapper Configuration

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

Demonstrates using the `forInboundOnlyWithMatchers` factory method to create a `JsonKafkaHeaderMapper` that excludes headers starting with 'abc' and includes all others.

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

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

### Build Reference Documentation

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

Generate reference documentation for Spring Kafka. The output will be in 'spring-kafka-docs/build/site'.

```bash
./gradlew antora
```

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

### Chaining Kafka and Database Transactions

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

Example of integrating Kafka transactions with database transactions using Spring Boot. The listener container starts the Kafka transaction, and the @Transactional annotation starts the DB transaction. The DB transaction commits first, ensuring idempotency if the Kafka transaction fails.

```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();
    }

}
```

```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
```

```sql

```

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

### Handle Reply Message without Explicit Headers (Java)

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 Spring Kafka 2.5, if reply topic and correlation ID headers are missing, the framework can populate them automatically. This example relies on the default REPLY_TOPIC header.

```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();
}
```

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

### Configure Topic Assignment Strategies

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

Demonstrates different ways to assign topics, patterns, or specific partitions to a listener.

```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) {
    ...
}
```

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

### Get Host Information for Remote State Store Query

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

Retrieve the `HostInfo` for the Kafka Streams instance that hosts a specific key in a given state store. This is essential for querying remote state stores in a distributed setup.

```java
@Autowired
private KafkaStreamsInteractiveQueryService interactiveQueryService;

HostInfo kafkaStreamsApplicationHostInfo = this.interactiveQueryService.getKafkaStreamsApplicationHostInfo("app-store", 12345, new IntegerSerializer());
```

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

### Configure DefaultErrorHandler with Custom Recoverer and BackOff

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 a DefaultErrorHandler with a custom recoverer for specific error handling logic and a FixedBackOff for retry attempts and delays. This setup is useful for scenarios like sending failed records to a dead-letter topic after a set number of retries.

```java
DefaultErrorHandler errorHandler = 
    new DefaultErrorHandler((record, exception) -> {
        // recover after 3 failures, with no back off - e.g. send to a dead-letter topic
    }, new FixedBackOff(0L, 2L));
```

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

### Build Complete Distribution

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

Create a complete distribution package including '-dist', '-docs', and '-schema' zip files. These are located in 'build/distributions'.

```bash
./gradlew dist
```

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

### Example Spring Bean

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

A simple Spring Bean with a method to be called by Kafka interceptors.

```java
public class SomeBean {

    public void someMethod(String what) {
        System.out.println(what + " in my foo bean");
    }

}
```

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

### Convenience Seek Methods in AbstractConsumerSeekAware

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

Demonstrates the use of convenience methods added to AbstractConsumerSeekAware for seeking to the beginning, end, or a specific timestamp across all assigned partitions.

```java
public class MyListener extends AbstractConsumerSeekAware {

    @KafkaListener(...) // Assuming this is a placeholder for actual listener annotation
    void listen(...)
        ...
    }
}

public class SomeOtherBean {

    MyListener listener;

    ...

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

}
```

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

### Consume from Embedded Kafka Topics

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

Example of using the EmbeddedKafkaBroker to consume messages from all created topics.

```java
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testT", "false", embeddedKafka);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
Consumer<Integer, String> consumer = cf.createConsumer();
embeddedKafka.consumeFromAllEmbeddedTopics(consumer);
```

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

### Retrieve Single Record with KafkaTestUtils

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

Example of using KafkaTestUtils to fetch a single record after sending a message.

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

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

### Initialize String Serializer and Deserializer

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

Create instances of ToStringSerializer and ParseStringDeserializer for entity conversion.

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

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

### Configuring KafkaListenerContainerFactory with ReplyTemplate and ReplyHeadersConfigurer

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

Set up a ConcurrentKafkaListenerContainerFactory with a replyTemplate and a ReplyHeadersConfigurer to manage reply messages and headers.

```java
@Bean
public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory = 
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(cf());
    factory.setReplyTemplate(template());
    factory.setReplyHeadersConfigurer((k, v) -> k.equals("cat"));
    return factory;
}
```

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

### Send Messages Blocking (Sync)

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

Synchronous message sending by invoking get() on the future, with recommended timeout handling.

```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);
    }
}
```

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

### Assign Explicit 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

Configure specific partitions and initial offsets for fine-grained control over message consumption.

```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) {
    ...
}
```

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

### Invalid Default Handler Header Usage

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

Example of an invalid configuration where discrete headers cannot be resolved correctly in a default handler.

```java
@KafkaHandler(isDefault = true)
public void listenDefault(Object object, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
    ...
}
```

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

### Clone Repository with Submodules

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

Clone the Spring Kafka repository and initialize all submodules.

```bash
git clone --recursive git@github.com:<your-github-username>/spring-kafka.git
```

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

### Receive ConsumerRecordMetadata in Kafka 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

Starting with version 2.5, you can receive record metadata in a `ConsumerRecordMetadata` parameter instead of discrete headers.

```java
@KafkaListener(...) public void listen(String str, ConsumerRecordMetadata meta) {
    ...
}
```

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

### JacksonJsonSerde for JSON Serialization

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

Example of using JacksonJsonSerde for JSON serialization/deserialization of message payloads in Kafka Streams. It delegates to Jackson serializers/deserializers.

```java
stream.through(Serdes.Integer(), new JacksonJsonSerde<>(Cat.class), "cats");
```

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

### Clone and Build Spring Kafka

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

Clone the Spring Kafka repository and build the project using Gradle. Java 17 or later is recommended.

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

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

### Get Deserialization Exception from ConsumerRecord Header

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

Retrieve a DeserializationException from a ConsumerRecord's header using `SerializationUtils.getExceptionFromHeader` when consuming a list of ConsumerRecords.

```java
@KafkaListener(id = "kgh2036", topics = "kgh2036")
void listen(List<ConsumerRecord<String, Thing>> in) {
    for (int i = 0; i < in.size(); i++) {
        ConsumerRecord<String, Thing> rec = in.get(i);
        if (rec.value() == null) {
            DeserializationException deserEx = SerializationUtils.getExceptionFromHeader(rec,
                    SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER, this.logger);

```

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

### Configure ContainerGroupSequencer for Sequential Listener Startup

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

Defines Kafka listeners with container groups and initializes the sequencer to manage their startup order.

```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");
}
```

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

### ConsumerSeekAware Interface Methods

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

Implement these methods in your Kafka listener to enable seeking capabilities. The `registerSeekCallback` is called on container start and partition assignment.

```java
void registerSeekCallback(ConsumerSeekCallback callback);

void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback);

void onPartitionsRevoked(Collection<TopicPartition> partitions);

void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback);
```

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

### Configure Blocking and Non-Blocking Retries Together

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

Combine blocking and non-blocking retry configurations. This example shows how to retry certain exceptions using blocking mechanisms, skip others entirely to the DLT, and forward others to non-blocking retry topics.

```java
@Override
protected void configureBlockingRetries(BlockingRetriesConfigurer blockingRetries) {
    blockingRetries
            .retryOn(ShouldRetryOnlyBlockingException.class, ShouldRetryViaBothException.class)
            .backOff(new FixedBackOff(50, 3));
}

@Override
protected void manageNonBlockingFatalExceptions(List<Class<? extends Throwable>> nonBlockingFatalExceptions) {
    nonBlockingFatalExceptions.add(ShouldSkipBothRetriesException.class);
}
```

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

### Apply Initial Offset to All Assigned Partitions

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

Use the wildcard '*' in partitionOffsets to apply an initial offset to all partitions assigned to the listener. Ensure only one wildcard is used per TopicPartition.

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

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

### Dynamically Assign All Partitions with SpEL

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

Use a SpEL expression to dynamically determine and assign all partitions for a topic when the application starts. This is useful for scenarios like loading distributed caches from compacted topics. Ensure appropriate offset reset and ack mode configurations.

```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);
        }
    }

}
```

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

### Global Embedded Kafka Log Output

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

Example log messages indicating the lifecycle of the global Embedded Kafka broker during test execution.

```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.
```

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

### Get Delivery Attempt from ConsumerRecord Headers

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

Retrieves the delivery attempt count from the `KafkaHeaders.DELIVERY_ATTEMPT` header of a `ConsumerRecord`. The value is stored as a 4-byte integer.

```java
int delivery = ByteBuffer.wrap(record.headers()
    .lastHeader(KafkaHeaders.DELIVERY_ATTEMPT).value())
    .getInt();
```

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

### 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

Sets the autoStartup property to false on a specific KafkaListener to prevent automatic container startup.

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

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

### Implement Custom Retry Topic Names Provider

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

Extend SuffixingRetryTopicNamesProvider to customize topic names. This example adds a 'my-prefix-' prefix to retry and DLT topics.

```java
public class CustomRetryTopicNamesProviderFactory implements RetryTopicNamesProviderFactory {

    @Override
    public RetryTopicNamesProvider createRetryTopicNamesProvider(
                DestinationTopic.Properties properties) {

        if (properties.isMainEndpoint()) {
            return new SuffixingRetryTopicNamesProvider(properties);
        }
        else {
            return new SuffixingRetryTopicNamesProvider(properties) {

                @Override
                public String getTopicName(String topic) {
                    return "my-prefix-" + super.getTopicName(topic);
                }

            };
        }
    }

}
```

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

### Use Custom KafkaListener 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

Apply the custom meta-annotation to a method to configure a Kafka listener. This example shows how to provide values for the aliased attributes.

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

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

### Invalid String Payload Header Resolution

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

Example showing that header resolution fails when the payload is a String, as the header parameter may incorrectly reference the payload.

```java
@KafkaHandler(isDefault = true)
public void listenDefault(String payload, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
    // payload.equals(topic) is True.
    ...
}
```

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

### ReplyingKafkaTemplate sendAndReceive Method

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

Demonstrates the `sendAndReceive` method of `ReplyingKafkaTemplate` for sending a `Message<?>` and receiving a reply asynchronously.

```java
RequestReplyMessageFuture<K, V> sendAndReceive(Message<?> message);
```

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

### Add DefaultErrorHandler to KafkaListenerContainerFactory

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

Demonstrates how to configure a ConcurrentKafkaListenerContainerFactory with a DefaultErrorHandler. This example sets a FixedBackOff with a 1-second delay and 2 retry attempts, overriding the default behavior.

```java
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setAckMode(AckMode.RECORD);
    factory.setCommonErrorHandler(new DefaultErrorHandler(new FixedBackOff(1000L, 2L)));
    return factory;
}
```

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

### Use KafkaMessageHeaderAccessor for Delivery Attempts

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

Starting with version 3.0.10, use KafkaMessageHeaderAccessor for simpler access to delivery attempt headers. The accessor can be provided as a parameter to the listener method.

```java
@RetryableTopic(backOff = @BackOff(...))
@KafkaListener(id = "dh1", topics = "dh1")
void listen(Thing thing, KafkaMessageHeaderAccessor accessor) {
    ...
}
```

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

### KafkaTemplate receive methods

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

Methods for retrieving records by partition and offset. Requires knowledge of specific partition and offset, and creates a new consumer per operation.

```java
ConsumerRecord<K, V> receive(String topic, int partition, long offset);

ConsumerRecord<K, V> receive(String topic, int partition, long offset, Duration pollTimeout);

ConsumerRecords<K, V> receive(Collection<TopicPartitionOffset> requested);

ConsumerRecords<K, V> receive(Collection<TopicPartitionOffset> requested, Duration pollTimeout);
```

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

### Demonstrate Pausing and Resuming Kafka Listeners

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

This Spring Boot application demonstrates pausing and resuming Kafka listener consumers. It uses the `KafkaListenerEndpointRegistry` to get a reference to a listener container and then calls `pause()` and `resume()` methods. It also prints Kafka events to the console.

```java
@SpringBootApplication
public class Application implements ApplicationListener<KafkaEvent> {

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

    @Override
    public void onApplicationEvent(KafkaEvent event) {
        System.out.println(event);
    }

    @Bean
    public ApplicationRunner runner(KafkaListenerEndpointRegistry registry,
            KafkaTemplate<String, String> template) {
        return args -> {
            template.send("pause.resume.topic", "thing1");
            Thread.sleep(10_000);
            System.out.println("pausing");
            registry.getListenerContainer("pause.resume").pause();
            Thread.sleep(10_000);
            template.send("pause.resume.topic", "thing2");
            Thread.sleep(10_000);
            System.out.println("resuming");
            registry.getListenerContainer("pause.resume").resume();
            Thread.sleep(10_000);
        };
    }

    @KafkaListener(id = "pause.resume", topics = "pause.resume.topic")
    public void listen(String in) {
        System.out.println(in);
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("pause.resume.topic")
            .partitions(2)
            .replicas(1)
            .build();
    }

}
```

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

### Programmatic DLT Routing with Exceptions

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

Configure DLT routing rules programmatically using RetryTopicConfigurationBuilder. This example routes messages causing DeserializationException to a DLT with a '-deserialization' suffix.

```java
@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .dltRoutingRules(Map.of("-deserialization", Set.of(DeserializationException.class)))
            .create(template);
}
```

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

### Build API Javadoc

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

Generate API Javadoc for the Spring Kafka project. Results are placed in the 'build/api' directory.

```bash
./gradlew api
```

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

### Configuring KafkaListenerContainerFactory with a Global Error Handler

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 setting a global error handler for all listeners managed by a ConcurrentKafkaListenerContainerFactory. This bean definition is typically part of your Spring configuration.

```java
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
        kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =

```

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

### Assigning a MessageListener to a Container

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

Demonstrates how to set a MessageListener on ContainerProperties and initialize a KafkaMessageListenerContainer.

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