Kafka

Kafka는 비동기 + 배치 동작으로 빠르게 이벤트를 저장하여, MSA 에서 많이 사용된다. 특정 서버에서 작업이 끝났을때 이벤트 로그를 등록 및 기록하고,  원하는곳에서 이 이벤트 로그를 확인함으로서 비동기로 분산되어 있는 서버간의 작업을 유기적으로 수행하도록 해준다.

 

RabblitMQ 와 큰 차이점이라고 하면, Kafka는 이벤트를 로그로서 영구저장하고, RabblitMQ 는 메세지큐 로서 메세지를 영구 저장하지 않고 소비 후 삭제한다. RabbitMQ가 실시간성은 매우 좋은 반면, 대량의 메세지 처리에는 Kafka가 좋다.
또한 Kafka 는 메세지를 컨슈머가 가져가는 형태인 반면, RabbitMQ의 경우 브로커가 push를 통해 전달하는 형태이다.

 

Kafka의 구성

•  크게 Producer, Consumer , Broker로 나뉜다.
  
  
• Producer
  • 이벤트로그를 발행하는 주체이다. 
  • send() 를 통해 발행시 다음의 순서로 동작한다.
    • Serializer를 통해 byte[] 로 Serialize 수행
    • 키값을 통해 어떤 Partition에 쓸지 결정 (Partitioner 가 수행)
      • Partition 은 브로커(kafka) 에 있는 물리적인 로그 개념이고 Partitioner는 Producer 내부 로직임.
    • RecordAccumulator 를 통해 같은 토픽, 같은 파티션 메세지를 메모리에 묶어 높음 (배치작업) - kafka가 빠른 이유
      • 배치 크기가 도달하거나, 지정한 시간에 도달했거나, 수동으로 flush를 했을때 네트워크에 전송
    • Sender Thread 를 통해 여러 배치를 동시에 네트워크로 전송 (비동기)
      
      
• Brocker
  • 한대의 kafka 서버이다.
  • 메세지를 실제로 디스크에 append-only 로 저장한다.
  • 다른 브로커에 복제 관리 (Replica)
  • 어떤 파티션에 쓸지 정하는 Partition Leader 관리
  • 각종 요청 처리
  • 특수 브로커로, 컨트롤러 브로커가 있다. 
    • 브로커중 하나로서, 장애를 감지하고 브로커들을 조율하는 역할을 한다.
  • 설정에서 이 역할을 지정할 수 있고, controller 역할이 부여된 경우 하나의 리더가 선출된다. (장애가 생기면 다른 컨트롤러 브로커가 승격)
    
    
• Consumer 
  •  단순히 메세지를 수신하고 지우는게 아니라, 메세지는 영구보존 되어 있고, offset을 이동시킨다.
  • Consumer Group을 통해 여러 Consumer를 묶을 수 있고, 하나의 메세지는 그룹내에서 한 Consumer만 처리한다.
    
    • 즉, 하나의 offset이 하나의 Consumer group에 쓰인다.
    • 메세지들 간의 순서가 상관없을때, 메세지의 처리를 각 컨슈머가 병렬로 처리 할 수 있다. 
  • 하나의 메세지가 하나의 Consumer 에서 실행되는것은 맞으나, 실행 성공여부는 알 수 없다. 
    • Consumer A 가 모든 처리를 완료했으나, offset commit 하기전에 죽으면 다른 Consumer가 똑같은 일을 또 할 수 있다. => Consumer 설계시 중복 처리 될 수 있음을 인지하고 개발해야 한다.
      
      
      

Kafka 가 빠른 이유

• 비동기 + 배치 동작 방식으로 돌아간다.
• 순차 쓰기 방식을 활용한다.
  • SSD가 빠른 이유는, Random Access를 통해 어떤 데이터든 한번에 접근이 가능하기 때문이다. 
  • 그러나 비싸고, 대부분 대용량 저장소는 HDD를 이용한다. 
  • HDD의 경우, 순차 쓰기, 읽기일 경우, 이전 주소값만 알면 바로 읽고 쓸 수가 있어서 속도가 매우 빠르다.
• page cache를 적극 활용한다.
  
  • OS의 메모리 캐시로, 디스크 데이터를 메모리에 자동 캐싱해 놓는다.
  • 읽기 성능도 빠르다 (최근 메세지는 대부분 여기에 있음)
• send() 를 통해 이벤트 로그 발행시, 바로 네트워크 전송을 하지 않고 메모리에 쌓은 후 배치로 전송한다.

Kafka의 데이터 모델

• Topic : 논리적 로그 스트림. 이벤트 종류 단위이다.
  • Topic 안에 여러 메세지가 있을 수 있다. 즉, DB로 생각해보면 테이블명 같은 느낌이다.
  • Topic은 하나의 비즈니스 로직 처리 흐름 단위로 구별하는게 좋다.
    • 예 : order-events, payment-events, user-events ...
  • 하나의 토픽에 전달되는 value의 이벤트 자료형은, 하나로 처리하는게 좋다. 
    • 즉 A, B, C EventArgs 클래스가 있다면, 이는 각각 다른 토픽으로 구성해야 한다.
  • 특정 작업 결과에 따라 순서를 맞춰 처리해야 하는 경우(순서가 중요한 경우), 하나의 토픽 내에서 처리한다.
    • 예 : 결제완료 -> 처리 A완료 -> 처리 B 완료 
    • type를 통해 상세 처리 과정을 별도로 구분지어 처리하는게 좋다.
    • 아래는 예 이다.

{
  "orderId": "123",
  "type": "ORDER_PAID",
  "payload": { ... }
}

• Partition : 물리적 로그 파일로, 토픽을 하드 디스크에 저장한다. 하나의 토픽은 여러 파티션으로 쪼개진다.
  • 파티션 내부에서 순서를 보장한다. 
  • 파티션을 몇개 둘 것인가? 설정을 통해 다중 파티션에 저장 가능하다.
  • 파티션의 개수는 컨슈머의 개수보다 같거나 커야 한다.
  • 한 파티션은 한 Consumer 에게만 할당되므로, 설정시 파티션 개수가 컨슈머 개수보다 크거나 같아야 한다.
    • 단 Consumer는 여러 파티션을 동시에 할당 받을 수 있다.
      
      
• Offset : 단순한 번호로서, 컨슈머가 어디까지 읽었는지를 기억한다. (위치 포인터)
  • kafka 내부 토픽으로서 kafka가 관리한다.
  • Consumer Group과 파티션 단위를 키로 하여 존재하게 된다.
  • 단일 Consumer 이면 offset이 컨슈머 마다 있지만, 여러개의 Comsumer 가 같은 그룹에 있다면, 그룹별로 offset을 공유한다.

Kafka의 메세지 전달보장 및 처리 의미

• At most once
  • 메세지가 0번 혹은 1번 처리될 수 있다.
  • 중요하지 않은 이벤트 처리시 사용한다. (예 : 메트릭, 로그 등)
  • 메세지를 poll 하고 먼저 offset commit 을 하고 처리 한다.
  • 속도가 빠르나 유실 가능성 있다.
• At least once (기본 동작 방식)
  • 메세지는 1번 이상 처리 된다.
  • 메세지를 poll 하고 처리 후 offset commit을 한다.
  • 중복 처리 가능성이 있으나, 유실되지는 않는다.
• Exactly once
  • 정확히 딱 한번만 처리된 것처럼 보이게 만드는걸로 불가능하거나 어려운 부분이므로 따로 다루지 않는다.

 

Consumer 에서 중복처리 방법 

• DB 쿼리문을 통한 방법

INSERT INTO orders (event_id, ...)
VALUES (...)
ON CONFLICT (event_id) DO NOTHING;

• 혹은 처리전에 exsist 를 통해 존재 여부를 체크하는것도 좋다.

 

Consumer Rebalance

• 파티션을 다시 나누는 과정이 발생하는것을 말한다. 이 동안은 메세지 소비가 멈추게 된다.
• Rebalance는 필연적으로 발생하는 정상과정이다.
  • 컨슈머 시작 (새 컨슈머 합류)
  • 컨슈머 종료 (정상, 비정상 모두)
  • max.poll.interval.ms 초과시
    • poll 호출간 최대 허용시간
    • 이 값이 너무 짧을경우, 비즈니스 로직을 처리하는 과정에서 이 시간이 넘어가면, 컨슈머가 죽었다고 판단한다. 
    • session.timeout.ms 및 heartbeat.interval.ms 관련
      • session.timeout.ms : heartbeat 미수신 허용 시간
      • heartbeat.interval.ms : 이 주기로 hearbeat을 브로커로 보냄
      • poll 할 때마다 heartbeat를 보내게 됨.
    • max.poll.interval.ms > session.timeout.ms > heartbeat.interval.ms 크기 순으로 설정해야 한다.
      • 그렇지 않으면 어떤거라도 timeout 발생하여 컨슈머가 죽었다고 판단, Rebalance 한다
      • 기본값 대신 비즈니스 로직 처리 시간을 고려하여 값을 설정해야 한다. 그렇지 않으면 컨슈머가 계속 죽었다고 판단하는 사태가 발생한다.
      • heartbeat는 보통 session / 3 으로 설정
  • 네트워크 문제
  • 파티션 증가시 발생 하게 된다.
• Rebalance 프로세스 
  • 모든 Consumer에게 stop 신호
  • 기존 파티션 할당 해제
  • 파티션 재 할당
  • 다시 poll 시작
• 만일 메세지를 poll 하여 처리하는 도중 발생했다면, 재시작시 다시 메세지를 가져와 처리하므로 중복처리 발생 가능성이 있다.
  • 이 때문에 위에서 설명한 방식으로 중복 처리가 되지 않도록 코드 작성이 필요하다.

 

Consumer  에러처리 전략

• 컨슈머 오류 발생시 제대로 처리 하지 않으면, 같은 메세지를 무한 재시도 하는 사태가 발생할 수 있다.
• 오류가 발생했을때 처리 패턴은 4가지 이다. 
  • 즉시 실패
  • 재시도
  • 건너뛰기
  • Dead Letter Topic (DLT)
    
    • 정상 처리할 수 없는 메세지를 격리하는 토픽
    • 원본메세지 + 예외정보를 포함한다.
    • 별도의 토픽으로 보관되며 재처리 가능하다.
      • 원본토픽명.DLT 이름을 가진다.
    • DLT 메세지는 로그분석, 운영자 수동 재처리, 별도 Consumer로 자동 재처리 한다.
• 이 중, 재시도(Kafka Retry) + DLT가 가장 많다.
  • 즉, 재시도 후 실패시 DLT로 해당 메세지를 보낸다.
  • 처리 실패 -> 설정 시간만큼 대기 -> 재시도 -> 실패시 DLT 전송 및 offset commit
  • 이를 통해 무한루프 (오류 발생 -> 재처리 순환) 방지를 한다.
  • 다수의 메세지가 있을경우, 이 메세지는 건너뛰고 다음 메세지를 수행한다.

 

Outbox 패턴

• DB와 kafka를 같이 사용할때 다음의 문제가 있을 수 있다. 
  • DB 트랜잭션 성공, kafka send 실패
  • kafka send 성공, DB 트랜잭션 롤백
  • DB 커밋 후 kafka send 시, kafka 실패하게 되면 DB 커밋을 되돌릴 수 없다. 
  • 반대의 경우, DB에는 데이터가 없는데 유령 이벤트가 생성된다.
• 이벤트를 kafka로 바로 보내지 말고, 별도의 DB 테이블에 이벤트 기록을 하자 는 패턴이다.
  • 하나의 트랜잭션에서 실제 DB작업 및 outbox 기록을  처리한다.
  • 테이블 구조 예는 다음과 같다.

CREATE TABLE outbox_event (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,

    -- Aggregate 정보
    aggregate_type   VARCHAR(50)  NOT NULL,   -- 예: ORDER, PAYMENT
    aggregate_id     VARCHAR(100) NOT NULL,   -- 예: order-123
    aggregate_version BIGINT      NOT NULL,   -- Aggregate 내부 순서 (1,2,3...)

    -- 이벤트 정보
    event_type       VARCHAR(100) NOT NULL,   -- 예: ORDER_PAID
    payload          JSON         NOT NULL,

    -- 발행 상태
    status           VARCHAR(20)  NOT NULL,   -- PENDING, SENT, FAILED

    -- 운영 메타
    created_at       TIMESTAMP    NOT NULL DEFAULT CURRENT_TIMESTAMP,
    published_at     TIMESTAMP    NULL,

    -- Aggregate 단위 중복 방지 (매우 중요)
    CONSTRAINT uk_outbox_aggregate_version
        UNIQUE (aggregate_type, aggregate_id, aggregate_version),

    -- Publisher 조회 최적화
    INDEX idx_outbox_status_created
        (status, created_at)
);

• aggregate_type 는 비즈니스 로직 관점에서의 도메인 (예 : payment, order)
• aggregate_id 는 해당 타입의 아이디로, 같은 대상임을 식별한다. (예 : order-123)
  • 이 값은 kafka의 key 와 동일 (파티션 결정요소)로 사용된다.
• 이벤트메세지 순서와 DB 순서가 다를 수 있으므로 별도의 키인 aggregate_version을 이용한다.
  • 같은 aggregate type, id안에서만 증가. 둘중 하나라도 다르면 증가하면 안됨.
• 이벤트 발행은 기존처럼 비즈니스 로직에서 하지 않고 별도의 publisher가 처리한다.
  • 주기적으로 outbox 테이블 에서 status가 PENDING 인 항목 조회
  • kafka send 수행 -> 성공시 SENT로 변경
• 스케줄러를 통해 SENT 항목을 주기적으로 지워줘야 한다.
• 같은 데이터를 여러번 가져오지 않게, Lock 을 걸어야 한다.