Skip to content

01.Kafka.md

Latest commit

 

History

History
419 lines (276 loc) · 16 KB

01.Kafka.md

File metadata and controls

419 lines (276 loc) · 16 KB

Kafka

🛰️ Kafka란?

Kafka는 대용량 데이터를 빠르게 처리하고, 실시간 스트리밍을 지원하는 분산 메시징 시스템이다.

❓ Kafka를 사용하는 이유

Kafka는 대규모 트래픽을 처리해야 하는 시스템 또는 MSA 환경에서 높은 성능, 유연한 확장성, 고가용성을 제공한다.
또한, 이벤트 기반 아키텍처를 구현하여 서비스 간 느슨한 결합을 가능하게 한다.

Kafka를 사용하는 주요 이유는 다음과 같다.

  • 높은 처리량: 파티션 기반 병렬 처리 구조로 초당 수십만 건의 이벤트 처리 가능
  • 내구성 보장: 메시지를 디스크에 저장하고 복제하여 데이터 손실 방지
  • 순서 보장: 같은 파티션 내에서는 메시지 순서를 보장

위와 같은 장점을 통해 Kafka는 다양한 실시간 데이터 처리 시스템에서 핵심 인프라로 사용되고 있다.

⚡️ Kafka 구성 요소

Kafka는 다음과 같은 주요 구성 요소들로 이루어져 있다.

Broker (브로커)

Broker는 하나의 Kafka 서버 인스턴스를 의미한다.
Producer로부터 메시지를 수신하고, offset을 기준으로 디스크에 저장한다.
Consumer의 요청에 따라 저장된 메시지를 전달하는 역할을 한다.

Producer (프로듀서)

Producer는 메시지를 브로커에 발행(Publish) 하는 역할을 한다.
기본적으로 Kafka는 어떤 Partition에 메시지를 기록할지 자동으로 결정하지만,
key 기반 라우팅 또는 사용자 정의 Partitioner를 통해 특정 Partition으로 메시지를 전송할 수 있다.

Consumer (컨슈머)

Consumer는 Kafka Broker에 저장된 메시지를 소비(Consume) 하는 역할을 한다.
하나 이상의 Topic을 구독하며, 각 Partition 단위로 offset을 관리하여 메시지를 순차적으로 읽는다.

  • Current Offset: 현재 읽고 있는 메시지의 위치
  • Committed Offset: 실제로 처리 완료된 메시지의 위치

offset을 적절히 관리하지 않으면 메시지 중복 소비 또는 유실이 발생할 수 있다.

Topic (토픽)

Topic은 Kafka 메시지를 분류하는 논리적 단위이다.
하나의 Topic은 여러 개의 Partition으로 구성되며,
Producer는 Topic 단위로 메시지를 발행하고, Consumer는 Topic을 구독하여 메시지를 수신한다.

Partition (파티션)

Partition은 Topic을 구성하는 물리적 단위이다.
각 Partition은 메시지 순서를 보장하며, 병렬 처리를 가능하게 한다.

동일한 리소스를 동시에 처리하면 안 될 경우, 해당 리소스를 Partition Key로 지정하여 동시성 제어와 순차 처리를 보장할 수 있다.

Consumer Group (컨슈머 그룹)

Consumer Group은 하나의 Topic을 여러 Consumer가 병렬로 처리할 수 있도록 구성하는 단위이다.
하나의 Partition은 오직 하나의 Consumer에게만 할당되므로,
Consumer Group을 통해 동일한 메시지를 중복 소비하지 않으면서 처리 부하를 분산할 수 있다.

예를 들어, 주문 완료 메시지를 결제 서비스, 배송 서비스 등 서로 다른 서비스들이 독립적으로 소비하고자 할 때 각각의 Consumer Group을 구성한다.

일반적으로 애플리케이션 단위로 Consumer Group을 생성한다.

Rebalancing (리밸런싱)

Rebalancing은 Consumer Group 내에서 Partition의 소유권이 변경되는 과정이다.
다음과 같은 상황에서 자동으로 발생한다.

  • Consumer Group에 새로운 Consumer가 추가될 때
  • 기존 Consumer가 장애로 인해 사라질 때
  • Topic에 새로운 Partition이 추가될 때

Rebalancing은 Kafka의 확장성과 가용성을 높이는 핵심 메커니즘이지만,
진행 중에는 일시적인 메시지 소비 중단이 발생할 수 있다.

Cluster (클러스터)

Cluster는 여러 Broker를 하나의 논리적 단위로 묶어 고가용성과 확장성을 제공한다.
Broker를 수평적으로 확장함으로써 메시지 수신 및 저장을 분산할 수 있으며,
운영 중인 Broker에 영향을 주지 않고 유연하게 증설이 가능하다.

Replication (레플리케이션)

Replication은 Kafka의 데이터 내구성을 확보하기 위한 핵심 기능이다.
각 Partition은 하나의 Leader Replica와 여러 개의 Follower Replica로 구성된다.

  • Leader Replica: 메시지의 쓰기/읽기를 담당하며, 단 하나만 존재한다.
  • Follower Replica: Leader의 데이터를 복제하여 백업 역할을 수행하며, 장애 발생 시 동기화된 Follower가 Leader로 승격된다.

Leader와 동기화되지 않은 Follower는 Leader로 승격될 수 없다.

img_1

🔁 Kafka 구성 요소의 흐름

카프카의 주요 구성 요소는 메시지의 발행, 저장, 소비 과정을 순차적으로 담당하며, 그 흐름은 다음과 같다.

  1. Producer는 메시지를 특정 Topic에 발행한다.
  2. Broker는 수신한 메시지를 지정된 Partition에 저장한다.
  3. Consumer는 Topic을 구독하고, 할당된 Partition으로부터 메시지를 소비한다.

파티션과 컨슈머 간의 관계

카프카의 병렬 처리 성능과 리소스 활용도는 Partition 수와 Consumer 수의 비율에 따라 달라진다.

프로듀서 수는 기본적으로 성능과 무관하며, 브로커에게 얼마나 잘 발행하는지가 중요하다.

1) 파티션 수 > 컨슈머 수

이 경우, 하나의 Consumer가 여러 Partition을 담당하게 된다.
메시지 처리는 순차적으로 이루어지므로 처리 속도가 느려질 수 있지만, 이를 의도적으로 지연시켜(throttling)
DB에 과도한 부하가 걸리지 않도록 조절할 수 있다.

✅ DB에 트래픽을 분산시키고자 할 때 유용하다.

img_2

2) 파티션 수 = 컨슈머 수

각 Consumer가 하나의 Partition을 전담하는 가장 이상적인 구조다.
모든 Partition이 병렬로 처리되므로 최대 처리 성능을 확보할 수 있다.

✅ 처리량 극대화가 필요한 환경에 적합하다.

img_3

3) 파티션 수 < 컨슈머 수

일부 Consumer는 할당받을 Partition이 없어 유휴 상태(idle) 로 대기하게 된다.
이러한 Consumer는 장애 발생 시 자동 리밸런싱을 통해 대체 Consumer로 전환될 수 있어, 고가용성 확보에 기여한다.

✅ 장애 복구 및 리밸런싱을 고려한 여유 전략으로 활용 가능하다.

img_4

⚠️ Kafka 주의사항

Kafka의 고가용성과 확장성을 제대로 활용하기 위해서는 다음과 같은 주의사항을 고려해야 한다.

메시지 유실 방지

도메인 로직이 완료된 후 이벤트가 정상적으로 발행되지 않으면, 해당 이벤트를 수신하는 Consumer가 로직을 수행할 수 없어 전체 시스템의 정합성이 깨질 수 있다.
이러한 상황을 방지하기 위해, 서비스 로직과 이벤트 발행을 원자적으로 함께 실행해야 하며, 이를 트랜잭셔널 메시징(Transactional Messaging) 이라고 한다.

트랜잭셔널 메시징을 구현하는 대표적인 방법은 다음과 같다:

  • Outbox 패턴: 도메인 로직과 이벤트 발행을 분리하여, 이벤트를 Outbox 테이블에 먼저 저장한 뒤, 별도의 프로세스가 이를 Kafka로 발행한다.
  • Change Data Capture (CDC): 데이터베이스의 변경 사항을 실시간으로 감지하여 Kafka 이벤트로 변환 및 전송하는 방식이다.

Outbox 패턴을 사용하면 도메인 로직 내에서 이벤트가 기록되므로, 도메인 간 결합도가 높아질 수 있다.
본 프로젝트에서는 이러한 결합을 최소화하기 위한 Outbox 구현 전략을 적용하였다.

메시지 중복 처리

Kafka는 리밸런싱 등 내부 이벤트로 인해 메시지를 중복 소비할 수 있다.
따라서 도메인 로직은 중복된 메시지가 들어오더라도 동일한 결과를 보장할 수 있도록 멱등성(Idempotency) 을 가져야 한다.
이러한 처리를 위한 전략 중 하나가 Inbox 패턴이다.

메시지 실패 대응

Kafka 메시지를 소비하는 과정에서, 역직렬화 오류 또는 비즈니스 로직 예외가 발생할 수 있다.
이러한 실패에 대비해 Dead Letter Queue(DLQ) 를 구성하면, 실패한 메시지를 별도의 토픽에 저장하여 이후 재처리할 수 있다.

📪 Outbox 패턴 구현

Outbox 패턴은 Kafka 사용 시 메시지 발행의 정합성을 보장하기 위한 핵심 전략 중 하나이다.
다음은 본 프로젝트에 적용된 Outbox 패턴의 구현 방식이다.

Outbox 테이블 및 엔티티 클래스

아래는 Outbox 테이블의 DDL과 JPA 엔티티 클래스 정의이다.

CREATE TABLE outbox (
    outbox_id BIGINT AUTO_INCREMENT PRIMARY KEY,
    event_id VARCHAR(50) NOT NULL,                   -- 이벤트 ID
    event_type VARCHAR(100) NOT NULL,                -- 이벤트 타입
    partition_key BIGINT NOT NULL,                   -- 파티션 키
    payload TEXT NOT NULL,                           -- 이벤트 페이로드
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP   -- 생성 시간
);
@Entity
public class Outbox {

    @Id
    @Column(name = "outbox_id")
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    private String eventId;

    @Enumerated(EnumType.STRING)
    @Column(length = 100)
    private EventType eventType;

    private Long partitionKey;

    @Lob
    @Column(columnDefinition = "TEXT")
    private String payload;

    private LocalDateTime createdAt;
}

Outbox 도메인 이벤트 발행

Outbox 이벤트는 Spring의 ApplicationEvent 기반으로 발행된다. OutboxEvent 클래스는 Auto(트랜잭션 존재)와 Manual(트랜잭션 없음) 두 가지 케이스를 제공한다.

  • Auto: 도메인 로직 내 @Transactional이 포함된 경우
  • Manual: @Transactional이 없는 경우 수동 발행

Outbox 객체를 생성할 때는 직렬화 포맷의 일관성을 유지하기 위해 Event 클래스를 활용한다.
이벤트 ID는 UUID로 생성되며, 해당 ID와 이벤트 타입, 페이로드를 Event 객체에 담아 JSON 직렬화된 문자열로 저장한다.
이 방식은 Kafka 메시지 소비 측에서도 일관된 역직렬화 구조를 유지할 수 있게 한다.

public class OutboxEventPublisherImpl implements OutboxEventPublisher {

    private final ApplicationEventPublisher eventPublisher;
    
    @Override
    public <T> void publishEvent(EventType type, Long partitionKey, T payload) {
        Outbox outbox = create(type, partitionKey, payload);
        OutboxEvent.Auto event = OutboxEvent.Auto.of(outbox);

        eventPublisher.publishEvent(event);
    }

    @Override
    public <T> void publishManualEvent(EventType type, Long partitionKey, T payload) {
        Outbox outbox = create(type, partitionKey, payload);
        OutboxEvent.Manual event = OutboxEvent.Manual.of(outbox);

        eventPublisher.publishEvent(event);
    }

    private <T> Outbox create(EventType type, Long partitionKey, T payload) {
        String eventId = UUID.randomUUID().toString();
        return Outbox.create(
            eventId,
            type,
            partitionKey,
            Event.of(eventId, type, payload).toJson()
        );
    }
}

Outbox의 payload는 다음과 같은 형식으로 직렬화된다.

{
    "eventId": "fee5d5ce-cdf7-4797-8baa-0cad19f80153",
    "eventType": "ORDER_CREATED",
    "payload": {
        // 실제 이벤트 페이로드 내용
    }
}

Outbox 도메인 이벤트 리스너

이벤트 발행 후, Outbox를 처리하는 리스너 클래스이다.

  • Auto 이벤트는 트랜잭션 커밋 전 Outbox 저장 → 커밋 후 Kafka 발행
  • Manual 이벤트는 즉시 Outbox 저장 후 Kafka로 발행
public class OutboxEventListener {
    
    @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
    public void createOutbox(OutboxEvent.Auto event) {
        log.info("createOutbox - Auto 아웃 박스 이벤트 수신: {}", event.getOutbox().getTopic());
        outboxService.createOutbox(event.getOutbox()); 
    }
    
    @Async
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void produceEvent(OutboxEvent.Auto event) {
        log.info("produceEvent - Auto 아웃 박스 이벤트 수신: {}", event.getOutbox().getTopic());
        outboxService.produceEvent(event.getOutbox());
    }
    
    @Async
    @EventListener
    public void handle(OutboxEvent.Manual event) {
        log.info("produceEvent - Manual 아웃 박스 이벤트 수신: {}", event.getOutbox().getTopic());
        outboxService.createOutbox(event.getOutbox());
        outboxService.produceEvent(event.getOutbox());
    }
}

Outbox 카프카 이벤트 리스너

Kafka로 전송된 Outbox 이벤트는 아래 리스너에서 수신되며,
수신된 이벤트의 eventId를 기반으로 Outbox 테이블에서 데이터를 제거한다.

public class OutboxMessageEventListener {

    private final OutboxService outboxService;

    @KafkaListener(topics = {
        Topic.COUPON_PUBLISH_REQUESTED,
        Topic.ORDER_COMPLETE_FAILED,
        Topic.ORDER_COMPLETED,
        Topic.ORDER_CREATED,
        Topic.PAYMENT_PAID,
        Topic.PAYMENT_FAILED,
        Topic.PAYMENT_CANCELED,
    }, groupId = GroupId.OUTBOX)
    public void handle(String message, Acknowledgment ack) {
        log.info("아웃 박스 이벤트 수신 {}", message);

        Event<?> event = Event.of(message, Object.class);
        outboxService.clearOutbox(event.getEventId());
        ack.acknowledge();
    }
}

🧨 DLQ(Dead Letter Queue) 구성

Kafka에서는 메시지 소비 중 예외가 발생했을 때, 일정 횟수 재시도 후에도 실패할 경우 해당 메시지를 DLQ(Dead Letter Queue) 로 전송하여
이후 별도로 분석하거나 재처리할 수 있도록 구성할 수 있다.

Spring Kafka에서는 Bean 설정을 통해 DLQ를 손쉽게 구성할 수 있으며, 아래는 기본적인 설정 예시이다.

DLQ 설정 예시

다음 설정은 1초 간격으로 최대 3회 재시도한 후에도 실패한 메시지를 DLQ로 전송하는 구성이다.

@Configuration
public class KafkaConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
        ConsumerFactory<String, String> consumerFactory,
        KafkaTemplate<String, String> kafkaTemplate
    ) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);

        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
        factory.setCommonErrorHandler(new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3)));

        return factory;
    }
}

DLQ 토픽 생성 방식

Spring Kafka는 기본적으로 DLQ 토픽 이름에 -dlt 접미사를 붙여 자동 생성한다. 예를 들어, 원래의 Topic이 order.created라면, DLQ 토픽은 order.created-dlt로 생성된다.

Kafka UI 또는 CLI를 통해 다음과 같이 DLQ 토픽이 생성된 것을 확인할 수 있다.

img_5