[Kafka] SpringBoot3.x.x에서 Kafka 연동하기

 

 

 

스프링 부트 3.1.2에 Kafka를 세팅해 보자

📌 서론

이전 포스트를 통해 docker-compose로 Kafka를 띄웠다.

이번 포스트에서는 SpringBoot3.1.2에 세팅한 카프카를 연결하고 우리가 원했던 대로 예외가 발생하면 예외 메시지를 카프카의 producer로 발행한 다음 docker-desktop의 콘솔을 통해 consumer로 메시지를 잘 받는지 확인했다.

이번 포스트에서는 그 과정을 설명한다.

이전 포스트를 통해 docker-compose로 kafka를 실행하고 오도록 하자 (M1 mac 기준)

[kafka] Docker로 카프카 실행하기 (KRaft 모드)

 

1. 스프링 부트에 Kafka 세팅하기

build.gradle에 의존성 추가

• 가장 빠르게 필요한 의존성의 버전을 알아보는 방법은 spring initialize를 사용하는 것이다.

Spring for Apache Kafka

// kafka  
implementation 'org.springframework.kafka:spring-kafka'

의존성을 추가한 다음 applicationo.yml에 카프카 설정 코드를 작성했다.

• 나는 active-profile 기능을 사용 중이라 yml 중 dev에만 설정을 했다.
• 아래의 설정 코드에서 consumer관련해서 빠진 부분이 있다. (아래의 포스트를 통해 확인해 보자)

spring:  
    kafka:  
      producer:    
        bootstrap-servers: localhost:10000,localhost:10001,localhost:10002 # 카프카 서버 주소  
        key-serializer: org.apache.kafka.common.serialization.StringSerializer  
        value-serializer: org.apache.kafka.common.serialization.StringSerializer  
      consumer:  
        group-id: error-handler-group  
        auto-offset-reset: earliest  
      template:  
        default-topic: error-messages

앞으로 계속 따라올 예정이라면 먼저 아래의 글을 통해 application.yml을 수정하고 진행하도록 하자

[kafka] Spring실행 시 consumer 연결문제 해결

 

2. 예외처리 코드에 카프카 메시지를 발행하는 코드를 추가했다.

중앙 집중 예외처리 코드에 Kafka 메시지를 발행하는 코드 추가

• 나는 @RestControllerAdvice를 통해 중앙 집중 예외처리를 하고 있는데 여기서 kafka에 메시지를 발행하도록 할 것이다.
• 오류가 발생하면 kafka에 메시지를 발행하도록 커스텀 예외처리 코드에 추가했다.

controllerAdvice 코드 수정

예외에 대한 응답을 처리하는 코드는 다음과 같이 동작한다.

• 아래와 같이 코드를 2개로 나눴다.
• 레시피아 에러는 kafka 메시지 발행을 하지 않고 그 외의 파악하기 힘든 위급한 에러들만 카프카에 메시지를 발행하도록 했다.

카프카 에러 메시지 발행

3. 예외를 발생시켜서 검증하기

먼저 kafka메시지를 발행하지 않는 예외를 강제로 던지기

• 레시피 단건 조회를 하면 강제로 레시피아 에러를 던지도록 했다.

예외를 던지는 코드

• 예외는 다음과 같다. (참고로 이 예외를 던졌을 때는 kafka 컨슈머에 아무런 메시지가 안 나와야 한다.)

커스텀 예외코드 확인

• postman을 통해 요청을 보냈다. (message가 null인 이유는 코드 자체에 null로 세팅해서 그렇다.)

포스트맨 요청 던지기

• docker-deskop의 콘솔에서 컨슈머는 아무런 메시지를 받지 않았다. (kafka 메시지 발행이 안되어 테스트에 성공했다.)

docker desktop 콘솔 확인

이번에는 kafka메시지를 발행하는 예외를 던졌다.

• kafka 메시지를 발행하는 IllegalArgumentException을 던지도록 코드를 수정했다.

카프카 메시지를 발행하는 예외

• 이번에도 아래와 같이 postman에 동일한 요청을 보냈다.

포스트맨에 요청 보내기

• docker-desktop의 콘솔에서 컨슈머를 확인했더니 다음과 같은 결과가 나왔다. (발행된 메시지를 잘 받았다.)

docker desktop 콘솔

이렇게 Kafka를 세팅했고 특정 예외가 발생하면 kafka producer로 메시지를 발행하고 consumer로 메시지를 받는 것까지 검증하는 데 성공했다. 

 

다음 포스트를 통해 슬랙의 webhook 설정을 해보자

[Kafka] 슬랙 webhook 설정하기 (kafka에서 호출)

사이드 팀원 "평양냉면"님의 블로그도 방문해주세요! 좋은 글이 많습니다.

하다보니 재미있는 개발