broker

Broker

producer가 보낸 메시지를 디스크에 영속화하고 consumer에게 전송하는 모듈입니다.

개요

producer로부터 받은 메시지를 디스크에 영속화하고, consumer에게 전송하는 모듈입니다.

요구사항

broker를 사용하려면 다음 환경이 필요합니다.

항목 최소 버전
Java 17
Spring Boot 3.2.0

설치

broker는 JitPack으로 배포됩니다.
build.gradle에 의존성을 추가해 주세요.

build.gradle (Groovy DSL)

build.gradle 
repositories {
    mavenCentral()
    maven { url "https://jitpack.io" }
}

dependencies {
    implementation "com.github.moko-meringue.mmmq:broker:0.0.2"
}

build.gradle.kts (Kotlin DSL)

build.gradle.kts 
repositories {
    mavenCentral()
    maven("https://jitpack.io")
}

dependencies {
    implementation("com.github.moko-meringue.mmmq:broker:0.0.2")
}

Broker

Broker는 producer로부터 메시지를 받아 TopicQueue에 저장하는 클래스입니다.
Spring 자동 구성으로 빈이 등록되므로 사용자가 직접 정의할 필요는 없습니다.

POST /mmmq/messages 엔드포인트

Broker는 @PostMapping("/mmmq/messages")를 통해 메시지를 수신합니다.
producer는 이 경로로 메시지를 보내고, broker는 메시지를 TopicQueue에 저장한 뒤 ACK 혹은 NACK을 응답합니다.

응답 
{ "acknowledgement": "ACK" }

응답

응답 의미
ACK 메시지가 디스크에 영속화되었습니다.
NACK broker가 디스크 쓰기에 실패했습니다. 메시지는 저장되지 않았습니다.

TopicQueue

TopicQueue는 토픽 별 메시지를 저장하는 append-only 영속 큐입니다.
같은 토픽으로 들어온 메시지는 모두 해당 토픽의 TopicQueue에 도착 순서대로 저장되고, 하나 이상의 Dispatcher가 자신의 오프셋을 통해 메시지를 소비합니다.
중앙 집중식 구조 덕분에, 메시지 생산과 소비가 완전히 분리되고, 메시지 순서가 보장되며, Dispatcher는 독립적으로 메시지를 소비할 수 있습니다.

메시지 저장

TopicQueue는 메시지를 디스크에 저장합니다.

메시지가 디스크에 어떤 형태로 저장되는지에 대한 자세한 내용은 메시지 영속화 항목을 참고하세요.

오프셋 기반 메시지 조회

TopicQueue에 쌓이는 각 메시지는 오프셋이라는 절대 순번으로 식별됩니다.
토픽의 첫 메시지는 0, 다음은 1, 그 다음은 2 형태로 도착 순서에 따라 1씩 증가합니다.

오프셋 조회 모델 
TopicQueue - order.new
┌──────┬──────┬──────┬──────┬──────┬──────┬─────┐
│ msg  │ msg  │ msg  │ msg  │ msg  │ msg  │ ... │   ← 메시지
└──────┴──────┴──────┴──────┴──────┴──────┴─────┘
0      1      2      3      4      5                ← 오프셋

메시지 조회는 오프셋을 통해 이루어집니다.

오프셋을 사용해 TopicQueue에서 메시지를 조회하는 자세한 과정은 메시지 조회 항목을 참고하세요.

동적 TopicQueue 생성

등록되어 있지 않은 토픽의 메시지가 도착하면 broker는 새 TopicQueue를 생성합니다.

동적 생성 모델의 장점

  • 리플레이: consumer가 토픽보다 나중에 생성되어도 메시지가 디스크에 그대로 쌓여 있습니다. 오프셋 0부터 시작하면 그동안의 모든 메시지를 리플레이할 수 있습니다.
  • 운영 결합도 감소: 토픽을 미리 등록하는 단계가 없으므로, 새 토픽이 도입되어도 broker를 업데이트하거나 재시작할 필요가 없습니다.

동시성 모델

  • 쓰기: TopicQueue는 동시에 하나의 쓰기를 진행합니다 (ReentrantLock 기반).
    메시지는 도착한 순서대로 디스크에 저장되며, 같은 토픽 안에서 메시지 순서가 보존됩니다.
  • 읽기: 읽기는 잠금 없이 진행됩니다.
    쓰기 도중 읽기가 발생하더라도, 애플리케이션 레벨에서 저장이 완료된 메시지만 노출하므로, 아직 저장 중인 메시지는 안전하게 보호됩니다.

Dispatcher

Dispatcher는 TopicQueue에 저장된 메시지를 consumer에게 전송하는 클래스입니다.
하나의 Dispatcher는 자신의 ConsumerId로 식별되며, Host로 메시지를 전송할 consumer를 가리킵니다.

여러 Dispatcher를 동시에 정의할 수 있습니다.
같은 Host를 공유하는 Dispatcher들은 같은 consumer로 향하지만, 각자 다른 ConsumerId를 메시지에 실어 보내므로 서로 독립된 메시지 흐름이 됩니다.

정의

Dispatcher는 설정 파일에 선언합니다.
하나의 Dispatcher는 메시지를 전송할 consumer의 Host, 자신의 ConsumerId, 구독할 TopicPattern으로 정의되며, broker는 시작할 때 설정 파일을 읽어 각 항목을 Dispatcher 빈으로 등록합니다.

참고
Dispatcher는 ConsumerId 단위로 구분됩니다.
같은 Host와 같은 TopicPattern을 가졌더라도 ConsumerId가 다르면 서로 다른 Dispatcher이며, 각자 독립된 메시지 흐름을 가집니다.

Dispatcher 설정 파일

설정 파일의 경로는 영속화 루트 디렉터리 바로 아래의 dispatchers.json으로 고정되며, 기본값은 ./mmmq/dispatchers.json입니다.
루트 디렉터리는 mmmq.broker.persistence.root-dir 설정으로 변경할 수 있고, 설정 파일의 경로만 따로 지정할 수는 없습니다.
최상위는 배열이고, 항목 하나가 Dispatcher 하나가 됩니다.

dispatchers.json 
[
  {
    "consumerId": "order-created",
    "host": { "protocol": "HTTP", "address": "localhost", "port": 8081 },
    "pattern": "order.created"
  },
  {
    "consumerId": "order-shipped",
    "host": { "protocol": "HTTP", "address": "localhost", "port": 8081 },
    "pattern": "order.shipped"
  }
]

위 예시는 같은 consumer 호스트(localhost:8081)로 향하는 두 Dispatcher를 보여줍니다.
같은 Host를 공유하지만 ConsumerId(order-created, order-shipped)가 다르므로, consumer 안에서는 서로 다른 메시지로 구분됩니다.

설정 필드

필드 타입 설명
consumerId 문자열 이 Dispatcher의 식별자이자 consumer에 전달되는 라우팅 키. mmmq-consumer-id 헤더로 함께 전송됩니다.
host 객체 메시지를 전송할 consumer의 Host. protocol, address, port로 구성됩니다.
pattern 문자열 구독할 단일 TopicPattern

host의 protocol은 HTTP 또는 HTTPS이며 대소문자를 가리지 않습니다.

TopicQueue 구독 모델

Dispatcher는 자신의 TopicPattern과 매칭되는 모든 TopicQueue를 구독합니다.

Dispatcher는 구독한 TopicQueue 별로 두 가지 자원을 관리합니다.

  • 오프셋: TopicQueue에서 다음에 조회할 메시지의 오프셋입니다.
    메시지 전송이 끝날 때마다 커밋을 통해 다음 값(+1)으로 업데이트합니다.
  • 워커 스레드: TopicQueue를 전담하는 스레드입니다.
    워커는 TopicQueue의 조회 가능한 메시지를 모두 소비한 뒤 종료되고, 새 메시지가 도착하면 다시 실행됩니다.

broker 구동 시

broker가 시작되면 설정 파일을 읽어 각 항목을 Dispatcher 빈으로 등록합니다.
설정 파일이 없으면 빈 배열([]) 파일을 만들고, Dispatcher 없이 정상적으로 기동합니다.

이때 모든 Dispatcher의 ConsumerId가 서로 다른지 검사합니다.
중복된 ConsumerId, 알 수 없는 protocol, 잘못된 JSON처럼 정의가 올바르지 않으면 broker 시작이 거부됩니다.

검사가 끝나면 각 Dispatcher는 자신의 TopicPattern과 매칭되는 모든 TopicQueue를 구독합니다.
구독한 TopicQueue마다 오프셋과 워커 하나가 할당되고, 메시지 소비가 시작됩니다.

새 TopicQueue 생성 시

새 TopicQueue가 생성되면 모든 Dispatcher는 새 토픽과 자신의 TopicPattern을 비교해 구독 여부를 판단합니다.
자신의 TopicPattern과 새 TopicQueue의 토픽이 매칭된다면, 그 Dispatcher는 새 TopicQueue를 구독합니다.

메시지 도착 시

구독 중인 TopicQueue에 메시지가 도착하면 워커가 그 메시지를 조회해 consumer에게 전송합니다.
전송 시에는 자신의 ConsumerId를 mmmq-consumer-id 헤더에 실어 보냅니다.
ACK을 받으면 오프셋을 커밋하고 다음 메시지로 진행합니다.
더 이상 조회 가능한 메시지가 없으면 다음 메시지가 도착하기 전까지 대기합니다.

Dispatcher가 TopicQueue에서 메시지를 조회하는 과정에 대한 자세한 내용은 메시지 조회 항목을 참고하세요.

재시도와 지수적 백오프

consumer가 NACK을 반환하면, Dispatcher는 메시지를 재전송합니다.
메시지는 최대 세 번 재전송되며, 전부 실패하면 로그를 남기고 다음 메시지로 진행합니다.

consumer가 정상 응답(ACK 또는 NACK)을 반환하지 못하면 Dispatcher는 지수적 백오프 전략으로 메시지를 재전송합니다.
네트워크 단절이나 HTTP 4xx/5xx 응답이 해당합니다.
대기 시간은 1초에서 시작해 두 배씩 늘어나 최대 60초까지 길어지며, 정상 응답이 돌아올 때까지 재시도가 반복됩니다.

메시지 영속화

broker는 producer가 보낸 메시지를 토픽 별 디렉터리의 세그먼트 파일과 인덱스 파일에 나누어 영속화합니다.
세그먼트 파일과 인덱스 파일의 디스크 동기화(fsync)가 모두 완료된 후 producer에게 ACK을 반환하므로, ACK을 받은 메시지는 디스크에 영속화된 상태로 남습니다.
OS, 하드웨어, JVM 프로세스가 갑작스럽게 중단되더라도, ACK을 받은 메시지의 영속성은 보장됩니다.

디렉터리 레이아웃

루트 디렉터리 아래에 Dispatcher 설정 파일(dispatchers.json)과 topics/ 디렉터리가 있습니다.
topics/ 아래에 토픽 별 하위 디렉터리가 만들어지고, 각 토픽 디렉터리 안에 필요한 파일이 함께 들어 있습니다.

파일 트리 
./mmmq/                             # 기본 루트 디렉터리 (설정으로 변경 가능)
├── dispatchers.json                # Dispatcher 설정 파일 (고정 경로)
└── topics/
    ├── order.created/
    │   ├── 0000000000000000000.mmm
    │   ├── 0000000000000000000.idx
    │   ├── 0000000000000123456.mmm
    │   ├── 0000000000000123456.idx
    │   └── checkpoints/            # Dispatcher별 체크포인트 디렉터리
    │       ├── order-created.checkpoint
    │       └── order-audit.checkpoint
    └── notification.sent/
        ├── 0000000000000000000.mmm
        ├── 0000000000000000000.idx
        └── checkpoints/
            └── notification-mailer.checkpoint

각 파일의 역할은 다음과 같습니다.

파일 역할
.mmm 세그먼트 파일
.idx 세그먼트 별 오프셋 인덱스 파일
.checkpoint Dispatcher 별 체크포인트 파일
dispatchers.json Dispatcher 설정 파일

루트 디렉터리는 다음 설정으로 변경할 수 있습니다.

application.yml 
mmmq:
  broker:
    persistence:
      root-dir: ./mmmq

세그먼트

토픽 큐의 메시지는 시간이 흐르면서 무한히 쌓일 수 있습니다.
이를 한 파일에 모두 담으면 단일 파일이 한없이 커지므로, broker는 일정 크기마다 세그먼트를 만들어 분할 저장합니다.

세그먼트는 토픽 큐를 일정 크기 단위로 나누어 저장하는 append-only 디스크 파일입니다.
세그먼트는 메시지를 저장하는 세그먼트 파일과 오프셋 인덱스로 구성됩니다.
각 세그먼트의 파일명은 그 세그먼트가 시작하는 오프셋(19자리)으로 이루어집니다 (예: 0000000000000001234.mmm).

각 메시지는 세그먼트 파일 안에 entry로 직렬화되어 저장됩니다.

Entry 구조 
┌──────────┬─────────┬─────────────────────────┐
│  length  │   CRC   │  message bytes (JSON)   │
│   4 B    │   4 B   │       length - 4 B      │
└──────────┴─────────┴─────────────────────────┘
  • length: CRC와 메시지 바이트의 합.
  • CRC: 메시지 바이트 전체에 대한 CRC32C 체크섬.
  • message bytes: Jackson으로 직렬화한 Message의 JSON 바이트.

세그먼트 끝에 메시지 entry가 append된 직후 fsync가 호출됩니다.

세그먼트 체인

한 토픽 큐가 가진 모든 세그먼트는 세그먼트 체인으로 구성됩니다.
세그먼트 체인은 토픽 큐의 세그먼트들을 시작 오프셋 순으로 정렬해 둔 자료구조로, 토픽 큐와 세그먼트 사이의 의사소통을 담당합니다.
하나의 세그먼트의 용량이 가득 차면 회전이 발생해 새 세그먼트가 만들어지고, 그 세그먼트가 세그먼트 체인 끝에 추가됩니다.
메시지는 항상 세그먼트 체인의 가장 마지막 세그먼트에 추가됩니다.

세그먼트 회전

세그먼트 체인은 세그먼트 용량이 일정 수준을 넘으면 새 세그먼트를 생성하는 회전을 수행합니다.
이후에 도착하는 메시지는 새 세그먼트에 저장되고, 이전 세그먼트에는 더 이상 메시지가 저장되지 않습니다.
새 세그먼트의 시작 오프셋은 직전 세그먼트의 시작 오프셋 + 직전 세그먼트가 저장한 메시지 개수를 더한 값으로 설정됩니다.

회전 검사는 메시지를 추가하기 직전에 이루어집니다.
검사 시점의 세그먼트 용량이 임계값 미만이면 메시지는 현재 세그먼트에 추가되고, 추가된 메시지의 크기만큼 파일 크기가 늘어나면서 세그먼트 용량 임계값을 초과할 수 있습니다.

세그먼트 회전 
        [ Segment 1 ]                            [ Segment 2 ]
   (0000000000000001200.mmm)               (0000000000000001601.mmm)
┌─────────────────────────────┐         ┌─────────────────────────────┐
│     message 1200 ~ 1600     |  ────▶  |        message 1601 ~       |
└─────────────────────────────┘         └─────────────────────────────┘
      Size: 64MB (Closed)                     Size: 12KB (Active)

새 세그먼트 시작 오프셋 = 직전 세그먼트 시작 오프셋(1200) + 직전 세그먼트 저장 메시지 개수(401) = 1601

회전 임계값은 기본 64 MB로 설정되어 있으며, 설정을 통해 조정할 수 있습니다.

application.yml 
mmmq:
  broker:
    persistence:
      segment:
        max-bytes: 67108864 # 64 MB

오프셋 인덱스

세그먼트에는 가변 길이의 메시지가 순서대로 저장됩니다.
특정 오프셋의 메시지를 찾기 위해선 세그먼트 파일의 첫 부분부터 순차적으로 탐색해야 합니다.
이 과정에서 수많은 시스템 콜과 디스크 탐색이 발생하므로, 세그먼트에 메시지가 쌓일수록 읽기 비용이 크게 증가합니다.

오프셋 인덱스는 이 비용을 최소화하기 위해 메시지 별 시작 byte 주소를 저장해두는 별도 파일입니다.
오프셋 인덱스는 각 세그먼트별로 하나씩 존재하며, 세그먼트 파일과 같은 시작 오프셋을 공유하는 .idx 파일로 구성됩니다.

오프셋 인덱스는 오프셋에 해당하는 메시지의 세그먼트 파일 내 byte 주소를 저장합니다.
모든 값은 8 byte 고정 길이이므로, 세그먼트의 N번째 메시지의 byte 주소는 N × 8 byte 위치에서 읽을 수 있습니다.

.idx 파일 구조 
       ┌──────────┬──────────┬──────────┬──────────┐
       │   3425   │   3837   │   4002   │   4233   │
       └──────────┴──────────┴──────────┴──────────┘
byte    0          8         16         24

오프셋 인덱스는 세그먼트에 저장된 메시지의 Single Source Of Truth입니다.
특정 오프셋의 메시지가 세그먼트에 존재하는지는 세그먼트가 아니라 오프셋 인덱스를 기준으로 판단합니다.
세그먼트에 메시지가 저장되어 있어도 오프셋 인덱스에 해당 메시지의 byte 주소가 등록되어 있지 않으면, 메시지는 절대 노출되지 않습니다.

메시지 entry가 세그먼트에 저장된 후, broker는 해당 entry의 byte 주소를 .idx에 append하고 fsync를 호출합니다.
fsync가 끝난 뒤에야 메시지가 Dispatcher에게 노출되며, producer에게 ACK를 반환할 수 있습니다.

오프셋 인덱스로 메시지 찾기

세그먼트 체인은 입력 오프셋에 해당하는 메시지를 찾기 위해 다음 단계를 수행합니다.

  1. 세그먼트 선택: 시작 오프셋이 입력 오프셋 이하인 세그먼트 중 시작 오프셋이 가장 큰 세그먼트를 선택합니다.
  2. 상대 오프셋 계산: 입력 오프셋 - 세그먼트의 시작 오프셋으로 세그먼트 내에서의 상대 오프셋을 구합니다.
  3. byte 주소 계산: 오프셋 인덱스에서, 상대 오프셋 × 8 byte 주소에 있는 long 값을 읽어 세그먼트 파일 내에서 메시지가 시작하는 byte 주소를 얻습니다.
  4. entry 읽어 역직렬화: 세그먼트 파일에서, byte 주소에 위치한 entry를 읽어 메시지를 역직렬화합니다. 세그먼트 entry 구조는 세그먼트 항목을 참고하세요.

체크포인트와 커밋

체크포인트는 Dispatcher가 TopicQueue에서 다음에 조회할 오프셋을 저장한 파일입니다.
각 토픽 디렉터리의 checkpoints/ 하위에 <consumerId>.checkpoint 형태로 저장됩니다.

Dispatcher가 한 메시지의 처리를 마친 뒤 체크포인트의 오프셋을 다음 값으로 옮기는 동작을 커밋이라 부릅니다.
커밋이 디스크에 반영되는 시점과 장애 시 재전송 동작에 대한 자세한 내용은 체크포인트 커밋 항목을 참고하세요.

파일 위치 
./mmmq/topics/
└── order.created/
    ├── ...
    └── checkpoints/
        ├── order-fulfillment.checkpoint
        ├── billing.checkpoint
        └── audit.checkpoint

체크포인트 파일은 단일 8 byte long 값을 담습니다.
이 값은 해당 Dispatcher가 다음에 조회할 메시지의 오프셋입니다.
체크포인트는 8 byte의 오프셋 값을 매번 덮어쓰는 방식입니다.

토픽의 메시지 스트림 위에서 체크포인트가 가리키는 위치는 다음과 같습니다.

체크포인트 값이 3이면 오프셋 0, 1, 2는 Dispatcher가 이미 처리한 메시지이고, 다음에 조회할 메시지는 오프셋 3입니다.

체크포인트는 Dispatcher 단위로 분리되어 있으므로, 같은 토픽을 여러 Dispatcher가 구독하더라도 오프셋은 독립적으로 관리됩니다.

여러 Dispatcher의 독립 진도 
order.created 토픽 큐

    ┌─────┬─────┬─────┬─────┬─────┬─────┬─────┐
    │  0  │  1  │  2  │  3  │  4  │  5  │ ... │
    └─────┴─────┴─────┴─────┴─────┴─────┴─────┘
       ▲                 ▲           ▲
       │                 │           │
       │                 │           └─── billing.checkpoint = 5
       │                 └─── analytics.checkpoint = 3
       └─── audit.checkpoint = 0

같은 메시지가 세 개의 Dispatcher에 의해 서로 다른 시점에 처리될 수 있고, 한 Dispatcher가 뒤처져도 다른 Dispatcher의 진행에는 영향이 없습니다.

체크포인트 커밋

Dispatcher는 메시지를 consumer에게 전송하고 ACK를 받은 후 자신의 체크포인트를 다음 오프셋으로 갱신합니다.
.checkpoint 파일 0번 위치의 기존 값을 다음 오프셋으로 덮어쓴 뒤 fsync를 호출합니다.
fsync가 끝나야 해당 Dispatcher가 어디까지 처리했는지가 디스크에 남으며, 장애 후 재시작해도 완료된 오프셋 앞의 메시지를 다시 조회하지 않습니다.

주의
consumer가 ACK을 반환했더라도, 커밋이 완료되기 전에 장애가 발생하면 그 메시지는 복구 후 재전송됩니다.
consumer 핸들러는 같은 메시지가 두 번 도착하더라도 결과가 동일하도록 멱등하게 구현되어야 합니다.

NACK로 인한 재시도가 전부 실패해 다음 오프셋으로 넘어가는 과정에서도 커밋이 호출되므로 체크포인트는 다음 오프셋으로 이동합니다.

메시지 처리 파이프라인

메시지 처리의 전체 흐름은 메시지 생산, 메시지 저장, 메시지 조회, 메시지 소비 순서로 이어집니다.

메시지 생산

producer는 topic과 content를 담은 메시지를 만들어 /mmmq/messages 엔드포인트로 POST 요청을 전송합니다.
broker는 메시지의 토픽 이름을 기준으로 어떤 TopicQueue에 저장할지 결정합니다.

해당 토픽의 TopicQueue가 이미 있으면 기존 큐에 메시지를 저장합니다.
TopicQueue가 존재하지 않으면 새 TopicQueue가 생성되고, 그 큐에 메시지가 저장됩니다.

토픽 데이터는 루트 디렉터리의 topics/ 아래에 저장됩니다.
루트 디렉터리의 기본값은 ./mmmq이며, mmmq.broker.persistence.root-dir 설정으로 변경할 수 있습니다.
만약 order.created 토픽의 첫 메시지가 도착하면 broker는 ./mmmq/topics/order.created/ 디렉터리를 생성합니다.

새 토픽 큐의 파일 구조 
./mmmq/topics/
└── order.created/
    ├── 0000000000000000000.mmm
    ├── 0000000000000000000.idx
    └── checkpoints/

생산 단계는 메시지를 저장할 토픽 큐와 디스크 위치를 준비하는 단계입니다.

메시지 저장

TopicQueue는 같은 토픽으로 들어온 메시지를 도착 순서대로 저장합니다.
먼저 도착한 메시지는 더 작은 오프셋을 가지며, 같은 토픽 안에서는 이 오프셋 순서가 유지됩니다.

broker는 활성 세그먼트 파일 끝에 entry를 append하고, fsync를 호출합니다.
메시지가 세그먼트에 저장되었지만 아직 오프셋 인덱스에 작성되지 않았으므로, Dispatcher는 해당 메시지를 조회할 수 없습니다.

세그먼트 파일에 메시지를 저장한 후, entry의 시작 byte 주소를 오프셋 인덱스에 기록하고 fsync를 호출합니다.
세그먼트 안의 상대 오프셋이 6인 메시지가 byte 412에서 시작한다면, 오프셋 인덱스의 48 byte 위치(6 × 8)에 412가 저장됩니다.
오프셋 인덱스에 fsync까지 끝난 뒤에야 메시지는 Dispatcher가 조회할 수 있는 상태가 됩니다.

활성 세그먼트가 mmmq.broker.persistence.segment.max-bytes 기준을 넘으면 다음 메시지는 새로운 세그먼트에 저장됩니다.
새로운 세그먼트의 파일명은 직전 세그먼트의 시작 오프셋과 메시지 개수를 더한 값으로 만들어집니다.
직전 세그먼트의 시작 오프셋이 1200이고 메시지 개수가 401이라면, 새 세그먼트 파일은 0000000000000001601.mmm이 됩니다.

메시지 조회

토픽 큐에 메시지가 들어오면, 해당 토픽을 구독 중인 Dispatcher가 메시지 소비를 시작합니다.
Dispatcher는 자신의 체크포인트에 저장된 오프셋부터 메시지를 조회하며, 같은 토픽을 여러 Dispatcher가 구독하더라도 각자의 체크포인트를 기준으로 독립적으로 조회합니다.

broker는 저장 파일을 기준으로 절대 오프셋에 해당하는 메시지의 디스크 위치를 찾습니다.
다음과 같이 두 세그먼트로 나뉜 토픽에서 절대 오프셋 1030의 메시지를 조회하는 과정을 예로 듭니다.

  • 세그먼트 A: 파일명 0000000000000000000.mmm, 시작 오프셋 0, 메시지 1,024개 저장.
  • 세그먼트 B: 파일명 0000000000000001024.mmm, 시작 오프셋 1024, 그 이후 메시지 저장.
  1. 세그먼트 선택: 오프셋 1030 이하인 세그먼트 중 시작 오프셋이 가장 큰 세그먼트를 선택합니다.
    시작 오프셋이 1024인 세그먼트 B를 선택합니다.
  2. 상대 오프셋 계산: 오프셋에서 세그먼트 시작 오프셋을 빼서 세그먼트 안의 위치를 계산합니다.
    1030 - 1024 = 6이므로, 세그먼트 B 안에서는 6번째 entry입니다.
  3. 인덱스 조회: 세그먼트 B의 .idx 파일에서 상대 오프셋 6에 해당하는 슬롯을 읽습니다.
    인덱스 슬롯은 8 byte이므로 위치는 6 × 8 = 48 byte이고, 이 위치에서 세그먼트 안의 entry 시작 byte 주소(412)를 얻습니다.
  4. entry 읽기: 세그먼트 B의 412번째 byte 주소에서 4 byte length header를 먼저 읽습니다.
    다음으로, length byte만큼 body를 읽고, body 안에서 4 byte CRC32C와 나머지 메시지 JSON 바이트를 분리합니다.
  5. 검증과 역직렬화: 메시지 JSON 바이트로 CRC32C를 다시 계산해 entry에 저장된 값과 비교합니다.
    체크섬이 일치하면 JSON 바이트를 Jackson으로 Message 객체로 역직렬화하여 반환하고, 일치하지 않으면 로그를 남기고 다음 메시지로 넘어갑니다.

메시지 소비

메시지 조회가 끝나면 Dispatcher는 조회한 메시지를 consumer에게 전송합니다.
소비 단계의 핵심은 consumer 응답에 따라 체크포인트를 다음 오프셋으로 옮길지, 같은 오프셋에 머물지 결정하는 것입니다.

  1. 메시지 전송: Dispatcher는 조회한 메시지를 설정된 consumer의 /mmmq/messages 엔드포인트로 전송합니다.
    이때 자신의 ConsumerId를 mmmq-consumer-id 헤더에 실어 보냅니다.
  2. ACK 처리: consumer가 ACK을 반환하면 해당 메시지는 처리 완료로 간주됩니다.
    Dispatcher는 체크포인트의 값을 다음 오프셋으로 덮어쓴 뒤 fsync를 호출하고, 다음 메시지를 소비합니다.
  3. NACK 처리: consumer가 NACK을 반환하면 메시지를 다시 전송합니다.
    정해진 재시도 횟수(기본 3회)를 모두 사용해도 ACK을 받지 못하면 로그를 남기고 다음 오프셋으로 진행합니다.
  4. 통신 실패: consumer가 응답을 반환하지 못하거나 네트워크 오류가 발생하면 체크포인트는 이동하지 않습니다.
    Dispatcher는 지수 백오프로 대기한 뒤 메시지를 다시 전송합니다. 대기 시간은 1초부터 시작해 실패가 반복될 때마다 두 배씩 늘어나며, 최대 60초까지 증가합니다. 정상 응답을 받기 전까지 이 과정을 반복합니다.
  5. 손상 entry: 조회 과정에서 체크섬이 일치하지 않는 entry를 만나면 로그를 남기고 해당 오프셋을 건너뜁니다.
    이 경우에도 체크포인트는 다음 오프셋으로 이동하므로, 단일 entry 손상으로 인해 소비가 멈추지 않습니다.

시작 시 복구

복구는 broker가 다시 시작될 때 디스크에 남아 있는 세그먼트, 오프셋 인덱스, 체크포인트를 기준으로 토픽 큐와 Dispatcher의 진행 위치를 일관된 상태로 되돌리는 과정입니다.

먼저 데이터 루트 디렉터리를 확인합니다.
루트 디렉터리가 없으면 이전에 저장된 토픽이 없다고 판단하고 그대로 시작하며, 이후 첫 메시지가 들어올 때 필요한 토픽 디렉터리를 생성합니다.
루트 디렉터리가 있으면 그 아래의 각 하위 디렉터리를 토픽 디렉터리로 판단하고 TopicQueue를 등록합니다.
토픽 디렉터리에 세그먼트가 하나도 없으면 시작 오프셋이 0인 빈 세그먼트와 오프셋 인덱스를 준비합니다.

세그먼트는 오프셋 인덱스를 기준으로 다음과 같이 복구됩니다.

  • 오프셋 인덱스가 비어 있고 세그먼트도 비어 있음: 아직 저장된 메시지가 없는 빈 세그먼트로 판단하고 그대로 사용합니다.
  • 오프셋 인덱스가 비어 있는데 세그먼트에 byte가 남아 있음: 오프셋 인덱스에 등록되지 않은 데이터이므로 세그먼트를 0 byte로 truncate하고 fsync합니다.
  • 마지막 오프셋 인덱스 entry와 세그먼트 길이가 일치함: 마지막 entry까지 정상 저장된 상태로 판단하고 그대로 사용합니다.
  • 세그먼트가 마지막 entry 끝 위치보다 김: 오프셋 인덱스에 등록되지 않은 뒤쪽 byte 영역이 남은 상태입니다.
    broker는 이 영역을 Dispatcher에게 노출된 적 없는 데이터로 판단하고 마지막 entry 끝 위치까지만 남긴 뒤 fsync합니다.
  • 세그먼트가 마지막 entry 끝 위치보다 짧음: 오프셋 인덱스는 존재하지만 실제 메시지 byte가 부족한 상태입니다.
    이 경우 오프셋 인덱스를 신뢰할 수 없으므로 손상으로 판단하고 시작을 중단합니다.
  • 마지막 entry를 읽을 수 없거나 체크섬이 맞지 않음: 세그먼트가 손상된 상태로 판단하고 시작을 중단합니다.