백엔드 데이터 인프라 110편 — Kafka Message Format (Record Batch v2 · 바이트 구조)

이 글은 백엔드 데이터 인프라 시리즈 130편 중 110편이에요. 109편까지 디스크와 네트워크 영역을 잡았다면, 이번 110편은 한 단계 더 들어가서 바이트 단위인 Message Format을 다뤄요. Kafka가 wire(네트워크 전송)와 디스크에서 실제로 쓰는 정확한 binary 포맷이에요.

Message Format이 어렵게 느껴지는 이유

대부분 추상화 레이어 아래에 숨어 있어서 일상에서 직접 만질 일이 거의 없어요. 다만 디버깅이나 protocol 분석, custom client를 짤 때는 결국 이 레이어까지 내려와야 해요.

첫째, 여러 버전이 공존해요. magic byte 0, 1, 2가 모두 살아 있어서 호환성 영역을 신경 써야 해요.

둘째, Varint나 Delta encoding 같은 공간 절약 기법이 깔려 있어요. 압축이 잘 되면서도 효율적이라는 게 핵심이에요.

이 글에서는 v2 포맷의 모든 필드와 호환성, 그리고 디버깅 도구까지 한 번에 짚어볼게요.

Magic Byte — 3가지 버전

지금은 v2가 표준이에요. 옛 클라이언트가 v0나 v1로 publish하면 broker가 직접 변환해서 저장하는데, 이 변환 과정이 성능을 갉아먹어요.

Record Batch v2 구조

RecordBatch:
┌─────────────────────────────────────┐
│ baseOffset (8 bytes)                 │
│ batchLength (4 bytes)                │
│ partitionLeaderEpoch (4 bytes)       │
│ magic (1 byte) = 2                   │
│ crc (4 bytes)                        │
│ attributes (2 bytes) — compression·timestamp type 등
│ lastOffsetDelta (4 bytes)            │
│ baseTimestamp (8 bytes)              │
│ maxTimestamp (8 bytes)               │
│ producerId (8 bytes)                 │
│ producerEpoch (2 bytes)              │
│ baseSequence (4 bytes)               │
│ recordsCount (4 bytes)                │
│                                       │
│ ┌─ Record 1 (varint encoded) ────┐  │
│ │   length (varint)              │  │
│ │   attributes (1 byte)          │  │
│ │   timestampDelta (varint)      │  │
│ │   offsetDelta (varint)         │  │
│ │   keyLength (varint), key      │  │
│ │   valueLength (varint), value  │  │
│ │   headersCount (varint)        │  │
│ │   ┌─ Header 1 ──┐               │  │
│ │   │ key·value   │               │  │
│ │   └─────────────┘               │  │
│ └────────────────────────────────┘  │
│ ┌─ Record 2 ──────────────────────┐ │
│ ...                                  │
└─────────────────────────────────────┘

여러 record가 한 batch에 묶이는 게 핵심이에요. 그리고 batch 자체가 compression과 CRC의 단위가 돼요.

Batch Header 필드

Identification

• baseOffset = batch의 첫 record offset
• partitionLeaderEpoch = 89편에서 본 leader epoch
• producerId·producerEpoch·baseSequence = 86편 idempotent producer

Compression·Type Attributes

attributes (2 bytes)는 bit 별로 다른 의미를 가져요.

• bit 0~2 = compression type (none·gzip·snappy·lz4·zstd)
• bit 3 = timestamp type (CreateTime·LogAppendTime)
• bit 4 = transactional flag
• bit 5 = control batch (transaction marker)

Timestamps

• baseTimestamp = 첫 record timestamp
• maxTimestamp = batch의 max timestamp (timeindex 용)

각 record는 baseTimestamp부터의 delta만 저장해서 공간을 아껴요.

Counts

• lastOffsetDelta = batch의 last offset = baseOffset + lastOffsetDelta
• recordsCount = batch 안 record 수

Record 필드 — Varint 인코딩

각 record는 이렇게 생겼어요.

length:         varint  (총 길이)
attributes:     1 byte
timestampDelta: varint  (baseTimestamp 부터 delta)
offsetDelta:    varint  (baseOffset 부터 delta)
keyLength:      varint
key:            keyLength bytes
valueLength:    varint
value:          valueLength bytes
headersCount:   varint
headers:        ...

Varint란?

Variable-length integer(가변 길이 정수)예요. 작은 숫자에는 적은 byte를, 큰 숫자에는 많은 byte를 쓰는 인코딩이에요.

값 0~127     → 1 byte
값 128~16383  → 2 bytes
값 16384~...   → 3+ bytes
...

기존 고정 4 byte int에 비해 2~4배 공간을 줄여요. 대부분 length나 delta 값은 작은 숫자라서 효율이 잘 나와요.

여기서 시험 함정이 하나 있어요. Delta와 Varint를 조합하면 연속된 비슷한 timestamp나 offset이 아주 효율적으로 압축돼요. 예를 들면 이런 식이에요.

timestamps:
  1747475100000, 1747475100050, 1747475100100, 1747475100150
  ↓ delta from base 1747475100000
  0, 50, 100, 150
  ↓ varint (각 1~2 byte)

원본은 8 byte × 4 = 32 byte인데 varint로 가면 4~8 byte로 줄어서 4~8배 압축이 돼요.

Headers — Record Metadata

각 record는 key-value 형태의 메타데이터를 들고 있어요.

header 1: trace-id  → "abc123"
header 2: source    → "service-A"
header 3: schema-v  → "v2"

쓰임새는 다양해요.

• 분산 trace (OpenTelemetry)
• Schema 버전
• Routing hint
• 감사 로그 (user_id, request_id 등)

각 header는 key(varint length + bytes)와 value(varint length + bytes)로 구성돼요.

Control Batches — Transaction Marker

attributes의 control batch bit를 켜면 특수 batch가 돼요.

Transaction commit:  control batch with marker type COMMIT
Transaction abort:   control batch with marker type ABORT

88편 EOS(Exactly-Once Semantics)에서 본 transaction의 isolation.level=read_committed가 바로 이 marker를 보고 visible과 invisible을 결정해요.

CRC — 무결성 보장

crc (4 bytes)는 batch 전체(header 일부 + records)의 CRC-32C 체크섬이에요.

Producer 가 batch 만들 때 = CRC 계산
Broker 가 batch 받을 때 = CRC 검증
Disk 에 저장된 batch 읽을 때 = CRC 검증
Consumer 가 batch 받을 때 = CRC 검증

데이터 corruption이 발생하면 이 단계에서 바로 잡혀요.

Compression — Batch 단위

85편 efficiency에서 본 내용이에요. attributes의 compression type bit로 표시해요.

Producer:
  records → MessageSet 만듦
  → compression.type 으로 압축
  → Batch 의 records 영역이 *압축된 bytes*

Broker:
  압축된 batch 저장 (재압축 X)
  Consumer 에게 압축된 그대로 전송

Consumer:
  압축 해제 → record 들 추출

이게 바로 End-to-End 압축이에요. 한 번 압축하면 모든 단계를 그대로 통과해요.

Wire 와 Disk format 동일

핵심 사실 하나만 짚을게요. Network wire format과 Disk on-disk format이 똑같아요.

Producer → wire bytes → Broker → 같은 bytes 그대로 disk
                                       ↓
Consumer ← wire bytes ← Broker ← 같은 bytes 그대로 disk

이게 Zero-copy가 가능한 이유예요. 변환이 필요 없으니까 sendfile()로 kernel-only 전송이 돼요.

Forward / Backward Compatibility

Forward — 새 broker가 옛 client 처리

옛 클라이언트가 v1 형식으로 publish하면 broker가 v2로 변환한 뒤 저장해요. 또는 v1 그대로 저장해 두고 consumer가 요청할 때 변환하기도 해요.

Backward — 새 client 가 옛 broker 사용

새 client가 v2로 publish하는데 옛 broker가 v1만 지원한다면, broker가 거부하거나 변환을 요구해요. 그래서 버전 매칭 확인이 필수예요.

MessageFormatVersion

# Topic 레벨
message.format.version=3.0

특정 topic만 특정 format을 강제하는 옵션이에요. 호환성 마이그레이션 단계에서 자주 써요.

Java API — RecordBatch 직접 사용

대부분 추상화돼 있지만, 직접 만져야 할 때도 있어요.

import org.apache.kafka.common.record.*;

// 디스크 파일에서 직접 read
try (FileRecords records = FileRecords.open(new File("..."), false)) {
    Iterator<RecordBatch> batches = records.batches().iterator();
    while (batches.hasNext()) {
        RecordBatch batch = batches.next();
        System.out.println("baseOffset: " + batch.baseOffset());
        for (Record record : batch) {
            System.out.println("  offset: " + record.offset() + ", key: " + record.key());
        }
    }
}

kafka-dump-log.sh가 결국 이런 코드를 wrapping한 도구예요.

디버깅 — kafka-dump-log.sh --deep-iteration

$ kafka-dump-log.sh \
    --files /var/kafka-data/my-topic-0/00000000000000000000.log \
    --deep-iteration \
    --print-data-log

Dumping ...
baseOffset: 0 lastOffset: 49 count: 50 baseSequence: 0 lastSequence: 49 \
    producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false \
    isControl: false position: 0 CreateTime: 1747475100000 size: 1234 \
    magic: 2 compresscodec: ZSTD ...

| offset: 0 CreateTime: 1747475100000 keySize: 7 valueSize: 100 \
    sequence: 0 headerKeys: [trace-id, source] payload: ...
| offset: 1 CreateTime: 1747475100050 keySize: 7 valueSize: 100 \
    sequence: 1 headerKeys: [trace-id] payload: ...

batch metadata와 각 record, headers까지 모든 정보가 한 번에 떠요.

한계·실무 함정

1. 옛 client 의 변환 부담

옛 client(v0나 v1)가 대량으로 publish하면 broker가 변환 부담을 떠안아 CPU가 폭증해요. client 버전 upgrade가 사실상 필수예요.

2. Compression mismatch

Producer는 압축해서 보냈는데 Broker의 compression.type이 다르면 broker가 다시 압축해요. 그만큼 CPU에 부담이 가니 설정을 일관되게 맞추는 게 좋아요.

3. Large Headers

Headers가 너무 많으면 batch 크기가 커지고 메타데이터 비효율도 같이 따라와요. 핵심만 남기는 걸 권장해요.

4. CRC 실패

Disk corruption이 생기면 CRC mismatch로 잡혀서 broker가 segment를 손상 처리해요. 모니터링이 필요해요.

5. Producer ID overflow

producerId나 producerEpoch가 한계에 도달하는 경우인데, 실제로는 매우 드물어요. 발생한다면 Idempotent producer reset을 검토해요.

시험 직전 한 번 더 — Kafka Message Format 함정 압축 노트

• Magic byte v0 (0.7), v1 (0.8~0.9), v2 (0.10+, 현재)
• Record Batch v2 = 여러 record 를 batch 단위로 묶음
• Batch 자체가 compression·CRC 단위
• Batch Header = baseOffset·batchLength·partitionLeaderEpoch·magic·crc·attributes·lastOffsetDelta·baseTimestamp·maxTimestamp·producerId/Epoch/Sequence·recordsCount
• attributes bit 별 = compression type·timestamp type·transactional·control batch
• Record 필드 = length·attributes·timestampDelta·offsetDelta·keyLength/key·valueLength/value·headersCount/headers
• Varint encoding = 작은 숫자 = 적은 byte (2~4배 공간 절약)
• Delta + Varint = 연속 timestamp/offset 매우 효율적 (4~8배 압축)
• Headers = record metadata (trace-id·source·schema-v 등)
• Control Batch = transaction commit/abort marker
• isolation.level=read_committed 가 marker 보고 visible/invisible
• CRC-32C = batch 전체 무결성, 모든 단계 검증
• Compression Batch 단위 = end-to-end 한 번 압축
• Wire == Disk format = 동일 bytes → zero-copy 가능
• Forward = 새 broker 가 옛 client 처리 (변환)
• Backward = 새 client 가 옛 broker = 버전 매칭 필수
• message.format.version = topic 별 format 강제
• kafka-dump-log.sh --deep-iteration --print-data-log = 모든 정보 표시
• 함정 — 옛 client 변환 부담 (CPU 폭증)
• 함정 — Compression mismatch (재압축 부담)
• 함정 — Large Headers
• 함정 — CRC 실패 (disk corruption)
• 함정 — Producer ID overflow (매우 드묾)

공식 문서: Kafka Message Format 에서 자세한 binary 사양을 확인할 수 있어요.

시리즈 다른 편 (앞뒤 글 모음)

이전 글:

• 105편 — Kafka KRaft (Zookeeper 의 후속 · Quorum 운영)
• 106편 — Kafka Tiered Storage (S3 · 무한 Retention)
• 107편 — Kafka Log Compaction (Key 별 최신만 유지)
• 108편 — Kafka Log 파일 구조 (Segment · Index · Offset)
• 109편 — Kafka Network Layer (NIO · Selector · Thread Pool)

다음 글:

• 111편 — Kafka Consumer Rebalance Protocol (KIP-848 새 모델)
• 112편 — Kafka Transaction Protocol (EOS 내부 메커니즘)
• 113편 — Kafka Security Overview (3축 종합)
• 114편 — Kafka SSL/TLS (Keystore · mTLS · 인증서 운영)
• 115편 — Kafka SASL (SCRAM · PLAIN · OAuth · Kerberos)