Kafka로 대규모 확장 설계 (1)

카프카를 알아야 하는 이유?

카프카 말고도 분산 메세징 시스템은 AWS의 키네시스 등 비숫한 게 많다. 하지만 보편적으로 사용되는 시스템이기에 알아두면 다른 시스템도 이해하기 쉬워진다

다음과 같은 효과를 기대할 수 있음

• 대용량 처리
• 고가용성
• 메세지 유실 방지 등 다양한 기능
  • 메세지를 받는 서비스가 가용하지 않을 때 카프카가 그 메시지를 가지고 있다가 가용할 때 메세지를 다시 전송하는 매커니즘을 가지고 있음

카프카의 구성 요소

프로듀서와 컨슈머

• 프로듀서는 메세지를 카프카 시스템에 발행하는 서비스
• 컨슈머는 카프카 시스템에 발행된 메세지를 읽어오는 서비스
• 오프셋은 컨슈머가 어디까지 메세지를 읽었는지를 기록해둔 데이터

---

주문 서비스, 결제 서비스가 있고 주문을 완료하면 결제가 일어나야 한다고 할 때 카프카가 중간에서 메세지를 중계하면 서비스는 다른 서비스가 있다는 것을 알 필요가 없다

프로듀서와 컨슈머는 역할이 딱 고정되어있는 것이 아니라 상황에 따라 바뀐다. 예를 들어 SAGA 패턴을 써서 결제가 실패했다는 보상 트랜잭션을 실행시키기 위해 결제 서비스가 메세지를 발행하고, 주문 서비스가 메세지를 읽어들일 수 있다

MSA에서는 같은 역할을 하는 모듈이 여러개의 인스턴스로 올라가 있는데, 이 모듈들을 묶어서 ‘서비스’라고 한다. 또한, 서비스는 카프카에서 ‘컨슈머 그룹’또는 ‘프로듀서 그룹’이라는 용어와 대응된다

카프카 시스템 안에서는 큐 안에 순서대로 메세지가 쌓여있다. 이 메세지들을 컨슈머가 어느 순서까지 읽었느냐? 가 바로 오프셋이다 → 이로 인해 메세지 유실방지 기능이 작동

브로커

카프카는 기본적으로 클러스터이다. 이 클러스터를 구성하는 인스턴스들을 ‘브로커’ 라고 한다.

Controller와 Coordinator라는 특별한 역할을 하는 브로커가 있음

• Controller : 브로커들을 관장하는 브로커
• Coordinator : 컨슈머와 내부 큐를 실제 매칭
• Bootstrap Server : 클라이언트(컨슈머와 프로듀서)가 카프카 클러스터에 접속하기 위한 진입점이 되는 브로커. 이 브로커는 다른 프로커에 대한 접속 정보를 가지고 있기에, 접속을 중계할 수 있음. (브로커 엔드포인트?)
• 카프카 시스템은 스케일러블하기에 중간에 브로커가 삭제될수도 있고, 추가될수도 있음

메세지

카프카를 통해 프로듀서에서 컨슈머로 이동하는 데이터를 의미. Key-Value 쌍으로 이루어져 있음. 보통 key는 리소스 id, value에는 리소스 상세 정보가 담기게 됨

• key값에 따라서 어떤 브로커에 담길지 정하게 됨

토픽과 파티션

• Topic : 메세지의 종류를 분류하는 기준. 컨슈머는 이 단위로 메세지를 구독. 참고로 물리적인 개념과 논리적인 개념이 합쳐진 것임. 브로커 ≠ 토픽. 브로커 하나가 토픽이 될수도 있고, 여러개가 하나의 토픽으로 묶일수도 있다. 이 개념이 없다면 컨슈머는 모든 메세지를 카프카로부터 받게 될 것
• Partition : 실제 메세지가 담겨있는 물리적인 큐, 토픽은 여러개의 파티션으로 구성. 메세지는 이 파티션에 고르게 분산되어 저장되며, 소비된다.
  • 카프카는 서로 다른 파티션에 있는 메시지 간의 순서는 보장할 수 없다. 어떤 파티션에 저장될지는 키 값의 해싱을 통해서 정함 → 따라서 메세지 간의 순서를 만들고 싶으면 같은 키 값으로 메세지를 발행해야 한다.
  • 하나의 파티션은 하나의 컨슈머만 사용할 수 있음!! → 순서 보장을 위해

컨슈머 그룹

• 토픽에 있는 메세지를 구독하는 단위
• 보통 하나의 서비스를 나타냄
• 오프셋은 컨슈머 단위가 아니라 컨슈머 그룹 단위로 만들어진다

리밸런싱

• 중간에 컨슈머가 추가/삭제되거나 새로운 파티션이 추가됐을 때 컨슈머별로 할당된 파티션을 다시 고르게 분배하는 과정
• 이 과정 중에는 컨슈머가 카프카 시스템으로부터 메세지를 읽을 수 없음 (컨슈머와 파티션 간의 연결을 모두 끊고 다시 매칭하기 때문에)

클러스터

• 고가용성과 확장성을 위해 여러 브로커를 묶어 하나의 시스템으로 구성
• 이 때문에 스케일러블하게 운영 가능

레플리케이션

• 토픽 내에 있는 브로커가 삭제될 때, 브로커 내의 파티션에 있던 메세지들은 사라지지 않는다
• 카프카는 각 파티션에 대한 복제본을 같은 토픽 내의 다른 브로커에 예비 파티션으로 두는 기능을 제공
• 따라서 파티션이 삭제되면 다른 브로커에 있던 예비 파티션을 메인 파티션으로 승격시키면 됨
• 예비 파티션을 파티션마다 몇 개 둘지 정하는 것을 Replication Factor라고 한다 (1 = 없음)
• Leader Replica → 메인 파티션 , Follower Replica → 예비 파티션

카프카 인프라

카프카 배포 모드 : ZooKeeper vs KRaft

• ZooKeeper 모드 : 카프카 초기 버전부터 사용된 중앙 제어 방식 → 아직 많이 사용
• KRaft 모드 : 카프카 2.8 버전부터 도입된 P2P 방식

Apache Kafka vs Confluent Kafka

• 아파치 카프카 : 아파치 소프트웨어에서 오픈소스에서 제공하는 순수 카프카
• Confluent 카프카 : 아파치 카프카를 기반으로 Confluent사에서 개발한 상업용 배포판

기존에 링크드인에서 카프카 시스템을 개발하여 운영하다가 아파치 재단에 이를 기부. 이후에 링크드인 개발자들이 Confluent를 설립하여 기존 카프카 시스템에 여러 기능을 추가하여 상업용으로 개발

카프카로 기존 코드 개선하기

@Service
class ReservationService(
    ...
    private val eventPublisher: DomainEventPublisher // 새로 의존
) {

	eventPublisher.publish(ReservationCreatedEvent(
	    saved.id!!,
	    concert.id!!,
	    saved.seatNumber,
	    reserver.id!!
	))

...

@Component
class ReservationAfterCommitEventListener(
    private val concertRankingPort: ConcertRankingPort,
    private val dataFlatformPort: DataFlatformPort
) {

    private val log = KotlinLogging.logger { }

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    fun handleReservationCreate(event: ReservationCreatedEvent) {
        dataFlatformPort.transferData(CreateReservationInfo(
            event.reservationId,
            event.concertId,
            event.memberId
        ))
        concertRankingPort.checkAndMarkSoldOut(event.concertId)
    }
}

• 이전 스텝에서 위처럼 이벤트 기반으로 리팩토링을 끝마침. 관심사의 분리가 잘 이루어졌지만, 외부 API를 호출하는 dataFlatformPort 에서 호출이 실패한다면? 호출을 재전송하는 책임까지 우리가 가져야하나?
  • 그러면 전달을 실패하고 재전송하는 로직은 또 어디에 만드나?
  • 이 때 카프카를 사용하자. 카프카가 알아서 장애상황 해제 이후 적재되었던 주문정보를 알아서 전달한다

@Component
class ReservationAfterCommitEventListener(
    private val concertRankingPort: ConcertRankingPort,
    private val kafakaProducer: KafkaProducer
) {

    private val log = KotlinLogging.logger { }

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    fun handleReservationCreate(event: ReservationCreatedEvent) {
        kafakaProducer.produce(CreateReservationInfo(
            event.reservationId,
            event.concertId,
            event.memberId
        ))
        concertRankingPort.checkAndMarkSoldOut(event.concertId)
    }
}

• 예약 생성 이벤트를 카프카로 전달하면 책임은 여기서 끝
• 이후에 일어나는 부가 로직에 대한 관심사가 제거됨

실무 적용

쿠폰 선착순 발급 시나리오

• 프로듀서 : 쿠폰 발급 API 서비스
• 컨슈머 : 쿠폰 발급 서비스
• 선착순 쿠폰 발급을 하면 순간적으로 트래픽이 몰림. 이 때 순서보장을 위해 쿠폰 발급 이벤트의 key를 모두 같게 만들면 하나의 파티션에 이벤트가 몰리게 될 것이고, 하나의 파티션에는 하나의 컨슈머만 연결되어있기 때문에 인스턴스 하나에만 트래픽이 순간적으로 몰리게 될 것이다. 실무에서는 이 상황을 막기 위해 key를 같게 두지 않고 여러 파티션에 이벤트를 분산시키는 트릭을 사용한다. 순서 보장이 완벽하게 되지는 않지만 soft하게는 지켜짐

대규모 대기열

• 순서 보장을 단순하게 처리하고 싶다면 1개의 파티션만을 이용하여 카프카를 활용할 수 있음
• 단, 실무에서는 Redis를 활용한 구현이 보통은 효율적임

카프카 더 깊게 알아보기

Commit

• 파티션 내부적으로 어디까지 읽었는지 기록되는 오프셋을 이동시키는 행위를 커밋이라고 함. 컨슈머가 메세지를 컨슘하고 잘 읽었다는 응답을 보낼 때 발생
• Auto Commit
  • 카프카 컨슈머는 기본적으로 특정 주기마다 현재까지 처리한 메시지의 오프셋을 자동으로 커밋함
  • 컨슈머가 메세지를 읽었지만 아직 처리하지 못했는데 주기 때문에 커밋이 발생한 다음 장애가 발생하면? → 누락되는 메세지 발생
  • 메세지 처리가 완료된 후 커밋을 하지 못했는데 장애가 발생하면? → 중복 처리되는 메세지 발생
• Munual Commit
  • 개발자가 코드 내에서 직접 오프셋을 커밋하는 방식 (실전에서 사용)
  • 동기 방식과 비동기 방식이 있는데, 비동기 방식에서는 커밋 실패에 대한 추가적인 오류 처리가 필요할 수 있음

DLQ (Dead Letter Queue)

• 메세지에 중대한 문제가 있어서 재시도해도 처리할 수 없는 경우가 되었을 때, 이러한 메세지들을 별도로 모아두는 전용 토픽을 DLQ라고 함
• 오류 메세지를 유실시키지 않고 따로 모아두면, 이후 모니터링이 쉬워지고 오류가 해결된 후 재처리가 가능해짐

Retention 정책

• 카프카 브로커는 메세지를 한 번 저장하면 컨슈머가 메세지를 소비했는지 여부와 관계없이 일정 기간동안 메세지를 보관한다.
  • 이렇게 메세지를 보관함으로서 새로운 컨슈머 그룹이 구독을 시작하면 과거 데이터부터 읽을 수 있고, 컨슈머 장애나 버그 발생시 재처리가 수월해짐
• 하지만 공간 문제 때문에 메세지를 무한정 보관할수는 없으므로 시간 기반(발행된 지 얼마나 됐는지)이나 크기 기반(토픽의 파티션 크기)으로 삭제를 수행한다

멱등성 정책

• 카프카는 메세지를 최소 한번을 보장하지만, 정확히 한번을 보장하지는 않음
  • 같은 메세지를 두번 받을수 있다는 소리
• 중복 메세지를 받으면 문제가 되는 트랜잭션이 있을 수 있음
  • 이러한 경우는 “멱등성이 없다” 라고 함
  • 예를 들어 “잔액을 100원 더하라” 는 중복되면 잔액이 여러번 더해질수 있기 때문에 “잔액을 1100원으로 만들어라”가 멱등성을 보장할 수 있음
• 멱등성을 확보하기 위해서는 메세지마다 유니크한 ID (UUID)를 부여하고, 컨슈머는 이 ID를 기준으로 이미 처리된 메세지인지 확인하여 중복 처리를 방지할 수 있음 → 데이터베이스에 처리된 메세지 테이블을 만들어두고 조회하는 추가적인 I/O가 생길 수 있음
  • 위의 잔액 예시처럼 특정 상태로의 전환만 허용하거나 최종 상태를 덮어쓰는 방식만 허용하여 멱등성을 보장할 수 있음

Zero Payload vs Full-Payload

• 업데이트된 엔티티(객체?)의 ID만 받을 것인가 (zero) 아니면 업데이트된 모든 내용을 받을 것인가 (full)
• 보통 zero payload를 많이 사용함 → 업데이트되었다는 메세지를 전달받는 사이에 또 업데이트가 일어날수도 있기 때문에
  • 더해서, 대용량 트래픽 환경에서는 메세지 하나의 크기가 전체 시스템 성능에 영향을 끼칠수도 있음
  • zero payload에서는 ID를 통해 추가적인 조회를 수행해야함

 

/** 최근 gemini의 성능이 심각하게 떨어진것을 몸으로 체감중... 블로그 포스터도 글이 만족스럽지가 않아.. ㅠㅠ 이건 내 초안을 그대로 올린 것임 */