> ## Documentation Index > Fetch the complete documentation index at: https://blog.nvim.me/llms.txt > Use this file to discover all available pages before exploring further. # Consumer group의 읽기 위치 > Consumer group의 partition 소유권, current position, committed offset, lag를 처리 경계와 함께 연결해요. > 배송 service가 재시작되면, 처음부터 모든 주문을 다시 배송할까요? Consumer가 record를 읽는 동안에는 memory에 “지금 여기까지 봤다”는 위치가 있어요. 그런데 process가 멈추면 그 memory도 사라지죠. Kafka consumer group은 다시 시작할 위치를 **committed offset**으로 남겨요. 여기서 가장 많이 생기는 오해가 하나 있어요. > “Offset을 commit했으니 업무도 성공했다.” Offset commit은 Kafka에 다음 읽기 위치를 기록하는 동작이에요. Database update, 택배 접수, email 발송이 실제로 성공했다는 증명은 아니에요. [앞 글](/messaging/kafka/topic-partition-key-and-ordering)에서 두 주문이 key에 따라 partition 0과 1로 나뉘는 모습을 봤어요. 이번에는 배송팀과 분석팀이 같은 topic을 각자의 위치에서 읽고, lag가 어떻게 계산되는지 따라가 볼게요. 이 글의 동작 설명과 CLI 예제는 **Apache Kafka 4.3.1**을 기준으로 해요. Docker는 `apache/kafka:4.3.1`, Ubuntu 직접 설치는 OpenJDK 21과 `kafka_2.13-4.3.1.tgz`를 사용했어요. Console consumer의 설정은 application code의 정확한 처리·commit 설계를 대신하지 않아요. Group과 offset을 눈으로 확인하기 위한 실습이에요. *** ## 같은 목적의 consumer는 partition을 나눠 맡아요 배송 service instance 두 개가 같은 `shipping-service` group으로 `order-sequence`를 읽는다고 해볼게요. ```mermaid theme={null} flowchart LR subgraph T[order-sequence topic] P0[partition 0
order-41] P1[partition 1
order-45] end subgraph S[shipping-service group] S1[shipping consumer A] S2[shipping consumer B] end subgraph A[analytics-service group] A1[analytics consumer] end P0 --> S1 P1 --> S2 P0 --> A1 P1 --> A1 ``` 같은 group 안에서는 한 partition을 한 consumer가 맡아요. 그래서 배송 consumer A와 B는 record를 중복해서 나눠 갖는 것이 아니라 partition 소유권을 나눠 가져요. 반면 `analytics-service`는 group ID가 달라요. 배송팀이 이미 읽었어도 분석팀은 자기 committed offset에서 같은 record를 독립적으로 읽을 수 있어요. | 관계 | Record를 읽는 모습 | | ------------------------ | ---------------------------------- | | 같은 group, consumer 2개 | Partition을 나눠서 한 번의 업무 흐름을 병렬 처리해요 | | 다른 group 2개 | 각 group이 같은 record를 자기 목적에 맞게 읽어요 | | Partition보다 consumer가 많음 | 남는 consumer는 partition을 할당받지 못해요 | *** ## Offset에는 세 위치가 있어요 “현재 offset이 몇이에요?”라고만 물으면 서로 다른 값을 섞기 쉬워요. ```mermaid theme={null} flowchart LR R0[offset 0
처리 완료] R1[offset 1
처리 완료] R2[offset 2
poll로 가져옴] R3[offset 3
아직 fetch 전] E[log end offset 4] R0 --> R1 --> R2 --> R3 --> E C[committed offset 2
다음 재시작 위치] -.-> R2 P[current position 3
다음 poll 위치] -.-> R3 ``` ### Record offset 각 record가 partition log 안에서 가진 위치예요. 위 그림의 record offset은 `0, 1, 2, 3`이에요. ### Current position 현재 실행 중인 consumer가 다음에 가져올 위치예요. `poll()`이 record를 반환하면 application이 업무를 끝내기 전이라도 position은 앞으로 움직일 수 있어요. ### Committed offset Consumer group이 재시작할 때 사용할 다음 위치예요. Offset 0과 1 처리를 끝내고 committed offset이 2라면, 재시작 시 offset 2부터 읽어요. 보통 committed offset은 **다음에 읽을 위치**예요. Offset 1까지 처리했다면 2를 commit해요. 이 차이를 놓치면 한 record를 건너뛰거나 다시 읽는 off-by-one 오류가 생겨요. *** ## Group을 만들어 committed offset과 lag를 확인해요 [앞 글](/messaging/kafka/topic-partition-key-and-ordering)의 `order-sequence` topic에 네 record가 남아 있고 `shipping-service`, `analytics-service` group은 아직 없다고 가정해요. 이미 이 실습을 실행했다면 `topic-ordering-ready` snapshot을 복원하거나 새로운 disposable group ID를 사용하세요. ```bash theme={null} sudo docker exec aha-kafka \ /opt/kafka/bin/kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic order-sequence \ --group shipping-service \ --from-beginning \ --max-messages 4 \ --formatter-property print.partition=true \ --formatter-property print.offset=true ``` `--from-beginning`은 이 group에 committed offset이 아직 없을 때 earliest record부터 시작하게 해요. ```bash theme={null} sudo docker exec aha-kafka \ /opt/kafka/bin/kafka-consumer-groups.sh \ --bootstrap-server localhost:9092 \ --group shipping-service \ --describe ``` ```bash theme={null} sudo docker exec aha-kafka \ /opt/kafka/bin/kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic order-sequence \ --group shipping-service \ --from-beginning \ --timeout-ms 5000 ``` 새 record가 없으면 앞의 네 record를 출력하지 않고 timeout으로 끝나요. 기존 committed offset이 `--from-beginning`보다 우선하기 때문이에요. ```bash theme={null} sudo docker exec aha-kafka \ /opt/kafka/bin/kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic order-sequence \ --group analytics-service \ --from-beginning \ --max-messages 4 ``` 먼저 inactive 상태인 `shipping-service`가 어느 위치로 이동할지 확인해요. `--dry-run`에는 변경이 없습니다. ```bash theme={null} sudo docker exec aha-kafka \ /opt/kafka/bin/kafka-consumer-groups.sh \ --bootstrap-server localhost:9092 \ --group shipping-service \ --topic order-sequence \ --reset-offsets \ --to-earliest \ --dry-run ``` Partition 0과 1의 `NEW-OFFSET`이 모두 0인지 확인한 뒤 같은 범위를 실행해요. ```bash theme={null} sudo docker exec aha-kafka \ /opt/kafka/bin/kafka-consumer-groups.sh \ --bootstrap-server localhost:9092 \ --group shipping-service \ --topic order-sequence \ --reset-offsets \ --to-earliest \ --execute ``` ```bash theme={null} sudo docker exec aha-kafka \ /opt/kafka/bin/kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic order-sequence \ --group shipping-service \ --max-messages 4 \ --formatter-property print.partition=true \ --formatter-property print.offset=true ``` Kafka를 압축 해제한 `kafka_2.13-4.3.1` directory에서 실행해요. ```bash theme={null} bin/kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic order-sequence \ --group shipping-service \ --from-beginning \ --max-messages 4 \ --formatter-property print.partition=true \ --formatter-property print.offset=true ``` ```bash theme={null} bin/kafka-consumer-groups.sh \ --bootstrap-server localhost:9092 \ --group shipping-service \ --describe ``` ```bash theme={null} bin/kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic order-sequence \ --group shipping-service \ --from-beginning \ --timeout-ms 5000 ``` 새 record가 없으면 앞의 네 record를 출력하지 않고 timeout으로 끝나요. ```bash theme={null} bin/kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic order-sequence \ --group analytics-service \ --from-beginning \ --max-messages 4 ``` ```bash theme={null} bin/kafka-consumer-groups.sh \ --bootstrap-server localhost:9092 \ --group shipping-service \ --topic order-sequence \ --reset-offsets \ --to-earliest \ --dry-run ``` Partition 0과 1의 `NEW-OFFSET`이 모두 0인지 확인한 뒤 실행해요. ```bash theme={null} bin/kafka-consumer-groups.sh \ --bootstrap-server localhost:9092 \ --group shipping-service \ --topic order-sequence \ --reset-offsets \ --to-earliest \ --execute ``` ```bash theme={null} bin/kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic order-sequence \ --group shipping-service \ --max-messages 4 \ --formatter-property print.partition=true \ --formatter-property print.offset=true ``` 두 트랙에서 네 record를 모두 읽은 뒤 확인한 위치는 같았어요. Partition 행의 출력 순서는 달라질 수 있습니다. ```text theme={null} Consumer group 'shipping-service' has no active members. GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG shipping-service order-sequence 0 2 2 0 shipping-service order-sequence 1 2 2 0 ``` Console consumer가 끝났으므로 active member는 없지만, group의 committed offset은 남아 있어요. `analytics-service`는 group ID가 다르므로 자기 초기 위치에서 같은 네 record를 읽어요. 배송팀의 위치는 분석팀의 위치를 움직이지 않아요. `shipping-service`의 두 partition에서 `CURRENT-OFFSET=2`, `LOG-END-OFFSET=2`, `LAG=0`을 확인하고, 새 `analytics-service`가 같은 네 record를 읽었으며, dry run 뒤 earliest offset으로 옮긴 `shipping-service`가 네 record를 replay했다면 독립적인 위치와 reset 경계를 확인한 거예요. *** ## Lag는 아직 따라가지 못한 위치 차이예요 Consumer group 설명에서 partition별 lag는 다음처럼 읽을 수 있어요. ```text theme={null} LAG = LOG-END-OFFSET - CURRENT-OFFSET ``` Partition 0의 log end offset이 120이고 group의 current offset이 100이라면 position 차이는 20이에요. 하지만 이것만 보고 “20건의 배송이 실패했다”고 말하면 안 돼요. * Producer가 지금도 record를 쓰고 있다면 lag는 계속 변해요. * Record마다 처리 시간이 다르면 같은 lag라도 따라잡는 시간이 달라요. * Consumer가 poll한 뒤 처리 중인 record는 committed offset 관점의 lag에 남을 수 있어요. * Offset을 먼저 commit했다면 업무가 끝나지 않았어도 lag가 줄어 보일 수 있어요. * Partition별 skew가 있으면 topic 합계만으로 병목 partition을 놓칠 수 있어요. 운영에서는 partition별 lag, lag가 증가하는 속도, record 처리 시간, 오류율, producer 유입량을 함께 봐야 해요. Lag는 group의 읽기 위치와 log 끝의 차이예요. Database 반영이나 외부 API 호출 같은 side effect가 성공했는지는 application metric과 business 상태로 별도 확인해야 해요. *** ## Poll, 처리, commit 순서가 전달 의미를 만들어요 Consumer가 record를 받았다는 한 문장 안에는 세 단계가 숨어 있어요. Consumer client가 broker에서 record를 가져와 application에 반환해요. Current position은 앞으로 갈 수 있어요. 배송 row를 만들거나 외부 택배 API를 호출해요. Kafka 밖의 side effect가 일어나는 경계예요. Group이 다음에 시작할 offset을 Kafka에 저장해요. 처리와 commit의 순서를 바꾸면 실패 시 모습도 달라져요. ### 처리한 뒤 commit하면 같은 record를 다시 받을 수 있어요 ```mermaid theme={null} sequenceDiagram participant K as Kafka participant C as Shipping consumer participant D as Shipping DB K->>C: offset 7 poll C->>D: 배송 준비 저장 성공 C-xC: commit 전 process 종료 K->>C: 재시작 뒤 offset 7 다시 전달 ``` Database 저장은 끝났지만 offset commit 전에 process가 죽었어요. 재시작한 consumer는 committed offset부터 같은 record를 다시 읽을 수 있어요. 이것이 **중복 전달 가능성**이 생기는 한 경계예요. 이 방식에서는 record를 놓칠 위험을 줄이는 대신, side effect가 중복되지 않도록 `eventId`의 unique constraint나 이미 처리한 event table 같은 idempotency 장치가 필요해요. ### 처리 전에 commit하면 실패한 record를 건너뛸 수 있어요 ```mermaid theme={null} sequenceDiagram participant K as Kafka participant C as Shipping consumer participant D as Shipping DB K->>C: offset 7 poll C->>K: offset 8 commit C-xD: 배송 준비 저장 실패 Note over C,K: 재시작 위치는 offset 8 ``` 다음 위치를 먼저 commit했는데 business processing이 실패했어요. 재시작해도 group은 offset 8부터 시작하므로 offset 7을 자동으로 다시 처리하지 않아요. Kafka transaction과 idempotent producer가 Kafka 안의 consume-process-produce 흐름을 다루더라도, 임의의 database write, email, 택배 API 호출이 자동으로 한 번만 실행되는 것은 아니에요. Kafka 밖의 side effect는 그 system과 함께 일관성 전략을 설계해야 해요. *** ## Auto commit은 처리 완료 시점을 대신 판단해주지 않아요 Kafka 4.3 consumer의 `enable.auto.commit` 기본값은 `true`예요. 이 설정은 consumer가 background에서 주기적으로 offset을 commit하게 해요. 하지만 auto commit timer는 application의 business processing이 성공했는지 알지 못해요. `poll()`로 받은 batch를 오래 처리하거나 별도 thread에 넘기면, 처리 완료와 commit 시점이 어긋날 수 있어요. 그래서 application에서는 사용하는 client와 framework의 다음 동작을 정확히 확인해야 해요. * `poll()`이 반환한 record 범위 * Auto commit 여부와 interval * Synchronous 또는 asynchronous manual commit 시점 * Listener가 정상 반환하거나 예외를 던질 때 framework가 ack하는 시점 * Retry 중 partition pause 여부 * Process 종료와 rebalance 때 진행 중인 record 처리 방식 Spring의 `@KafkaListener` 같은 framework를 쓰더라도 이 경계가 없어지는 것은 아니에요. Container가 poll과 commit을 대신 관리할 뿐, ack mode와 error handler 설정에 따라 실제 시점은 달라져요. *** ## Consumer가 늘거나 사라지면 rebalance가 일어나요 같은 group에 consumer가 합류하거나 떠나면 partition 소유권을 다시 나눠야 해요. 이 과정을 rebalance라고 해요. ```mermaid theme={null} stateDiagram-v2 [*] --> OneConsumer: consumer A 시작 OneConsumer --> Rebalancing: consumer B 합류 Rebalancing --> TwoConsumers: partition 재할당 TwoConsumers --> Rebalancing: consumer A 종료 Rebalancing --> OneConsumer: consumer B가 두 partition 소유 ``` Rebalance 자체는 group이 scale하고 장애에서 회복하기 위한 정상 동작이에요. 다만 처리 중인 record, commit 시점, partition-local cache가 있다면 소유권이 바뀔 때 정리해야 해요. Consumer가 `max.poll.interval.ms` 안에 다시 poll하지 못하면 group이 실패한 member로 판단하고 partition을 재할당할 수 있어요. Kafka 4.3 client의 기본값은 5분이지만, 이 값을 늘리는 것만으로 느린 business processing이 해결되지는 않아요. Batch 크기, 처리 시간, concurrency, timeout과 idempotency를 함께 봐야 해요. *** ## Replay는 offset을 옮기는 운영 작업이에요 새 분석 logic으로 과거 record를 다시 계산하려면 새 group ID로 시작하거나 기존 group의 offset을 앞쪽으로 reset할 수 있어요. 하지만 기존 배송 group의 offset을 움직이면 실제 배송 side effect가 다시 실행될 수 있어요. Replay는 “record를 다시 읽기”와 “business action을 다시 실행하기”를 분리해 설계해야 해요. 먼저 대상 topic, partition, 현재 offset, 이동할 offset, retention 범위, consumer 정지 여부를 확인하세요. Dry run과 별도 replay group을 우선 검토하고, idempotency가 없는 배송·결제·알림 작업에는 기존 group reset을 바로 적용하지 마세요. Retention으로 옛 record가 이미 제거됐다면 그 이전 offset으로 돌아갈 수 없어요. Replay 가능 범위는 topic이 실제로 보관 중인 log 범위 안이에요. *** ## 자, 정리해볼까요? * 같은 consumer group의 consumer들은 partition 소유권을 나눠 갖고, 다른 group은 같은 record를 독립적으로 읽어요. * Record offset, 실행 중인 current position, group의 committed offset은 서로 다른 위치예요. * Committed offset은 보통 마지막 처리 record가 아니라 다음에 읽을 위치예요. * Lag는 log end offset과 committed position의 차이지, business 성공률이 아니에요. * Poll, business processing, offset commit의 순서에 따라 중복 또는 누락 가능성이 달라져요. * Replay와 rebalance에서도 side effect와 idempotency를 함께 설계해야 해요. Key가 record를 어느 partition으로 보내고 ordering 범위를 어떻게 만드는지 다시 봐요. KafkaTemplate과 KafkaListener를 쓰기 전에 delivery, retry, idempotency 경계를 연결해요.