Kafka - 04. consumer

consumer — topic에서 메시지를 읽는 주체

orders topic에 주문 이벤트가 쌓인다. 이제 그 메시지를 읽어 처리해야 한다 — 주문 처리 서비스가 읽어가고, 분석 서비스가 따로 읽는다. 이 "topic에서 메시지를 읽는 주체"가 consumer다.

이 글은 consumer가 뭔지부터 시작해, 실제 설정과 코드, auto-commit의 위험, LAG 모니터링까지 다룬다.

consumer란 — topic에서 메시지를 읽는 클라이언트

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)
}

consumer는 어떻게 쓰이나 — worked scenario

주문 처리 서비스: 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 — consumer가 자기 속도 제어(backpressure), replay(과거부터 다시) 자연스러움.

offset — 읽기 커서, 영속 저장

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

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

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

기본값 enable.auto.commit=true는 poll 직후(처리 전)에 offset을 넘길 수 있다. 처리 중 크래시 → 재시작 시 이미 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. 단, 재시도 시 중복 가능 → 처리 로직은 멱등(같은 입력 여러 번 처리해도 같은 결과)이어야 함.

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 수) 또는 처리 경량화.

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

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는 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).

요약 — 이 글의 결론

• consumer = 앱 내장 클라이언트. poll()로 topic에서 읽어 처리. pull 기반.
• offset = 읽기 커서. __consumer_offsets에 영구 저장.
• auto-commit 위험: 처리 전 commit → 유실. enable.auto.commit=false + 수동 commit → at-least-once.
• at-least-once → 중복 가능 → 처리 로직은 멱등 필수.
• consumer.properties: group.id, enable.auto.commit=false, poll loop 설정.
• LAG = LOG-END-OFFSET − CURRENT-OFFSET. consumer 지연 지표.
• consumer-groups.sh: list, describe(LAG 확인), reset-offsets.

생각해 볼 문제

1. consumer가 "pull"인 것이 왜 설계상 유리한가?
2. offset이 영구 저장된다는 것이 왜 중요한가?
3. auto-commit이 메시지 유실을 일으키는 과정을 설명하라.
4. 수동 commit으로 at-least-once를 달성하면 왜 중복이 생기는가?
5. consumer.properties에서 신뢰성을 위한 핵심 설정 3가지는?
6. LAG이 자라면 어떤 조치를 취할 수 있는가?
7. kafka-consumer-groups.sh로 LAG을 확인하는 명령은?

---

참고

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