kafka에 대해 공부해보자

💡요즘 왜 다들 카프카, 카프카 하는 걸까?

    - 대규모 실시간 데이터 스트리밍을 위한 분산 메세징 시스템

    - 높은 처리량 및 개발 효율을 위한 분산 시스템에서 고가용성과 유연함을 갖춘 연계시스템이 필요

 

카프카 Overview

향해 99

1. Producer & Consumer

• Producer - 메세지를 카프카 브로커에 적재(발행)하는 서비스
• Consumer - 카프카 브로커에 적재된 메시지를 읽어오는(소비) 서비스

         - 메세지를 읽을 때마다 파티션 별로 offset을 유지해 처리했던 메세지의 위치를 추적

         - CURRENT-OFFSET

            컨슈머가 어디까지 처리했는지를 나타내는 offset이며, 동일한 메세지를 재처리하지 않고, 처리하지 않은 메세지를 건너뛰지 않기위해 마지막까지 처리한 offset을 저장(커밋) 해야함

        - 만약 오류가 발생하거나 문제가 발생할 경우, 컨슈머 그룹 차원에서 --reset-offsets 옵션을 통해 특정시점으로 offset을 되돌릴 수 있음

 

2. Broker

• 카프카 서버 Unit
• Producer의 메세지를 받아 offset 지정 후 디스크에 저장
• Consumer의 파티션 Read에 응답해 디스크의 메세지 전송
• Cluster 내에서 각 1개씩 존재하는 Role Broker

        - Controller

           다른 브로커를 모니터링하고 장애가 발생한 Broker에 특정 토픽의 Leader 파티션이 존재한다면, 다른 브로커의 파티션 중 Leader를 재분배하는 역할을 수행

        - Coordinator

           컨슈머 그룹을 모니터링하고 해당 그룹 내의 특정 컨슈머가 장애가 발생해 매칭된 파티션의 메세지를 Consume 할 수 없는 경우, 해당 파티션을 다른 컨슈머에게 매칭해주는 역할 수행(Rebalance)

 

3.Message

• 카프카에서 취급하는 데이터의 단위로 <Key, Message> 형태로 구성

4. Topic & Partition

향해99

• Topic은 메세지를 분류하는 기준이며 N개의 Partition으로 구성
• Partition에 발행된 순서대로 컨슘함으로써 순차처리를 보장

         - 대용량 트래픽을 파티션의 개수만큼 병렬로 처리할 수 있어 빠른 처리 가능

         - 전체 메세지의 발행 순차처리를 보장하지 않지만, 같은 파티션의 메세지를 대해서는 순차처리를 보장

         - 동시성 제어의 개념을 생각했을 때, 동시에 처리되면 안되는 자원의 Id 등을 메세지의 키로 설정하면 순차처리가 보장되어야 하는 케이스는 보장도 되면서 병렬 처리로 높은 처리량을 보장한 하이브리드

• Producer에서 메세지를 발행할 때, 저장될 Partition을 결정하기 위해 메세지의 키 해시를 활용하며, 키가 존재하지 않을 경우 균형 제어를 위해 

key: "842"
hash: "55478" // "842".hashCode()
partitionCnt: "3"

targetPartition: 2 // 55478 % 3

- Partitioner

  메세지를 발행할 때, 토픽의 어떤 파티션에 저장될 지 결정하며 Producer 측에서 결정. 특정 메세지에 키가 존재한다면 키의 해시 값에 매칭되는 파티션에 데이터를 전송함으로써 키가 같은 메시지를 다건 발행하더라도, 항상 같은 파티션에 메세지를 적재해 처리순서를 보장할 수 있음.

 

- 한 Partition은 하나의 컨슈머에서만 컨슘할 수 있음

    - 하나의 파티션에 여러개의 컨슈머가 메시지를 컨슘하면 메시지 처리의 순서를 보장할 수 없게 됨

 

5. Consumer Group

• 하나의 토픽에 발행된 메세지를 여러 서비스가 컨슘하기 위해 그룹을 설정

       - 하나의 주문완료 메세지를 결제서비스에서도, 상품서비스에서도 컨슘

• 보통 소비 주체인 Application 단위로 Concumer Group을 생성, 관리함
• 같은 토픽에 대한 소비주체를 늘리고 싶다면, 별도의 컨슈머 그룹을 만들어 토픽구독..

향해 99

- 파티션의 개수가 그룹 내 컨슈머 개수보다 많다면 잉여 파티션의 경우 메세지가 소비될 수 없음을 의미함

- (참고) 포틱의 Patition 개수와 Consumer 개수에 따른 소비

향해99

 

향해99

 

향해99

 

6. Rebalancing

• Consmuer Group의 가용성과 확장성을 확보해주는 개념
• 특정 컨슈머로부터 다른 컨슈머로 파티션의 소유권을 이전시키는 행위
• e.g. Consumer Group 내에 Concumer가 추가될 경우, 특정 파티션의 소유권을 이전시키거나 오류가 생긴 Consumer로부터 소유권을 회수해 다른 Consumer에 배정함
• (주의) 리밸런싱 중에는 컨슈머가 메세지를 읽을 수 없음

Rebalancing Case

1. Concumer Group 내에 새로운 Consumer 추가

2. Consumer Group 내의 특정 Consumer 장애로 소비 중단

3. Topic 내에 새로운 Partition 추가

 

7. Cluster

• 고가용성 (HA)를 위해 여러 서버를 묶어 특정 서버의 장애를 극복할 수 있도록 구성
• Broker가 증가할 수록 메시지 수신, 전달 처리량을 분산시킬 수 있으므로 확장에 휴리
• (동작중인 다른 Broker에 영향 없이 확장이 가능하므로, 트래픽 양의 증가에 따른 브로커 증설이 손쉽게 가능)

8. Replication

• Cluster의 가용성을 보장하는 개념
• 각 Partition의 Replica를 만들어 백업 및 장애 극복
• Leader Replica - 각 파티션은 1개의 리더 Replica를 가진다. 모든 Producer, Consumer 요청을 리더를 통해 처리되게 하여 일관성을 보장한다.
• Follwer Replica - 각 파티션의 리더를 제외한 Replica이며 단순히 리더의 메세지를 복제해 백업한다. 만일, 파티션의 리더가 중단되는 경우 팔로워 중 하나를 새로운 리더를 선출한다.

비동기 메세지 통신으로 데이터를 처리할 때 주의할 점

비즈니스 로직 실행 이후에 이벤트가 발행되지 않을 경우, 해당 이벤트를 바라보니 Consumer 또한 본인의 비즈니스 로직을 수행할 수 없으므로 전체적인 비즈니스 흐름에 Hole이 생기거나 데이터 정합성에 문제가 생길 수 있다.

 

카프카와 비슷한 시스템

면접 단골 질문 [Messaging System 간 차이점]

Redis

• pub/sub 을 통해 MQ 구현 가능
• publisher 가 채널에 메세지를 게시 → 모든 채널 subscriber 가 메세지를 받아 처리
• channel : 이벤트 저장 x, subscriber 없으면 소실, subscriber 는 여러 channel 구독 가능

RabbitMQ

• AMQP 프로토콜의 메세지 큐
• producer 메세지 전송→ consumer 가 Queue 의 메세지 수신
• → exchange 가 키에 맞게 Queue 에 분배
• MQ 서버와 Queue 내용이 생명주기가 같음 → 종료 시 모두 삭제

Kafka

• pub-sub 기반의 메세지 발행/구독 시스템
• 파티션을 여러개 두어서 토픽의 메세지 쓰기 부하를 해소할 수 있음. ( 병렬 처리 )
• 또한 이 때문에 각 메세지의 처리 순서를 보장하지 못함. ( 고정 key 를 넣어서 동일 파티션에 순서대로 들어가게끔 할 수 있음 )
• 파티션은 늘린 후에 줄일 수 없음.
• 이전 메세지를 다시 읽어 올 수 있음. 이 경우, 중복 메세지 처리 등에 대해서도 고민해야함.

3. 카프카 비동기 메세지 통신을 통한 책임 분리

지난주에 해치웠다고 생각했던 로직을 다시한번 불러와볼까요?

개선로직 최종안

class OrderPaymentService {
	
	@Transactional
	public void pay() {
		decreaseUserPoint(); // 유저 포인트 차감
		updateOrderStatus(); // 주문 상태 변경
		savePayment(); // 결제 정보 저장
		eventPublisher.publish(new orderPaidEvent());
	}
}

class OrderPaidEventListener {

	@Async
	@TransactionalEventListener(phase = AFTER_COMMIT)
	public void sendOrderInfo() {
			try {
				dataPlatformMockApiClient.sendOrder();
			} catch(Exception e) {
				log.error("주문정보전달에 실패했어요~잉");
			}
	}
}

• 서비스 코드도 깔끔해지고, 외부 로직이 어떤것들이 있는지 관심사가 분리됨
• 트랜잭션 커밋 이후에 비동기적으로 수행되기 때문에 외부 api에 영향을 정확하게 제거함
• 와 해치웠나!!??
• 그런데… 데이터 수집 플랫폼이 일시적인 장애가 발생해서 api가 실패했다면?
  • 데이터 수집 플랫폼으로 데이터를 재 전달해줘야하는 책임은 우리한테 있는 억울한 느낌..
  • 재전송 로직은 또 어디다 만들지…

데이터 수집 플랫폼의 메세지의 수신불가 상황의 책임을 우리가 갖지 않으려면?? 카프카로 해결이 가능하다고??

• 데이터 수집 플랫폼이 일시적인 장애상황일 때에도 카프카에 주문정보를 정상적으로 적재해둔다면, 장애상황 해제 이후 적재되었던 주문정보를 “알아서” 컨슘해가면 되지 않을까?]

 

개선로직 최종안의 최종안

class OrderPaymentService {
	
	@Transactional
	public void pay() {
		decreaseUserPoint(); // 유저 포인트 차감
		updateOrderStatus(); // 주문 상태 변경
		savePayment(); // 결제 정보 저장
		eventPublisher.publish(new orderPaidEvent());
	}
}

class OrderPaidEventListener {

	@TransactionalEventListener(phase = AFTER_COMMIT)
	public void sendOrderInfo() {
			kafkaPublisher.publishOrderInfo();
	}
}

• 주문정보를 카프카로 전달한다면 데이터 수집 플랫폼에서 메세지를 잘 받았던 못 받던 내 책임은 끝!!
• 비동기 처리하지 않나요?
  • 웹서비스의 API는 서버의 자체 로직과 데이터를 적재하는 비용이 함께 포함되어 latency가 안정적이지 않을 수 있으나, 일반적으로 카프카에 메세지를 발행할때는 별도 로직 없이 발행된 메세지를 저장만 하기에 발행에 대한 비용이 더 적다고 판단
  • 비동기로 처리하기 위해 별도의 스레드에서 컨텍스트 스위칭하는 비용이 오히려 더 클 수도 있음
• 카프카에 메시지 발행이 실패했다면?
  • … 진짜 언제까지 실패 고민 해야하니…
  • 일반적인 웹 서비스보다는 고가용성에 특장점을 갖는 카프카가 더 안정적이지 않을까?
  • 근데… 그래도… 트랜잭션이 커밋된 이후에 카프카 메시지 발행을 하는데 이게 실패하면…?

 

Transactional Messaging

• Eventual Consistency - 결과적 일관성
  • 언젠가 모든 서비스 간의 데이터 정합성은 맞게 된다
  • 분산 시스템 / 비동기 메세징 환경에서 추구하는 방향
  • 대량의 트래픽을 수용할 수 있도록 작업의 단위 (Transaction) 를 작게, 비동기로 구성
• 비즈니스 로직 수행 + 후속 이벤트 발행을 원자적으로 함께 수행하는 방식
• Transactional Messaging 의 대표적인 패턴
• (1) Transactional Outbox Pattern

향해 99

• 도메인 로직이 성공적으로 수행되었다면, 이에 해당하는 이벤트 메세지를 Outbox Table 이라는 별도의 테이블에 저장하여 함께 Commit
• 동일한 트랜잭션 내에서 이벤트 발행을 위한 Outbox 데이터 적재까지 진행해 이벤트 발행에 대해 보장
• 이벤트 발행 상태 또한 Outbox 데이터에 존재하므로, 배치 프로세스 등을 이용해 미발행된 데이터에 대한 Fallback 처리가 용이

(2) Change Data Capture ( CDC )

향해 99

• 데이터베이스의 변경 사항을 모니터링하고 ( e.g. Debezium ) 다른 시스템으로 해당 변경사항에 대한 내용을 전파하는 소프트웨어 프로세스
• DB 에서 데이터가 변경되는 것을 감지하고, 해당 변경 건에 관한 이벤트를 발행

Transactional Outbox Pattern

E-commerce 의 주문과 재고 처리에 대해 고려해보자.

상대적으로 Lock 을 이용한 정합성 제어 등 대량의 트래픽이 몰려들었을 때에 제어가 힘든 재고 처리 로직을 주문 로직과 분리해낼 방법을 고려해보고, 본인의 관심사를 온전히 처리할 수 있도록 시스템을 설계해보면 아래와 같이 설계해 볼 수 있다.

 

향해 99

[주문 서비스]
// 주문 처리
Tx 1 {
	주문 생성 ( Order.Status = INIT(처리중) )
	아웃박스 생성 ( Outbox.Status = INIT )
	이벤트 발행 ( Application Event Publish )
}
// 주문 이벤트 발행
Application Event Listener {
	아웃박스 데이터 Kafka 에 전송
}

[Consumer 1 : 아웃박스 서비스]
// 아웃박스 처리
// 성공적으로 Kafka 에 이벤트가 발행되었음을 보장
Tx 1-1 {
	아웃박스 Update ( Outbox.Status = PUBLISHED )
}

[재고 서비스]
// 재고 처리
// Kafka 의 이벤트를 성공적으로 수취해 비즈니스 로직을 수행했음을 보장
Tx 2 {
	처리정보 생성 ( OrderId 에 대한 Processed 정보 생성 Processed.Status = RECEIVED )
	상품 재고 Update
	완료 이벤트 발행
}

Tx 2-1 {
	처리정보 Update ( Processed.Status = SUCCESS )
}

이벤트 발행 및 처리를 보장하기 위한 설계

• Publisher
  • 비즈니스 로직 수행
  • Outbox 저장
  비즈니스 로직 수행과 Outbox 저장을 동일한 트랜잭션 내에서 수행함으로써 비즈니스 로직 성공 = 이벤트 발행 을 항상 보장함
• Consumer 1
  • Outbox 변경
  카프카에 성공적으로 이벤트가 발행되었다면 Consumer 1 은 이를 읽어 Outbox 의 상태를 갱신해줌으로써 이벤트가 성공적으로 발행되었음을 보장
• Consumer 2
  • 비즈니스 로직 수행
  • Processed 저장
  이벤트를 처리하는 주체가 이벤트를 성공적으로 구독하였고 이벤트에 대한 비즈니스 로직을 처리하였음을 보장

개선로직 최종안의 최종안의 최종안

class OrderPaymentService {
	
	@Transactional
	public void pay() {
		decreaseUserPoint(); // 유저 포인트 차감
		updateOrderStatus(); // 주문 상태 변경
		savePayment(); // 결제 정보 저장
		orderCreatedEventRepository.save(orderCreatedEvent) // outbox 저장
		eventPublisher.publish(new orderPaidEvent());
	}
}

class OrderPaidEventListener {

	@TransactionalEventListener(phase = AFTER_COMMIT)
	public void sendOrderInfo() {
			kafkaPublisher.publishOrderInfo();
	}
}


class KafkaConsumer(orderCreatedMessage) {
	orderCreatedEventRepository.findById(message.id).published() // outbox 수정
}


// 배치프로세스 등으로 처리하면 좋겠죠.
@Scheduler(fixedRate = 5000) 
orderCreatedEventRepository.findAllByStatus(INIT).forEach {
	if(it.createdAt < now() - 5minutes) {
		slackApiClient.sendMessageToDeveloper();
		kafkaPublisher.publishOrderInfo();
	}
}