Kafka - 04. consumer

consumer — 읽었다고 끝이 아니다, commit이 끝이다

consumer가 poll()을 호출해 batch를 받아왔다. 처리를 시작했다. 그런데 처리 도중 프로세스가 크래시 났다 — auto-commit이 켜져 있었다면, 방금 받은 batch는 이미 "읽었다"고 표시돼 있어서, 재시작해도 그 batch는 다시 오지 않는다. 메시지가 조용히 사라진다. Kafka consumer 설계에서 가장 흔하게 만나는 유실 지점이다.

이 함정의 뿌리는 "읽기 = 소비 완료"라는 착각이다. Kafka에선 읽기(poll)와 소비 완료(commit)가 분리돼 있고, 그 사이에 애플리케이션 처리가 끼어든다. auto-commit은 이 둘을 "읽자마자 commit"으로 합쳐버려 처리 실패 시 유실을 만든다. 신뢰성이 필요하면 수동 commit으로 at-least-once를 세우고, 중복을 흡수하도록 처리 로직을 멱등하게 짜야 한다. poll 루프, offset, commit, LAG가 어떻게 엮여서 consumer의 건강과 신뢰성을 결정하는지를 추적한다.

consumer는 앱에 내장되는 클라이언트다

consumer도 producer처럼 앱에 내장되는 클라이언트 라이브러리다.

Java consumer 기본 코드

// 1. 설정
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "order-processors");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false");  // 수동 commit (신뢰성)

// 2. consumer 생성 + 구독
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("orders"));

// 3. poll loop
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> record : records) {
        processOrder(record.value());  // 비즈니스 처리
    }
    consumer.commitSync();  // 처리 후 commit (at-least-once)
}

주문 처리 서비스가 poll loop로 돌아가는 법

주문 처리 서비스: orders topic을 구독 → poll()로 새 주문 batch 당겨옴 → 각 주문 처리(재고·결제·배송) → offset commit.

flowchart LR
    T[orders topic] -->|poll| C[consumer]
    C -->|처리| B[(재고·결제)]
    C -->|commit| OS["__consumer_offsets"]

consumer는 pull이다 — broker가 밀어넣지 않는다

"producer가 쓰면 consumer에게 push된다"가 아니다. consumer가 능동적으로 poll()로 당겨온다. broker는 요청 없이 안 보낸다.

pull이 만드는 세 가지 이점

producer가 consumer에게 push하는 대신 consumer가 당기게 한 설계는 단순한 취향이 아니라 세 가지 실제 이득을 만든다:

• backpressure가 자연스럽다: consumer가 자기 속도로 당기므로, 처리가 느린 consumer는 느린 만큼만 가져온다. broker가 밀어넣으면 consumer가 압도당해 크래시 나거나 메모리가 찰 수 있다.
• broker가 consumer 상태를 추적할 부담이 줄어든다: push 모델에선 broker가 "어느 consumer가 어디까지 받았나"를 기억해야 한다. pull 모델에선 consumer가 자기 offset을 책임지므로 broker는 상태 비용을 덜는다.
• replay가 자연스럽다: consumer가 offset을 과거로 돌려 다시 당기면 replay가 된다. push 모델에선 이미 보낸 메시지를 다시 보내는 메커니즘을 따로 만들어야 한다.

이 세 가지가 합쳐서 Kafka를 "메시지를 소비하면 사라지는 큐"가 아니라 "consumer가 자기 속도로 읽는 로그"로 만든다. 같은 topic을 여러 consumer group이 각자 다른 속도로, 다른 offset에서 읽을 수 있는 것도 pull 모델 덕분이다.

offset — 읽기 커서, 영속 저장

consumer는 "어디까지 읽었나"를 offset으로 기억:

• __consumer_offsets 내부 topic에 영구 저장(메모리 아님).
• compacted → 같은 key의 최신 offset만 유지.
• consumer 장애 후 재시작해도 이어서 읽음.

__consumer_offsets의 구조 — 왜 compact인가

__consumer_offsets는 key가 (group_id, topic, partition) 튜플이고 value가 offset+metadata인 로그다. 같은 consumer group이 같은 partition을 계속 읽을수록, 같은 key에 대한 갱신이 계속 쌓인다. 이 topic은 cleanup.policy=compact로 돼 있어, 같은 (group, topic, partition) key의 옛 offset은 버려지고 최신값만 남는다(장 07). 그래서 아무리 오래 운영해도 partition 하나당 하나의 최신 offset만 유지된다 — 상태 저장 topic이 자라서 디스크를 잡아먹는 일을 막는 설계다.

__consumer_offsets는 Kafka가 스스로 관리하는 내부 topic이다. partition 수·retention·cleanup.policy를 사용자가 변경하면 안 된다(장 02·07). 이 topic이 깨지면 모든 consumer group의 읽기 위치가 날아간다.

position vs committed offset — 메모리와 영속의 차이

consumer에는 "읽을 위치"가 두 개 있다:

• position: consumer 메모리에 있는 "다음에 읽을 offset". poll마다 갱신.
• committed offset: __consumer_offsets에 영속된 "읽었다고 commit한 위치".

이 둘이 항상 같지 않다는 게 at-least-once 중복의 원인이다. consumer가 position=100까지 읽고 처리했지만 commit을 position=95에만 해둔 채 크래시 나면, 재시작 시 committed=95부터 다시 읽는다 — offset 96~100이 중복 처리된다. 수동 commit에서 처리-후-commit 순서를 지키면 이 격차를 최소화하지만, 처리 중 크래시는 여전히 중복을 만들 수 있다(그래서 처리 로직이 멱등해야 한다).

auto-commit의 위험 — 메시지가 조용히 사라진다

기본값 enable.auto.commit=true는 poll 직후(정확히는 auto.commit.interval.ms 주기마다 poll 호출 시)에 offset을 넘길 수 있다. 처리 전에 commit이 일어나면, 처리 중 크래시 → 재시작 시 이미 commit된 offset부터 → 그 메시지들 영영 유실.

sequenceDiagram
    participant C as consumer (auto-commit)
    participant B as broker
    C->>B: poll() → batch[m1,m2,m3]
    Note over C: auto-commit: offset=m4 (처리 전!)
    C->>C: m1 처리 중 크래시
    Note over C: 재시작 → 다음 poll은 m4부터
    Note over C: m1,m2,m3 유실

해법: enable.auto.commit=false + 처리 후 수동 commit → at-least-once. 단, 재시도 시 중복 가능 → 처리 로직은 멱등(같은 입력 여러 번 처리해도 같은 결과)이어야 함.

auto-commit이 사실상 at-most-once인 이유

"auto-commit이 at-least-once를 깬다"는 말이 왜 정확한지 보자. auto-commit은 poll 시점(또는 interval 주기)에 자동으로 offset을 넘긴다. 처리가 끝나기 전이다. 그래서 처리 실패+크래시 시, 이미 넘긴 offset 이후의 메시지는 다시 안 온다 → 유실 방향. 반대로 처리 후 수동 commit이면, commit 전 크래시 시 재시작이 미처리 메시지를 다시 가져온다 → 중복 방향(at-least-once).

두 방향의 차이가 비즈니스에 미치는 영향이 다르다. 주문 처리라면 유실(주문 사라짐)이 중복(주문 두 번)보다 치명적일 수 있다. 메트릭 수집이라면 중복(숫자 부풀음)이 유실(데이터 빠짐)보다 치명적일 수 있다. 따라서 auto-commit vs 수동 commit의 선택은 "유실 vs 중복 중 어느 쪽을 감당할 것인가"라는 비즈니스 결정과 직결된다.

poll loop 건강 — 처리가 느리면 퇴출된다

consumer는 poll loop를 돌면서 일정 주기로 broker에 heartbeat를 보낸다. 처리가 너무 느려서 다음 poll까지 오래 걸리면, broker가 "이 consumer가 죽었거나 멈췄다"고 판단해 그룹에서 퇴출시키고 partition을 다른 consumer에게 넘긴다(rebalance, 장 05).

flowchart TD
    A["poll() → batch[m1..m100]"] --> B["for m in batch: process(m)"]
    B --> C{"처리 시간 > max.poll.interval.ms?"}
    C -->|"O"| D["broker: consumer 퇴출 → rebalance"]
    C -->|"X"| E["다음 poll()"]
    D --> F["partition이 다른 consumer에게 이관<br/>원래 consumer는 처리 중이었음 → 중복 가능"]

이 퇴출을 가르는 핵심 설정 두 개:

• max.poll.interval.ms(기본 300000=5분): 두 poll 사이의 최대 허용 시간. 이걸 넘기면 퇴출.
• max.poll.records(기본 500): 한 poll에 가져올 최대 record 수. 처리가 느린 로직이면 이 값을 줄여서 max.poll.interval.ms 안에 끝내게 해야 한다.

두 설정이 trade-off 관계다 — max.poll.records를 크게 하면 한 번에 많이 가져오지만 처리 시간이 길어져 max.poll.interval.ms를 넘길 위험이 크고, 작게 하면 안전하지만 poll 왕복이 잦아진다. 처리 시간이 변동적이면 보수적으로(작게) 잡는 게 정석이다. 퇴출+rebalance가 일어나면 partition이 다른 consumer에게 넘어가고, 원래 consumer가 처리 중이던 batch는 그 새 consumer에게도 다시 할당될 수 있어 중복이 발생한다 — 그래서 멱등 처리는 at-least-once의 필수 짝이다.

멱등 처리 — at-least-once의 필수 짝

at-least-once는 "중복이 일어날 수 있지만 유실은 없다"는 보장이다. 그런데 중복이 일어나면 어떻게 되는가 — 결제가 두 번 일어나거나, 알림이 두 번 가거나, 집계 숫자가 부풀거나 한다. 이걸 막으려면 처리 로직 자체가 멱등해야 한다: 같은 입력을 여러 번 처리해도 결과가 한 번 처리한 것과 같아야 한다.

멱등 처리의 전형 패턴:

• DB upsert: INSERT ... ON CONFLICT UPDATE 또는 MERGE. 같은 key가 오면 갱신만 하므로 중복 insert가 안 생긴다.
• 멱등 키(idempotency key): 메시지에 고유 ID를 두고, 처리 전에 "이미 처리한 ID인가"를 DB/캐시에서 확인. 이미 했으면 스킵.
• 상태 머신: 주문 상태가 pending → paid → shipped로 단방향 전이만 허용. paid 상태의 주문에 다시 paid 이벤트가 와도 무시.

멱등 처리를 안 하면 at-least-once는 "중복이 일어날 수 있다"는 경고일 뿐 안전망이 안 된다. Kafka consumer를 쓸 때 "이 처리가 중복 입력을 견디는가"를 항상 묻는 습관이 필요하다. 반대로 exactly-once(EOS, 장 08)는 멱등 로직 없이도 중복을 막아주지만, 비용(성능·복잡도)이 크므로 모든 곳에 쓰지는 않는다.

consumer.properties — 전체 설정 예제

신뢰성 우선 (수동 commit)

bootstrap.servers=localhost:9092
group.id=order-processors

key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

# 수동 commit (신뢰성)
enable.auto.commit=false

# poll loop 건강
max.poll.records=500
max.poll.interval.ms=300000
session.timeout.ms=45000
heartbeat.interval.ms=3000

# offset reset (committed 없을 때)
auto.offset.reset=earliest

처리량 튜닝

fetch.min.bytes=1048576    # 1MB까지 모아서 fetch
max.poll.records=1000

kafka-consumer-groups.sh — group 관리 명령

# group 목록
bin/kafka-consumer-groups.sh --list --bootstrap-server localhost:9092

# group 상세 (offset, LAG)
bin/kafka-consumer-groups.sh --describe --group order-processors --bootstrap-server localhost:9092

GROUP             TOPIC       PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
order-processors  orders      0          150             155             5
order-processors  orders      1          200             200             0

LAG — consumer가 못 따라가는 정도

LAG = LOG-END-OFFSET − CURRENT-OFFSET. LAG이 크고 자라면 consumer가 처리량 못 감당 → consumer 추가(≤ partition 수) 또는 처리 경량화.

LAG은 consumer 건강의 가장 직접적 지표다. LAG이 0이면 consumer가 producer를 따라잡고 있다는 뜻이고, 자라면 처리량이 입력량을 못 따른다는 뜻이다. 다만 LAG이 "일시적 스파이크"인지 "지속적 증가"인지를 구분해야 한다 — 트래픽 피크 때 잠깐 올랐다가 떨어지면 정상이지만, 시간이 지나도 줄지 않으면 consumer 병목이다.

flowchart LR
    P[producer] -->|LOG-END-OFFSET 증가| T[(topic)]
    C[consumer] -->|CURRENT-OFFSET 증가| T
    T -.->|"LAG = LOG-END − CURRENT"| M["LAG 메트릭<br/>(consumer 그룹별)"]
    M -->|지속 증가| A["consumer 추가 또는 처리 경량화"]

LAG 해소의 두 방향:

• consumer 추가: 같은 group에 consumer를 더 띄워 partition을 나눠 읽게 함(장 05). 단 partition 수가 상한이므로, consumer 수 > partition 수이면 남는 consumer는 idle.
• 처리 경량화: batch 처리, 비동기 I/O, 캐싱으로 한 consumer당 처리량을 올림. consumer를 늘리기 어울리지 않을 때(예: 외부 API 호출이 병목) 이 방향이 더 효과적.

LAG 모니터링 없이 Kafka consumer를 운영하면 병목을 감지 못 한다. kafka-consumer-groups.sh --describe 또는 Prometheus/Datadog 같은 모니터링으로 LAG을 그래프로 두고, 임계값 경보를 거는 게 운영 표준이다.

offset reset

# group의 offset을 처음으로 reset
bin/kafka-consumer-groups.sh --reset-offsets --group order-processors \
  --topic orders --to-earliest --execute --bootstrap-server localhost:9092

# 특정 offset으로 reset
bin/kafka-consumer-groups.sh --reset-offsets --group order-processors \
  --topic orders:0 --to-offset 100 --execute --bootstrap-server localhost:9092

# 최신으로 reset (skip backlog)
bin/kafka-consumer-groups.sh --reset-offsets --group order-processors \
  --topic orders --to-latest --execute --bootstrap-server localhost:9092

reset은 consumer가 활성 상태가 아닐 때 해야 안전하다(활성 중이면 reset과 자동 commit이 충돌). 보통 consumer를 모두 종료한 뒤 reset하고 다시 띄운다. --to-datetime으로 특정 시각으로 되돌리거나, --to-offset으로 정확한 offset을 지정할 수도 있다. 잘못 reset하면 중복(replay) 또는 유실(skip)이 일어나므로, reset 전에 --dry-run으로 결과를 미리 보는 습관이 필요하다.

reset 옵션	의미	결과
--to-earliest	partition 시작으로	전체 replay(중복 가능)
--to-latest	partition 끝으로	backlog 스킵(유실 가능)
--to-offset N	특정 offset으로	정확한 재설정
--to-datetime	특정 시각으로	시각 기준 replay
--shift-by N	현재 offset ± N	상대 이동

Kafka 4.3 실습 — 직접 확인하기

읽기 + offset 확인

# Kafka 4.3 — 신규 group, 처음부터
bin/kafka-console-consumer.sh --topic cons-demo --group g1 \
  --from-beginning --bootstrap-server localhost:9092

# 같은 group으로 재실행 → committed offset부터 (출력 없음, 이미 끝)
bin/kafka-console-consumer.sh --topic cons-demo --group g1 \
  --bootstrap-server localhost:9092

확인할 것: 같은 group이면 committed offset 이어받음(--from-beginning 무시됨).

LAG 확인

bin/kafka-consumer-groups.sh --describe --group g1 --bootstrap-server localhost:9092

consumer 그룹 상태와 멤버 확인

# Kafka 4.3, KRaft 단일 노드
# 그룹 상태(Stable/PreparingRebalance/...)와 멤버 수
bin/kafka-consumer-groups.sh --describe --group g1 --state --bootstrap-server localhost:9092
# 각 멤버가 어떤 partition을 담당하는지
bin/kafka-consumer-groups.sh --describe --group g1 --members --verbose --bootstrap-server localhost:9092

확인할 것: STATE가 Stable이면 정상, PreparingRebalance/CompletingRebalance면 멤버십 변화 중. 멤버-partition 할당 분포도 볼 수 있다(장 05 상세).

offset reset으로 replay 관찰

# g1의 offset을 처음으로 되돌려 replay 효과 보기
bin/kafka-consumer-groups.sh --reset-offsets --group g1 --topic cons-demo \
  --to-earliest --execute --bootstrap-server localhost:9092
# 같은 group으로 consumer 다시 띄우면, 과거 메시지부터 다시 읽힘(replay)
bin/kafka-console-consumer.sh --topic cons-demo --group g1 --bootstrap-server localhost:9092

확인할 것: 이미 읽었던 메시지가 다시 나오는가. 이게 "consumer가 자기 offset만 기억하면 replay 가능"한 로그 모델의 직접 증명이다.

흔히 묻는 것, 흔히 틀리는 것

오해	정정
"consumer는 broker가 push"	pull(poll) 기반
"auto-commit이 안전하다"	반대. 처리 전 commit → 유실 위험
"offset은 메모리에만"	__consumer_offsets에 영구 저장
"--from-beginning이 항상 처음부터"	committed offset 있으면 무시됨
"처리가 느려도 괜찮다"	max.poll.interval.ms 초과 → 그룹 퇴출

더 깊이

• position vs committed offset: position(메모리, 다음 읽을 offset) vs committed(영구). 차이가 중복의 원인.
• partition별 개별 commit: commitSync(offsets)로 처리 완료한 partition만 부분 commit.
• auto.offset.reset: committed 없을 때만 동작(earliest/latest/none). earliest는 처음부터(과거 재처리), latest는 새 메시지부터(과거 스킵), none은 예외. 새 그룹이 처음 접할 때 이 설정이 동작한다.
• commitSync vs commitAsync: commitSync는 ack 대기(느리지만 확실), commitAsync는 비동기(빠르지만 실패 시 무시될 수 있음). 처리량이 중요하면 async를 쓰되, 종료 시점엔 sync로 확정 commit하는 패턴이 흔하다.
• static membership(KIP-345): group.instance.id로 consumer identity를 고정하면 재시작 시 rebalance를 최소화한다(장 05). 상태 저장 consumer에 유리.
• consumer interceptor: ConsumerInterceptor로 poll 전후에 후킹(메트릭·로깅)을 넣을 수 있다.

요약 — 이 글의 결론

• consumer = 앱 내장 클라이언트. poll()로 topic에서 읽어 처리. pull 기반.
• offset = 읽기 커서. __consumer_offsets에 영구 저장(compact). position(메모리)과 committed(영속)이 다를 수 있다.
• auto-commit 위험: 처리 전 commit → 유실(at-most-once 방향). enable.auto.commit=false + 수동 commit → at-least-once.
• at-least-once → 중복 가능 → 처리 로직은 멱등 필수(DB upsert, 멱등 키, 상태 머신).
• poll loop 건강: 처리가 max.poll.interval.ms 초과 → 퇴출 + rebalance → 중복 가능. max.poll.records를 처리 시간에 맞게 조절.
• consumer.properties: group.id, enable.auto.commit=false, poll loop 설정(max.poll.records/interval, session.timeout).
• LAG = LOG-END-OFFSET − CURRENT-OFFSET. consumer 지연의 가장 직접적 지표. 지속 증가 시 consumer 추가 또는 처리 경량화.
• consumer-groups.sh: list, describe(LAG 확인), reset-offsets.

생각해 볼 문제

1. consumer가 "pull"인 것이 왜 설계상 유리한가? backpressure·replay 관점에서.
2. offset이 영구 저장된다는 것이 왜 중요한가? __consumer_offsets가 compact인 이유는?
3. auto-commit이 메시지 유실을 일으키는 과정을 설명하라. 왜 사실상 at-most-once인가?
4. 수동 commit으로 at-least-once를 달성하면 왜 중복이 생기는가? position과 committed의 차이.
5. consumer.properties에서 신뢰성을 위한 핵심 설정 3가지는?
6. LAG이 자라면 어떤 조치를 취할 수 있는가? 두 방향의 trade-off는?
7. kafka-consumer-groups.sh로 LAG을 확인하는 명령은?
8. 처리가 느려서 max.poll.interval.ms를 넘기면 어떤 일이 벌어지는가? 중복과 어떻게 연결되나?
9. at-least-once에서 멱등 처리가 필수인 이유는? 멱등하게 만드는 세 가지 패턴을 말하라.
10. auto.offset.reset=earliest/latest/none이 각각 언제 의미 있는가?

---

참고

• KafkaConsumer.java (poll loop) - 접근 2026-07-09
• Implementation — distribution.md (offset tracking) - 접근 2026-07-09
• Consumer configs - 접근 2026-07-09