> For the complete documentation index, see [llms.txt](https://tingreavinash.gitbook.io/the-tech-toolbox/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://tingreavinash.gitbook.io/the-tech-toolbox/notes/apache-kafka/kafka-with-spring-boot.md).

# Kafka with Spring Boot

#### Dependencies

* `spring-kafka` : For Spring Kafka support
* `kafka-clients` : For Kafka client support

#### Configuration

* Configure Kafka broker URL and port in `application.properties`:

```
spring.kafka.bootstrap-servers=<kafka-broker-url>:<port>
```

#### Producer Configuration

* Use `KafkaTemplate` to send messages to Kafka topics
* Create a `KafkaTemplate` bean:

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

* Define a `ProducerFactory` bean:

```java
@Bean
public ProducerFactory<String, String> producerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    return new DefaultKafkaProducerFactory<>(configProps);
}
```

* Use `KafkaTemplate` to send messages to Kafka topics:

```java
kafkaTemplate.send("myTopic", "myMessage");
```

#### Consumer Configuration

* Create a `KafkaListenerContainerFactory` bean:

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

* Define a `ConsumerFactory` bean:

```java
@Bean
public ConsumerFactory<String, String> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "myGroupId");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    return new DefaultKafkaConsumerFactory<>(props);
}
```

* Use `@KafkaListener` annotation to consume messages from Kafka topics:

```java
@KafkaListener(topics = "myTopic", groupId = "myGroupId")
public void listen(String message) {
    System.out.println("Received message: " + message);
}
```