Kafka 컴팩션은 어떻게 키별 최신 값 하나만 남기는가 — log cleaner 내부 동작

__consumer_offsets 토픽은 컨슈머 그룹이 커밋을 반복할수록 같은 키(그룹·토픽·파티션)에 새 오프셋을 계속 덮어쓴다. 그런데도 이 토픽이 디스크를 무한히 잡아먹지 않는 이유가 궁금했다. "오래된 메시지를 지우니까"라고 막연히 생각했지만, 그렇다면 방금 커밋한 최신 오프셋도 시간이 지나면 지워질 텐데 그건 말이 안 된다. 답은 시간 기반 삭제(retention)가 아니라 로그 컴팩션(log compaction) 이었다.

1. 컴팩션은 retention과 직교한다

가장 먼저 바로잡아야 했던 오해는 "컴팩션 = 오래된 메시지 삭제"였다. Kafka의 cleanup.policy는 두 개의 서로 독립적인 정책을 가진다.

정책	무엇을 기준으로 지우나	지우는 단위
delete (기본)	시간(retention.ms)·크기(retention.bytes)	세그먼트 파일 통째
compact	"같은 키의 더 새로운 레코드가 있는가"	레코드 하나씩

두 축이 직교하므로 cleanup.policy=compact,delete로 둘 다 켤 수도 있다. 컴팩션의 판정 기준은 "얼마나 오래됐나"가 아니라 "이 키의 최신본이 뒤에 또 있나"다. 그래서 컴팩션 토픽은 키의 개수가 유한하면 아무리 오래 써도 로그 크기가 대략 "키 개수 × 최신 값 크기"로 수렴한다.

로그 컴팩션은 파티션 로그를 key → 마지막 offset 맵으로 스캔한 뒤 다시 쓰면서, 각 키의 "가장 최근 레코드 하나"만 남기는 백그라운드 재작성 과정이다.

2. clean / dirty 구간과 cleaner point

컴팩션 대상 로그는 개념적으로 두 구간으로 나뉜다. 이미 유일화가 끝난 clean 구간과, 중복 키가 아직 남아 있는 dirty 구간이다.

     cleaned (이미 컴팩션됨)        dirty (아직 미처리)      active seg
 |------------------------------|------------------------|===========|
 0                          cleaner point            log end offset
 └ 키가 이미 유일화된 구간 ┘   └ 중복 키가 남아있는 구간 ┘  └ 쓰기 중 ┘

• cleaner point는 cleaner-offset-checkpoint 파일에 파티션별로 저장되는 "여기까지 완료" offset이다. 다음 라운드는 이 지점부터 dirty 끝까지만 훑는다.
• 지금 append 중인 active segment는 절대 컴팩션하지 않는다. 그래서 방금 쓴 같은 키의 두 레코드가 잠깐 공존할 수 있다. 컴팩션은 "언젠가 유일해진다"는 최종 보장이지 즉시성 보장이 아니라는 점을 여기서 이해했다.

log cleaner 스레드(log.cleaner.threads, 기본 1)는 매 라운드마다 가장 더러운 로그를 고르는데, 그 척도가 dirty ratio다.

dirty ratio = dirtyBytes / (dirtyBytes + cleanBytes)

min.cleanable.dirty.ratio(기본 0.5)를 넘는 로그만 후보가 되고, 그중 ratio가 가장 높은 것을 먼저 청소한다. 이 값은 재작성 IO 비용과 중복 잔존량 사이의 트레이드오프 노브다 — 낮추면 자주 돌아 중복이 적지만 IO가 늘고, 높이면 IO는 아끼지만 오래된 중복이 오래 남는다.

3. 2단계 알고리즘: OffsetMap 구축 → 재복사

Kafka 소스의 Cleaner.clean은 dirty 구간을 두 번 훑는다.

1단계 — build offset map. dirty 구간을 앞에서 뒤로 스캔하며 SkimpyOffsetMap에 key → 그 키가 등장한 마지막(가장 큰) offset을 채운다. 이름이 "skimpy(빈약한)"인 이유가 재미있다. 원본 키를 저장하지 않고 키의 MD5 해시 16B + offset 8B = 24B 슬롯만 오픈 어드레싱 해시테이블에 넣기 때문이다. 맵 용량은 log.cleaner.dedupe.buffer.size(스레드 수로 나눔) / 24로 정해진다. 즉 이 버퍼가 한 라운드에 유일화할 수 있는 키 개수의 상한이며, dirty 구간이 너무 크면 한 번에 다 못 담아 여러 세그먼트 그룹으로 쪼개 처리한다.

2단계 — recopy & swap. clean 구간 시작부터 dirty 끝까지 세그먼트를 순서대로 읽으며 각 레코드의 유지/폐기를 판정해 새 세그먼트로 복사한다. 판정 논리를 소스대로 개념 재현하면 이렇다.

// 1단계: dirty 구간 스캔 → 키의 최신 offset 기록
for (record <- dirtySegment.records)
  offsetMap.put(record.key, record.offset)   // 같은 키면 큰 offset으로 덮임

// 2단계: 전체 재복사하며 유지 판정
def shouldRetain(r: Record): Boolean = {
  val latest = offsetMap.get(r.key)           // 맵에 없으면 -1
  if (r.offset < latest) false                // 뒤에 같은 키의 최신본 존재 → 폐기
  else if (r.value == null)                   // tombstone
    now - r.timestamp <= deleteRetentionMs    // 유예 안 지났으면 유지
  else true
}

작은 세그먼트들은 컴팩션 후 더 작아지므로, segment.bytes를 넘지 않는 선에서 여러 원본 세그먼트를 하나의 새 세그먼트로 병합(grouping) 하기도 한다. 세그먼트 수 폭발과 파일 핸들 낭비를 막기 위해서다. 다 쓰면 원본을 새 세그먼트로 원자적 교체하고 cleaner point를 dirty 끝으로 전진시킨다.

4. offset은 불변, 연속성은 깨진다

컴팩션의 결정적 성질 하나: 살아남은 레코드의 offset은 원래 값 그대로다. 재부여하지 않는다. 그래서 컴팩션된 로그를 순차 소비하면 offset이 5, 6, 9, 13...처럼 구멍(gap) 을 가진다.

이건 버그가 아니라 설계다. offset을 다시 촘촘하게 매기면 이미 그 offset을 커밋해 둔 컨슈머와 팔로워 replication이 전부 어긋난다. 컨슈머는 gap을 정상으로 다뤄야 하고(다음 fetch offset은 언제나 "마지막으로 읽은 offset + 1"), 이 offset 안정성 덕분에 컴팩션이 도는 중에도 진행 중인 컨슈머의 위치가 흔들리지 않는다.

5. tombstone: 삭제를 "전파"한 뒤 지우기

키를 삭제하려면 value=null 레코드, 즉 tombstone을 보낸다. 컴팩션은 tombstone을 만나면 그 키의 이전 값들을 전부 지운다. 그런데 tombstone 자체를 즉시 없애면 문제가 생긴다 — 그 사이 오프라인이던 컨슈머는 "이 키가 삭제됐다"는 사실을 영영 못 보고, 자기 로컬 상태에 옛 값을 그대로 들고 있게 된다.

그래서 tombstone은 clean 구간에 편입된 시점부터 delete.retention.ms(기본 24h) 동안 유지된 뒤에야 다음 라운드에서 제거된다. 이 유예가 "모든 컨슈머가 삭제를 관측할 시간"을 보장한다. 이 값을 너무 짧게 잡으면 느린 컨슈머가 삭제 이벤트를 놓친다는 점이 실무 함정이다.

한 키의 생애:
  put(k,v1) → put(k,v2) → ... → put(k,vN) → delete(k=null) → (전파 대기) → 소멸
  컴팩션 후:         vN만 남음        →      tombstone만 남음  →  delete.retention.ms 후 소멸

6. 정리

• 컴팩션은 시간이 아니라 "같은 키의 더 새로운 레코드가 있는가"로 지운다. retention(delete)과 직교한다.
• 내부는 SkimpyOffsetMap으로 dirty 구간의 key→최신 offset을 만든 뒤, 전체를 재복사하며 최신본과 (유예 중인) tombstone만 남기는 2단계다.
• offset은 절대 재부여하지 않아 gap이 생기고, tombstone은 delete.retention.ms만큼 삭제를 전파한 뒤 사라진다.

더 파고들 만한 것: __consumer_offsets와 KRaft 메타데이터 로그가 컴팩션을 "상태 스냅샷"으로 활용하는 방식, 그리고 min.compaction.lag.ms / max.compaction.lag.ms가 dirty ratio 트리거와 어떻게 상호작용하는지.

참고 자료

• Apache Kafka Documentation — Log Compaction
• Kafka 소스: core/src/main/scala/kafka/log/LogCleaner.scala, LogCleanerManager.scala, SkimpyOffsetMap
• KIP-58: Make Log Compaction point configurable