[kafka]saga pattern 예시

 

1. kafka, zookeeper 구축

version: "3"

services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"

  kafka:
    image: wurstmeister/kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
      KAFKA_CREATE_TOPICS: "one-fail:1:1, two-fail:1:1, three-fail:1:1, final-fail:1:1"
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

zookeeper는 localhost:2181, kafka는 localhost:9092로 접속.

 

2. application.yml

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: group-01
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.LongDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.LongSerializer

3. kafka로 메서드 호출

...
public class ........
...
    private final KafkaTemplate kafkaTemplate;
    
    public ....... {
        ...
        kafkaTemplate.send("one-fail", 1L);
        ...
    }
...

4. kafka로 호출된 메서드 구현

@Component
public class ... {

    @KafkaListener(topics = "one-fail", groupId = "group-01")
    public ... {
    	...
        System.out.println("kafka > one-fail");
        ...
    }

---

5. application도 docker로 같이 올리기

// docker-compose.yml
version: "3"

services:
  one:
    build: appone
    mem_limit: 350m
    ports:
      - "8081:8080"
    depends_on:
      - kafka

  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"

  kafka:
    image: wurstmeister/kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: kafka
      KAFKA_CREATE_TOPICS: "one-fail:1:1, two-fail:1:1, three-fail:1:1, final-fail:1:1"
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

one이라는 프로젝트 이름으로 depends_on 에 kafka 추가

KAFKA_ADVERTISED_HOST_NAME을 kafka로 지정(application.yml에서 이 이름으로 사용할 예정)

// application.yml
spring:
  kafka:
    bootstrap-servers: kafka:9092
    consumer:
      group-id: group-01
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.LongDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.LongSerializer

spring.kafka.bootstrap-servers를 "kafka:9092"로 설정

consumer, producer 서버가 다르다면, 각 하위 설정으로 "bootstrap-servers"를 설정하면 된다고 함.

---

6. 구현방법(뇌피셜 주의)

묶어야 할 transaction 처리 도중, 실패시 kafka메서드 호출. 데이터 일관성 유지.

[method1, method2, method3]을 한 transaction으로 묶는다 치면.

• method1 처리 도중 실패 > kafka 호출(근데 호출 할 필요 있을까?)
• method1 처리 성공 > method2 처리 도중 실패 > kafka 호출: method1에서 처리했던 작업을 원상복구(method1에서 insert했으면 kafka에서 delete)
• method1 처리 성공 > method2 처리 성공 > method3 처리 도중 실패 > kafka 호출: method1 원상복구, method2 원상복구

만약에 method1에서 method2 호출 후 다시 method1로 돌아온 후 실패 시?

그림과 같은 로직이 있다고 치자.

• method2 호출 전 method1 처리 도중 실패 > kafka 호출(근데 호출 할 필요 있을까?)
• method2 처리 도중 실패 > kafka 호출: method2, method1에서 처리했던 로직 원상복구
• method2 처리 후 method1 처리 도중 실패 > kafka 호출: method2, method1에서 처리했던 로직 원상복구

만약 @Service처럼 exception발생시 자동 rollback 설정이 되어 있다면,

method1에서는 exception만 날려줘도 rollback이 될거임.

resultCd != '01'일때 exception발생 시키고, method2 작업의 rollback은 kafka를 호출로 해야 할듯함.