Spring Boot 3 With Kafka

Coding 

 On this step I will not introduce more specific on how to create multiple project while it depends on your IDE or sometimes manaul by hard copy and patse lol.

• Common's project 
  • Create OrderDto with directory dto package (/dto) for using as payload or message to queue.

package com.tra22.kafka.dto;

import lombok.*;
import java.io.Serializable;

@Getter
@Setter
@Builder
@ToString
@NoArgsConstructor
@AllArgsConstructor
public class OrderDto implements Serializable {
    private int orderId;
    private double price;
}

• • Create KafkaProducerService with directory producer (/producer) for using as global produce message to topic

package com.tra22.kafka.producer;

import com.tra22.kafka.dto.OrderDto;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
@Slf4j
@RequiredArgsConstructor
public class KafkaProducerService {
    private final KafkaTemplate<String, OrderDto> kafkaTemplate;
    public void send(String topicName, String key, OrderDto value) {
        var future = kafkaTemplate.send(topicName, key, value);
        future.whenComplete((sendResult, exception) -> {
            if (exception != null) {
                future.completeExceptionally(exception);
                log.error("error: ", exception);
            } else {
                future.complete(sendResult);
                log.info("result: {}", sendResult);
            }
            log.info("order producer : {} ", value);
        });
    }
}

• Producer's project
  • Add properties config with kafka like below

server:
  port: 8089
spring:
  kafka:
    bootstrap-servers: localhost:9092
    # producer config
    producer:
       key-serializer: org.apache.kafka.common.serialization.StringSerializer
       value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

• • For this code I'm just testing so I'm try to using on main class (it's bad don't follow me lol please using controllers, services layer to doing it). Also I called KafkaProducerService from Common's project. Please take note: if your project namespace or package is same so you don't need to add @ComponentScan (it'll auto scan for us and can use) but if your namespace or package different so you will faced issue no bean found your Producer's project. For fix it you can uncomment and changed your namespace or package your Common's project.

package com.tra22.kafka;

import com.tra22.kafka.dto.OrderDto;
import com.tra22.kafka.producer.KafkaProducerService;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
//@ComponentScan(basePackages = {"com.tra22.common"})
public class Main {
    public static void main(String[] args) {
        SpringApplication.run(Main.class, args);
    }
    @Bean
    public CommandLineRunner commandLineRunner(
            KafkaProducerService kafkaTemplate
    ) {
        return args -> {
            kafkaTemplate.send("order-topic", "orderId", OrderDto.builder().orderId(1).price(100.23).build());
        };
    }
}

• Consumer's project
  • Add properties config with kafka like below

server:
  port: 8082
spring:
  kafka:
    bootstrap-servers: localhost:9092
    # consumer config
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "*"

• • Create KafKaTopicListeners with directory event (/event) with @KafkaListener  is used to designate a bean method as a listener for a listener container. The bean is wrapped in a MessagingMessageListenerAdapter configured with various features, such as converters to convert the data, if necessary, to match the method parameters.

package com.tra22.kafka.event;

import com.tra22.kafka.dto.OrderDto;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class KafKaTopicListeners {
    @KafkaListener(topics = {"order-topic"}, groupId = "order-group")
    public void consume(OrderDto orderDto) {
        log.info("order consumer : {} ", orderDto);
    }
}