Spring Kafka

https://docs.spring.io/spring-kafka/
Admin
The provided content is an index page for the Spring Kafka documentation and does not contain a...
Tokens:36,656
Snippets:212
Trust Score:10
Update:1 year ago
Show doc for...
Tokens:
Raw
### Database Table Creation for Transactional Example

Source: https://docs.spring.io/spring-kafka/reference/tips

SQL statement to create a simple table named 'mytable' with a single 'data' column of type varchar(20). This table is used in the transactional examples to store data inserted during the database operations.

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

```

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

### Chaining DB and Kafka Transactions (Spring Boot Application)

Source: https://docs.spring.io/spring-kafka/reference/tips

This Spring Boot application demonstrates chaining Kafka and database transactions. The listener starts the Kafka transaction, and the @Transactional annotation starts the DB transaction. The DB transaction commits first, ensuring idempotency if the Kafka transaction redelivers.

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

}

```

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

### Kafka Producer and Listener with Java Config (Spring Kafka)

Source: https://docs.spring.io/spring-kafka/reference/3.1/quick-tour

This snippet demonstrates a complete Spring Kafka application setup using Java configuration, without Spring Boot. It includes a Kafka sender class, a Kafka listener class, and a configuration class to define beans for Kafka components like `KafkaTemplate`, `ProducerFactory`, `ConsumerFactory`, and `ConcurrentKafkaListenerContainerFactory`. This setup is crucial for applications needing Kafka integration without the auto-configuration provided by Spring Boot.

```java
public class Sender {

    public static void main(String[] args) {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Config.class);
        context.getBean(Sender.class).send("test", 42);
    }

    private final KafkaTemplate<Integer, String> template;

    public Sender(KafkaTemplate<Integer, String> template) {
        this.template = template;
    }

    public void send(String toSend, int key) {
        this.template.send("topic1", key, toSend);
    }

}

public class Listener {

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

}

@Configuration
@EnableKafka
public class Config {

    @Bean
    ConcurrentKafkaListenerContainerFactory<Integer, String>
                        kafkaListenerContainerFactory(ConsumerFactory<Integer, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        return factory;
    }

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

    private Map<String, Object> consumerProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // ...
        return props;
    }

    @Bean
    public Sender sender(KafkaTemplate<Integer, String> template) {
        return new Sender(template);
    }

    @Bean
    public Listener listener() {
        return new Listener();
    }

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

    private Map<String, Object> senderProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        //...
        return props;
    }

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

}
```

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

### Apache Kafka Streams Support Example

Source: https://docs.spring.io/spring-kafka/reference/kafka/exactly-once

Demonstrates basic integration with Apache Kafka Streams within a Spring application. This typically involves defining Kafka Streams processors and topology.

```java
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class KafkaStreamsConfig {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        KStream<String, String> source = streamsBuilder.stream("input-topic");
        KStream<String, String> processedStream = source.mapValues(value -> value.toUpperCase());
        processedStream.to("output-topic");
        return source; // Or return the processedStream depending on your needs
    }
}
```

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

### Example Usage of KafkaTestUtils for Record Consumption (Java)

Source: https://docs.spring.io/spring-kafka/reference/testing

An example demonstrating the practical application of KafkaTestUtils.getSingleRecord to retrieve and assert a single consumed record from a Kafka topic after sending a message.

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

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

### Configure KafkaListener with Sequence Startup

Source: https://docs.spring.io/spring-kafka/reference/3.1/kafka/receiving-messages/sequencing

Demonstrates how to configure `@KafkaListener` annotations to ensure that listeners start up in a specific sequence. This is useful for managing dependencies between different message processing tasks.

```java
@Configuration
public class KafkaListenerConfig {

    @Bean
    public NewTopic topic1() {
        return TopicBuilder.name("my-topic-1").partitions(1).replicas(1).build();
    }

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

    @KafkaListener(topics = "my-topic-1", id = "listener1", autoStartup = "false")
    public void listen1(String message) {
        System.out.println("Received message from topic 1: " + message);
    }

    @KafkaListener(topics = "my-topic-2", id = "listener2", autoStartup = "false")
    public void listen2(String message) {
        System.out.println("Received message from topic 2: " + message);
    }

    @Bean
    public ApplicationRunner runner(KafkaListenerEndpointRegistry registry) {
        return args -> {
            System.out.println("Starting listener 1...");
            registry.getListenerContainer("listener1").start();
            System.out.println("Listener 1 started.");

            System.out.println("Starting listener 2...");
            registry.getListenerContainer("listener2").start();
            System.out.println("Listener 2 started.");
        };
    }
}
```

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

### Configure Topic and Partition Initial Offset

Source: https://docs.spring.io/spring-kafka/reference/3.1/kafka/receiving-messages/sequencing

Explains how to configure the initial offset for topic partitions when a Kafka consumer starts. This allows control over whether the consumer starts reading from the earliest available message, the latest, or a specific timestamp.

```java
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.IsolationLevel;

@Configuration
public class OffsetConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        
        factory.getContainerProperties().setListenerInitialOffset("earliest"); // or "latest", "none"
        
        // Example: Setting specific offsets for specific partitions
        factory.getContainerProperties().setPartitionOffset(
            new TopicPartition("my-specific-topic", 0), 
            "100", // offset value as String
            true // commit if false
        );
        
        return factory;
    }
}
```

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

### Application Properties for Kafka and DB Transactions

Source: https://docs.spring.io/spring-kafka/reference/tips

Configuration properties for setting up MySQL datasource and Kafka consumer/producer settings. Notably, it enables read_committed isolation level for Kafka consumers and sets a transaction ID prefix for producers.

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

```

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

### Manage a Single Embedded Kafka Broker Instance - Java

Source: https://docs.spring.io/spring-kafka/reference/testing

This Java class, `EmbeddedKafkaHolder`, manages a single instance of `EmbeddedKafkaBroker` for reuse across multiple test classes. It ensures the broker is started only once and provides a static method to retrieve the broker instance. It is designed for Spring Boot environments where it can replace the `spring.kafka.bootstrap-servers` property. The broker is started using `afterPropertiesSet()` and can be explicitly destroyed if needed.

```java
public final class EmbeddedKafkaHolder {

    private static EmbeddedKafkaBroker embeddedKafka = new EmbeddedKafkaZKBroker(1, false)
            .brokerListProperty("spring.kafka.bootstrap-servers");

    private static boolean started;

    public static EmbeddedKafkaBroker getEmbeddedKafka() {
        if (!started) {
            synchronized (this) {
		if (!started) {
		    try {
	                embeddedKafka.afterPropertiesSet();
		    }
	            catch (Exception e) {
	                throw new KafkaException("Embedded broker failed to start", e);
		    }
	            started = true;
		}
	    }
        }
        return embeddedKafka;
    }
}
```

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

### Producer-Only Transaction Synchronization (KafkaTemplate)

Source: https://docs.spring.io/spring-kafka/reference/tips

Demonstrates producer-only transaction synchronization with a database transaction manager ('dstm'). The KafkaTemplate will align its transaction with the DB transaction, ensuring both commit or rollback together.

```java
@Transactional("dstm")
public void someMethod(String in) {
    this.kafkaTemplate.send("topic2", in.toUpperCase());
    this.jdbcTemplate.execute("insert into mytable (data) values ('" + in + "')");
}

```

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

### Committing Kafka Transaction First (Nested @Transactional)

Source: https://docs.spring.io/spring-kafka/reference/tips

Example using nested @Transactional methods to commit the Kafka transaction before the database transaction. This approach ensures the DB transaction only commits if the Kafka transaction is successful.

```java
@Transactional("dstm")
public void someMethod(String in) {
    this.jdbcTemplate.execute("insert into mytable (data) values ('" + in + "')");
    sendToKafka(in);
}

@Transactional("kafkaTransactionManager")
public void sendToKafka(String in) {
    this.kafkaTemplate.send("topic2", in.toUpperCase());
}

```

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

### Configure Sequential Kafka Listener Startup with Java

Source: https://docs.spring.io/spring-kafka/reference/3.2-SNAPSHOT/kafka/receiving-messages/sequencing

This Java snippet demonstrates configuring multiple Kafka listeners with different `containerGroup` IDs. It also shows how to create a `ContainerGroupSequencer` bean to manage the startup order of these groups, ensuring listeners in 'g1' complete before 'g2' starts. The sequencer controls auto-startup and idle event intervals.

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

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

### Mock Producer Factory Configuration for Spring Kafka (Non-Transactional)

Source: https://docs.spring.io/spring-kafka/reference/testing

Configures a MockProducerFactory for creating non-transactional MockProducers. This is useful for testing scenarios where transactions are not required, simplifying producer setup.

```java
@Bean
ProducerFactory<String, String> nonTransFactory() {
    return new MockProducerFactory<>(() ->
            new MockProducer<>(true, new StringSerializer(), new StringSerializer()));
}
```

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

### Spring Boot Application Setup for Kafka Interceptors

Source: https://docs.spring.io/spring-kafka/reference/kafka/interceptors

This Java code sets up a Spring Boot application to configure Kafka producer and consumer factories. It demonstrates how to include custom interceptors and inject a 'SomeBean' dependency into them via configuration properties. This is crucial for enabling Spring Bean injection into interceptors managed by Kafka.

```java
@SpringBootApplication
public class Application {

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

    @Bean
    public ConsumerFactory<?, ?> kafkaConsumerFactory(SomeBean someBean) {
        Map<String, Object> consumerProperties = new HashMap<>();
        // consumerProperties.put(..., ...)
        // ...
        consumerProperties.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, MyConsumerInterceptor.class.getName());
        consumerProperties.put("some.bean", someBean);
        return new DefaultKafkaConsumerFactory<>(consumerProperties);
    }

    @Bean
    public ProducerFactory<?, ?> kafkaProducerFactory(SomeBean someBean) {
        Map<String, Object> producerProperties = new HashMap<>();
        // producerProperties.put(..., ...)
        // ...
        producerProperties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MyProducerInterceptor.class.getName());
        producerProperties.put("some.bean", someBean);
        return new DefaultKafkaProducerFactory<>(producerProperties);
    }

    @Bean
    public SomeBean someBean() {
        return new SomeBean();
    }

    @KafkaListener(id = "kgk897", topics = "kgh897")
    public void listen(String in) {
        System.out.println("Received " + in);
    }

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

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

}

```

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

### Embedded Broker with @EmbeddedKafka Annotation

Source: https://docs.spring.io/spring-kafka/reference/testing

This example shows how to use the @EmbeddedKafka annotation to create an embedded Kafka broker for testing. It configures the topics and the bootstrap server property, with the latter being the default since version 3.0.10.

```java
@RunWith(SpringRunner.class)
@EmbeddedKafka(topics = "someTopic",
        bootstrapServersProperty = "spring.kafka.bootstrap-servers") // this is now the default
public class MyApplicationTests {

    @Autowired
    private KafkaTemplate<String, String> template;

    @Test
    public void test() {
        ...
    }

}
```

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

### Apache Kafka Streams Integration

Source: https://docs.spring.io/spring-kafka/reference/search

Provides an example of integrating Apache Kafka Streams with Spring for Kafka. This allows building stream processing applications using Spring's programming model, leveraging Kafka's powerful stream processing capabilities.

```java
@Configuration
@EnableKafkaStreams
public class KafkaStreamsConfig {

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG)
    public StreamsConfig kStreamsConfigs() {
        Map<String, String> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-streams-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // Add other Kafka Streams configurations as needed
        return new StreamsConfig(props);
    }

    @Bean
    public KStream<String, String> kStream(StreamsBuilder builder) {
        KStream<String, String> stream = builder.stream("input-topic");
        stream.mapValues(value -> value.toUpperCase())
              .to("output-topic");
        return stream;
    }
}
```

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

### Implementing ApplicationListener for ConsumerStoppedEvent

Source: https://docs.spring.io/spring-kafka/reference/3.2-SNAPSHOT/kafka/thread-safety

This example shows how to implement the `ApplicationListener` interface to receive and process `ConsumerStoppedEvent`. This approach can be used as an alternative to `@EventListener` for cleaning up thread-specific resources.

```java
import org.springframework.context.ApplicationListener;
import org.springframework.kafka.event.ConsumerStoppedEvent;
import org.springframework.stereotype.Component;

@Component
public class ConsumerStoppedApplicationListener implements ApplicationListener<ConsumerStoppedEvent> {

    @Override
    public void onApplicationEvent(ConsumerStoppedEvent event) {
        // Clean up ThreadLocal instances or remove thread-scoped beans
        System.out.println("Consumer stopped event received, cleaning up for group: " + event.getGroupId());
    }
}
```

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

### Configure MessagingMessageConverter with Header Mapper (Java)

Source: https://docs.spring.io/spring-kafka/reference/kafka/headers

Provides a Java Bean configuration example for creating a MessagingMessageConverter and setting a DefaultKafkaHeaderMapper with string encoding enabled. This converter can then be used with KafkaTemplate.

```java
@Bean
MessagingMessageConverter converter() {
    MessagingMessageConverter converter = new MessagingMessageConverter();
    DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
    mapper.setEncodeStrings(true);
    converter.setHeaderMapper(mapper);
    return converter;
}
```

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

### Asynchronous @KafkaListener Return Types

Source: https://docs.spring.io/spring-kafka/reference/kafka/exactly-once

Explains and provides an example of using asynchronous return types with the `@KafkaListener` annotation, often used with CompletableFuture for non-blocking operations.

```java
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.concurrent.CompletableFuture;

@Component
public class AsyncListener {

    @KafkaListener(topics = "asyncTopic")
    public CompletableFuture<String> processAsync(String message) {
        return CompletableFuture.supplyAsync(() -> {
            // Simulate asynchronous processing
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return "Processed: " + message;
        });
    }
}
```

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

### Configuring KafkaListenerContainerFactory with Custom Reply Headers

Source: https://docs.spring.io/spring-kafka/reference/kafka/receiving-messages/annotation-send-to

This example shows how to configure a ConcurrentKafkaListenerContainerFactory with a custom ReplyHeadersConfigurer. It demonstrates overriding the shouldCopy method and providing additional headers to be sent with reply messages. A KafkaTemplate is also set.

```java
@Bean
public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory = 
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(cf());
    factory.setReplyTemplate(template());
    factory.setReplyHeadersConfigurer(new ReplyHeadersConfigurer() {

      @Override
      public boolean shouldCopy(String headerName, Object headerValue) {
        return false;
      }

      @Override
      public Map<String, Object> additionalHeaders() {
        return Collections.singletonMap("qux", "fiz");
      }

    });
    return factory;
}
```

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

### Dynamically Assign All Partitions to KafkaListener

Source: https://docs.spring.io/spring-kafka/reference/tips

This snippet demonstrates how to use a SpEL expression within the `@KafkaListener` annotation to dynamically determine and assign all partitions for a given topic. It includes a custom `PartitionFinder` bean that utilizes a `ConsumerFactory` to fetch partition information at application startup. This approach is useful for scenarios like loading compacted topics into a cache, where Kafka's group management is bypassed. The `AckMode` should be set to `MANUAL` to avoid offset commits when no consumer group ID is specified.

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

}
```

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

### Retrieve a Queryable Store by Name and Type

Source: https://docs.spring.io/spring-kafka/reference/streams

This example shows how to retrieve a queryable state store from the `KafkaStreamsInteractiveQueryService`. It uses the store's name ('app-store') and specifies the type as `keyValueStore` using `QueryableStoreTypes.keyValueStore()`.

```java
@Autowired
private KafkaStreamsInteractiveQueryService interactiveQueryService;

ReadOnlyKeyValueStore<Object, Object>  appStore = interactiveQueryService.retrieveQueryableStore("app-store", QueryableStoreTypes.keyValueStore());
```

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

### Configure Custom RetryTopicConfiguration Beans

Source: https://docs.spring.io/spring-kafka/reference/3.2-SNAPSHOT/retrytopic/retry-config

This example demonstrates configuring multiple RetryTopicConfiguration beans with specific retry policies for different topics. It allows for fine-grained control over backoff strategies, maximum attempts, concurrency, and which exceptions trigger retries, as well as including or excluding specific topics.

```java
@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .fixedBackOff(3000)
            .maxAttempts(5)
            .concurrency(1)
            .includeTopics("my-topic", "my-other-topic")
            .create(template);
}

@Bean
public RetryTopicConfiguration myOtherRetryTopic(KafkaTemplate<String, MyOtherPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .exponentialBackoff(1000, 2, 5000)
            .maxAttempts(4)
            .excludeTopics("my-topic", "my-other-topic")
            .retryOn(MyException.class)
            .create(template);
```

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

### Multi-Listener Class with @SendTo Annotations

Source: https://docs.spring.io/spring-kafka/reference/kafka/receiving-messages/annotation-send-to

This example shows a class with multiple KafkaListeners. One listener forwards to a static topic, while another uses a runtime SpEL expression to determine the reply topic based on headers. A KafkaTemplate is required for each.

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

}
```

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

### KafkaListener with @SendTo using Runtime SpEL Expression

Source: https://docs.spring.io/spring-kafka/reference/kafka/receiving-messages/annotation-send-to

This example demonstrates a KafkaListener that sends its return value to a topic determined by a runtime SpEL expression evaluating the request's value. It requires a KafkaTemplate to be configured in the listener container factory.

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

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

### Using KafkaTemplate to Receive

Source: https://docs.spring.io/spring-kafka/reference/spring-projects

Demonstrates a less common but supported pattern where KafkaTemplate can be used to receive messages, typically in scenarios involving request-reply or specific testing setups. This leverages the template's ability to send and receive within a unified interface.

```java
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;

public class KafkaTemplateReceiveExample {

    private final KafkaTemplate<String, String> kafkaTemplate;
    // Or ReplyingKafkaTemplate for request-reply scenarios
    // private final ReplyingKafkaTemplate<String, String, String> replyingKafkaTemplate;

    public KafkaTemplateReceiveExample(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
        // this.replyingKafkaTemplate = replyingKafkaTemplate;
    }

    public void sendAndReceive(String requestTopic, String replyTopic, String request) {
        // This is a simplified example. For actual request-reply, ReplyingKafkaTemplate is preferred.
        // This demonstrates sending and then potentially setting up a listener to receive.
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(requestTopic, request);
        
        future.addCallback(new org.springframework.util.concurrent.ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                System.out.println("Sent message successfully: " + request);
                // In a real scenario, you'd have a listener for the replyTopic
                // or use ReplyingKafkaTemplate which handles this internally.
            }

            @Override
            public void onFailure(Throwable ex) {
                System.err.println("Failed to send message: " + request + ", error: " + ex.getMessage());
            }
        });
    }
}
```

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

### Enable Global Embedded Kafka Listener - Properties

Source: https://docs.spring.io/spring-kafka/reference/testing

This configuration snippet shows how to enable the `GlobalEmbeddedKafkaTestExecutionListener` for JUnit Platform by setting the `spring.kafka.global.embedded.enabled` property to `true`. This listener starts a single global `EmbeddedKafkaBroker` for the entire test plan. Additional properties like count, ports, topics, and partitions can be configured to customize the broker's behavior.

```properties
spring.kafka.global.embedded.enabled=true
spring.kafka.embedded.count=1
spring.kafka.embedded.ports=0
spring.kafka.embedded.topics=topic1,topic2
spring.kafka.embedded.partitions=1
spring.kafka.embedded.broker.properties.location=classpath:kafka-broker.properties
spring.kafka.embedded.kraft=true
```

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

### Spring Kafka: Test KafkaTemplate with Embedded Kafka (Java)

Source: https://docs.spring.io/spring-kafka/reference/testing

This Java code demonstrates how to test Spring Kafka's `KafkaTemplate` using an `EmbeddedKafkaRule`. It sets up an embedded Kafka broker, configures a Kafka listener container to consume messages, and then uses `KafkaTemplate` to send messages. Assertions verify that the correct messages are received. This example relies on Spring Kafka and JUnit.

```java
public class KafkaTemplateTests {

    private static final String TEMPLATE_TOPIC = "templateTopic";

    @ClassRule
    public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, TEMPLATE_TOPIC);

    @Test
    public void testTemplate() throws Exception {
        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testT", "false",
            embeddedKafka.getEmbeddedKafka());
        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.getEmbeddedKafka().getPartitionsPerTopic());
        Map<String, Object> producerProps =
                            KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka());
        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"));
    }

}
```

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

### Configure Multi-Handler KafkaListener for Null Payloads (Java)

Source: https://docs.spring.io/spring-kafka/reference/3.1/kafka/tombstones

This example shows how to configure a class-level `@KafkaListener` with multiple `@KafkaHandler` methods to manage different payload types, including null payloads using `KafkaNull`. A dedicated handler is set up to process `KafkaNull` for key deletion.

```java
@KafkaListener(id = "multi", topics = "myTopic")
static class MultiListenerBean {

    @KafkaHandler
    public void listen(String cat) {
        ...
    }

    @KafkaHandler
    public void listen(Integer hat) {
        ...
    }

    @KafkaHandler
    public void delete(@Payload(required = false) KafkaNull nul, @Header(KafkaHeaders.RECEIVED_KEY) int key) {
        ...
    }

}
```

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

### KafkaTestUtils Helper Methods for Testing Spring Kafka

Source: https://docs.spring.io/spring-kafka/reference/testing

Introduces the KafkaTestUtils class from 'spring-kafka-test', which offers static helper methods for common testing tasks such as consuming records and retrieving offsets. Refer to its Javadocs for detailed usage.

```java
org.springframework.kafka.test.utils.KafkaTestUtils
```

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

### Spring Boot Kafka Producer App Configuration

Source: https://docs.spring.io/spring-kafka/reference/3.1/quick-tour

Configures a Spring Boot application to produce messages to a Kafka topic. It includes a `NewTopic` bean for topic creation and an `ApplicationRunner` to send a test message upon startup. This is useful for basic message production scenarios.

```java
@SpringBootApplication
public class Application {

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

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("topic1")
                .partitions(10)
                .replicas(1)
                .build();
    }

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

}

```

```kotlin
@SpringBootApplication
class Application {

    @Bean
    fun topic() = NewTopic("topic1", 10, 1)

    @Bean
    fun runner(template: KafkaTemplate<String?, String?>) =
        ApplicationRunner { template.send("topic1", "test") }

    companion object {
        @JvmStatic
        fun main(args: Array<String>) = runApplication<Application>(*args)
    }

}

```

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

### Spring Boot Kafka Consumer App Configuration

Source: https://docs.spring.io/spring-kafka/reference/3.1/quick-tour

Sets up a Spring Boot application to consume messages from a Kafka topic. It defines a Kafka listener and configures topic creation. The `application.properties` file specifies the auto-offset reset policy for the consumer. This snippet is applicable if the topic does not already exist on the broker.

```java
@SpringBootApplication
public class Application {

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

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("topic1")
                .partitions(10)
                .replicas(1)
                .build();
    }

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

}

```

```kotlin
@SpringBootApplication
class Application {

    @Bean
    fun topic() = NewTopic("topic1", 10, 1)

    @KafkaListener(id = "myId", topics = ["topic1"])
    fun listen(value: String?) {
        println(value)
    }

}

fun main(args: Array<String>) = runApplication<Application>(*args)

```

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

### Configuring Kafka Topics with Spring

Source: https://docs.spring.io/spring-kafka/reference/retrytopic/access-topic-info-runtime

This section explains how to configure Kafka topics for use with Spring. It covers defining topic names, partitions, and replicas, which are essential for managing Kafka data streams within a Spring application. No specific dependencies are required beyond the core Spring for Apache Kafka library.

```java
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.apache.kafka.clients.admin.NewTopic;

@Configuration
public class KafkaTopicConfig {

    @Bean
    public NewTopic myTopic() {
        return new NewTopic("my-topic", 10, 1);
    }

    // Other configurations for KafkaTemplate if needed
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }
}
```

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

### Add Spring Kafka Dependency (Gradle)

Source: https://docs.spring.io/spring-kafka/reference/3.1/quick-tour

This snippet demonstrates how to include the spring-kafka library in a Gradle project. It specifies the dependency for compilation.

```gradle
compile 'org.springframework.kafka:spring-kafka:3.1.10'
```

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

### Add Kafka Topics Programmatically (Java)

Source: https://docs.spring.io/spring-kafka/reference/testing

Demonstrates how to add Kafka topics with specific partition counts and replication factors using `EmbeddedKafkaRule`. It shows creating topics with default configurations and dynamically adding more topics with custom settings.

```java
public class MyTests {

    @ClassRule
    private static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, false, 5, "cat", "hat");

    @Test
    public void test() {
        embeddedKafkaRule.getEmbeddedKafka()
              .addTopics(new NewTopic("thing1", 10, (short) 1), new NewTopic("thing2", 15, (short) 1));
        ...
    }

}
```

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

### Kafka Producer and Listener with Kotlin Config (Spring Kafka)

Source: https://docs.spring.io/spring-kafka/reference/3.1/quick-tour

This snippet showcases a Spring Kafka application configured using Kotlin, demonstrating the same producer and listener functionality as the Java version but with Kotlin's concise syntax. It defines the necessary beans for Kafka integration, including `KafkaTemplate`, `ProducerFactory`, `ConsumerFactory`, and `ConcurrentKafkaListenerContainerFactory`. This is suitable for Kotlin-based projects requiring Spring Kafka features without Spring Boot's auto-configuration.

```kotlin
class Sender(private val template: KafkaTemplate<Int, String>) {

    fun send(toSend: String, key: Int) {
        template.send("topic1", key, toSend)
    }

}

class Listener {

    @KafkaListener(id = "listen1", topics = ["topic1"])
    fun listen1(`in`: String) {
        println(`in`)
    }

}

@Configuration
@EnableKafka
class Config {

    @Bean
    fun kafkaListenerContainerFactory(consumerFactory: ConsumerFactory<Int, String>) =
        ConcurrentKafkaListenerContainerFactory<Int, String>().also { it.consumerFactory = consumerFactory }


    @Bean
    fun consumerFactory() = DefaultKafkaConsumerFactory<Int, String>(consumerProps)

    val consumerProps = mapOf(
        ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to "localhost:9092",
        ConsumerConfig.GROUP_ID_CONFIG to "group",
        ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to IntegerDeserializer::class.java,
        ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
        ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "earliest"
    )

    @Bean
    fun sender(template: KafkaTemplate<Int, String>) = Sender(template)

    @Bean
    fun listener() = Listener()

    @Bean
    fun producerFactory() = DefaultKafkaProducerFactory<Int, String>(senderProps)

    val senderProps = mapOf(
        ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to "localhost:9092",
        ProducerConfig.LINGER_MS_CONFIG to 10,
        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to IntegerSerializer::class.java,
        ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java
    )

    @Bean

```

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

### Non-Blocking Retries Configuration in Spring Kafka

Source: https://docs.spring.io/spring-kafka/reference/kafka/exactly-once

Provides an example of configuring non-blocking retries for message processing in Spring Kafka. This pattern is useful for handling transient failures without blocking the consumer thread.

```java
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

@Configuration
public class KafkaRetryConfig {

    @Bean
    public DefaultErrorHandler errorHandler() {
        // Non-blocking retries with a fixed back-off
        SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler(
            new FixedBackOff(1000L, 5) // Delay of 1 second, 5 retries
        );
        return errorHandler;
    }

    // Ensure your KafkaListenerContainerFactory is configured to use this errorHandler
    // factory.setErrorHandler(errorHandler()); 
}
```

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

### Disable DLT Container Auto-Startup

Source: https://docs.spring.io/spring-kafka/reference/retrytopic/dlt-strategies

This snippet illustrates how to prevent the DLT container from starting automatically when using the `@RetryableTopic` annotation by setting `autoStartDltHandler` to `false`. This allows for manual control over the DLT handler's lifecycle.

```java
@RetryableTopic(autoStartDltHandler = false)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    // ... message processing
}
```

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

### Add Kafka Topics with Error Handling (Java)

Source: https://docs.spring.io/spring-kafka/reference/testing

Introduces a version of the `addTopics` method that returns a `Map<String, Exception>` for improved error handling. This allows capturing and inspecting exceptions for each topic creation attempt, differentiating between success (null value) and failure.

```java
Map<String, Exception> results = embeddedKafkaRule.getEmbeddedKafka()
        .addTopics(new NewTopic("topic-with-error", 1, (short) 1));
```

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

### Configuring Message Listener Containers in Spring Kafka

Source: https://docs.spring.io/spring-kafka/reference/kafka/exactly-once

Demonstrates how to configure message listener containers for receiving messages from Kafka topics using Spring Kafka. This includes setting up container factories and listener properties.

```java
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(ConsumerFactory<Object, Object> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        // Additional configurations can be added here, e.g., concurrency, error handling
        return factory;
    }
}
```

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

### Configure KafkaJaasLoginModuleInitializer for Kerberos

Source: https://docs.spring.io/spring-kafka/reference/kafka/kerberos

This Java snippet demonstrates how to configure the KafkaJaasLoginModuleInitializer bean. It sets the control flag and provides a map of Kerberos-specific options such as keytab path, principal, and store key settings. This is essential for enabling JAAS and Kerberos authentication in Spring Kafka.

```java
@Bean
public KafkaJaasLoginModuleInitializer jaasConfig() throws IOException {
    KafkaJaasLoginModuleInitializer jaasConfig = new KafkaJaasLoginModuleInitializer();
    jaasConfig.setControlFlag("REQUIRED");
    Map<String, String> options = new HashMap<>();
    options.put("useKeyTab", "true");
    options.put("storeKey", "true");
    options.put("keyTab", "/etc/security/keytabs/kafka_client.keytab");
    options.put("principal", "kafka-client-1@EXAMPLE.COM");
    jaasConfig.setOptions(options);
    return jaasConfig;
}
```

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

### Disable DLT Container Auto-Startup via Builder

Source: https://docs.spring.io/spring-kafka/reference/retrytopic/dlt-strategies

This example demonstrates disabling the DLT container's auto-startup using the `RetryTopicConfigurationBuilder` by calling `autoStartDltHandler(false)`. This provides programmatic control over the DLT container's initiation.

```java
@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .autoStartDltHandler(false)
            .create(template);
}
```

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

### Kafka Streams API Usage (Java)

Source: https://docs.spring.io/spring-kafka/reference/streams

Demonstrates the basic usage of the Apache Kafka Streams API within a Java application. It shows how to build a processing topology using StreamsBuilder and manage the lifecycle of the KafkaStreams instance.

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

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

### Define StreamsBuilderFactoryBean for Kafka Streams

Source: https://docs.spring.io/spring-kafka/reference/streams

Defines a StreamsBuilderFactoryBean as a Spring bean to manage Kafka Streams lifecycle and expose a StreamsBuilder singleton. It takes KafkaStreamsConfiguration as input. Starting with version 2.2, KafkaStreamsConfiguration is preferred over StreamsConfig.

```java
@Bean
public FactoryBean<StreamsBuilder> myKStreamBuilder(KafkaStreamsConfiguration streamsConfig) {
    return new StreamsBuilderFactoryBean(streamsConfig);
}
```

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

### Configure Kafka Streams Support

Source: https://docs.spring.io/spring-kafka/reference/3.1/kafka/receiving-messages/sequencing

Provides guidance on integrating Apache Kafka Streams with Spring applications. This allows for building stream processing applications directly within the Spring ecosystem, leveraging Spring's dependency injection and configuration management.

```java
import org.apache.kafka.streams.StreamsBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class KafkaStreamsConfig {

    @Bean
    public StreamsBuilder streamsBuilder() {
        return new StreamsBuilder();
    }

    @Bean
    public org.springframework.kafka.support.TopicBuilder<String, String> topicBuilder() {
        // Example: Defining topics for Kafka Streams
        return org.springframework.kafka.support.TopicBuilder;
    }

    // Define your Kafka Streams topology here, using the StreamsBuilder
    @Bean
    public void kafkaStreamsTopology(StreamsBuilder builder) {
        builder.stream("input-topic")
               .mapValues(value -> value.toUpperCase())
               .to("output-topic");
    }
}
```

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

### KafkaTemplate Local Transaction Execution

Source: https://docs.spring.io/spring-kafka/reference/3.1/kafka/transactions

This example shows how to execute a series of KafkaTemplate send operations within a local, self-contained transaction using KafkaTemplate.executeInTransaction. The transaction is automatically committed if the provided callback completes successfully, and rolled back if any exception is thrown.

```java
boolean result = template.executeInTransaction(t -> {
    t.sendDefault("thing1", "thing2");
    t.sendDefault("cat", "hat");
    return true;
});
```

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

### Kafka Streams Configuration with Spring for Apache Kafka

Source: https://docs.spring.io/spring-kafka/reference/streams

This Java configuration sets up a Kafka Streams application within a Spring Boot environment. It defines the necessary properties for Kafka Streams, including bootstrap servers and serializers/deserializers. It also includes a state listener for monitoring stream state transitions and a KStream pipeline that processes messages from 'streamingTopic1', converts them to uppercase, groups them by key, applies windowed aggregation, filters based on value length, and outputs to 'streamingTopic2'.

```java
@Configuration
@EnableKafka
@EnableKafkaStreams
public class KafkaStreamsConfig {

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kStreamsConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "testStreams");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, WallclockTimestampExtractor.class.getName());
        return new KafkaStreamsConfiguration(props);
    }

    @Bean
    public StreamsBuilderFactoryBeanConfigurer configurer() {
        return fb -> fb.setStateListener((newState, oldState) -> {
            System.out.println("State transition from " + oldState + " to " + newState);
        });
    }

    @Bean
    public KStream<Integer, String> kStream(StreamsBuilder kStreamBuilder) {
        KStream<Integer, String> stream = kStreamBuilder.stream("streamingTopic1");
        stream
                .mapValues((ValueMapper<String, String>) String::toUpperCase)
                .groupByKey()
                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMillis(1_000)))
                .reduce((String value1, String value2) -> value1 + value2,
                		Named.as("windowStore"))
                .toStream()
                .map((windowedId, value) -> new KeyValue<>(windowedId.key(), value))
                .filter((i, s) -> s.length() > 40)
                .to("streamingTopic2");

        stream.print(Printed.toSysOut());

        return stream;
    }

}

```

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

### Configure Embedded Kafka Broker with Properties and Placeholders (Java)

Source: https://docs.spring.io/spring-kafka/reference/testing

This snippet shows how to use the @EmbeddedKafka annotation with advanced configurations, including specifying topics using property placeholders, defining broker properties directly, and loading broker properties from an external file. It demonstrates property placeholder resolution from the Spring Environment and external configuration files. Dependencies include Spring Test and Spring Kafka.

```java
@TestPropertySource(locations = "classpath:/test.properties")
@EmbeddedKafka(topics = { "any-topic", "${kafka.topics.another-topic}" },
        brokerProperties = { "log.dir=${kafka.broker.logs-dir}",
                            "listeners=PLAINTEXT://localhost:${kafka.broker.port}",
                            "auto.create.topics.enable=${kafka.broker.topics-enable:true}" },
        brokerPropertiesLocation = "classpath:/broker.properties")

```

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

### Exclude Test Support Binder in @SpringBootTest

Source: https://docs.spring.io/spring-kafka/reference/testing

This example demonstrates how to exclude the spring-cloud-stream test support binder when using an embedded broker for tests in a Spring Boot application. This is necessary to ensure the real binder is used instead of the test binder.

```java
@RunWith(SpringRunner.class)
@SpringBootTest(properties = "spring.autoconfigure.exclude="
    + "org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration")
public class MyApplicationTests {
    ...
}
```
Tokens:
Raw