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:

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

• Define a ProducerFactory bean:

@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:

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

Consumer Configuration

• Create a KafkaListenerContainerFactory bean:

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

• Define a ConsumerFactory bean:

@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:

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