코딩 보관소

정확히 한 번

카프카는 기본적으로 최소 한 번(at-least-once) 전달을 보장한다.

여기에 멱등적 프로듀서와 트랜잭션 두 기능을 더하면 정확히 한 번(exactly-once) 처리가 가능하다.

멱등적 프로듀서 (Idempotent Producer)

해결하는 문제

• 브로커가 메시지를 저장했지만 응답(ack)이 유실되면 프로듀서가 재시도하면서 중복 메시지가 생긴다.

동작 방식

• 프로듀서마다 고유한 프로듀서 ID(PID) 가 부여되고, 파티션별로 시퀀스 번호가 메시지에 붙는다.
• 브로커는 PID/파티션별로 최근 시퀀스 번호(최대 5개 배치)를 기억하고, 이미 받은 번호가 오면 중복으로 판단해 저장하지 않는다.
• 번호가 건너뛰어 도착하면 OutOfOrderSequenceException이 발생한다.

사용법과 제약

• enable.idempotence=true (카프카 3.0부터 기본값)
• 조건: acks=all, retries > 0, max.in.flight.requests.per.connection ≤ 5
• 한계
  • 프로듀서 내부 재시도로 인한 중복만 막는다. 애플리케이션이 send()를 두 번 호출하거나, 프로듀서 인스턴스 여러 개가 같은 데이터를 보내는 경우는 막지 못한다.
  • 프로듀서가 재시작되면 새 PID를 받으므로 이전 상태가 이어지지 않는다.

트랜잭션 개요

보장하는 것

• 여러 파티션/토픽에 대한 원자적 쓰기 (전부 성공하거나 전부 실패)
• 카프카 내부 읽기-처리-쓰기(consume-process-produce) 패턴에서 정확히 한 번 처리

보장하지 않는 것

• 외부 시스템에 대한 부수 효과(이메일 발송, REST 호출 등)
• 카프카에서 읽어 외부 DB에 쓰는 경우 (카프카 → DB)
• DB에서 읽어 카프카에 쓰는 경우 (DB → 카프카)
• 클러스터 간 복제
• 발행/구독 패턴에서 컨슈머 측의 중복 처리 방지

트랜잭션이 해결하는 두 가지 문제

1. 애플리케이션 크래시: 처리 후 오프셋 커밋 전에 죽어 재처리하면 결과가 중복된다. 결과 쓰기와 오프셋 커밋을 한 트랜잭션으로 묶어 해결한다.
2. 좀비 애플리케이션: 리밸런싱 후에도 이전 인스턴스가 살아 있으면 같은 데이터를 중복 처리한다. 펜싱(fencing) 으로 막는다.

좀비 펜싱

• 전통적 방식: transactional.id를 지정하면, 같은 ID로 새 프로듀서가 initTransactions()를 호출할 때 에포크(epoch) 가 증가하고 옛 프로듀서의 요청은 ProducerFencedException으로 거부된다.
• KIP-447 (카프카 2.5+): 입력 파티션마다 transactional.id를 따로 두지 않아도 된다. sendOffsetsToTransaction(offsets, consumer.groupMetadata())로 컨슈머 그룹 메타데이터(세대 ID 등)를 이용해 펜싱한다. 프로듀서 하나로 여러 파티션을 처리할 수 있어 구조가 단순해지고 자원 사용량이 줄었다.

트랜잭션 사용 흐름 (코드)

producer.initTransactions();          // 한 번만 호출, 좀비 펜싱 수행
while (true) {
    records = consumer.poll(...);
    producer.beginTransaction();
    for (record : records) {
        producer.send(변환된 결과);
    }
    producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
    producer.commitTransaction();     // 실패 시 abortTransaction()
}

• 컨슈머는 enable.auto.commit=false로 두고, 오프셋 커밋을 트랜잭션에 맡긴다.
• ProducerFencedException 등 복구 불가능한 예외가 나면 프로듀서를 닫고 종료해야 한다. abortTransaction() 후 재시도할 수 있는 예외도 있다.

컨슈머 설정: isolation.level

• read_uncommitted (기본값): 커밋 여부와 관계없이 모든 메시지를 읽는다. 중단된 트랜잭션의 메시지도 보인다.
• read_committed: 커밋된 트랜잭션의 메시지와 트랜잭션에 속하지 않은 메시지만 읽는다. LSO(Last Stable Offset) 까지만 읽으므로, 진행 중인 트랜잭션이 있으면 그 뒤 메시지는 트랜잭션이 끝날 때까지 전달되지 않아 지연이 늘어날 수 있다.

성능 고려사항

• 트랜잭션 오버헤드는 트랜잭션 안의 메시지 수와 거의 무관한 트랜잭션당 고정 비용(코디네이터 호출, 마커 쓰기)이다. 한 트랜잭션에 메시지를 많이 담을수록 처리량이 좋아지고, 짧게 끊을수록 지연이 줄어든다.
• initTransactions()는 비용이 크므로 프로듀서 생성 시 한 번만 호출한다.

Kafka Streams

• processing.guarantee=exactly_once_v2 설정만으로 정확히 한 번 처리를 켤 수 있다. (구버전의 exactly_once는 exactly_once_v2로 대체되었다.)
• EOS 모드에서는 commit.interval.ms 기본값이 100ms로 짧아진다.

트랜잭션 동작 흐름과 개념

구성 요소와 용어

용어

트랜잭션 상태 (코디네이터 관점)

코디네이터는 각 transactional.id마다 다음 상태 중 하나를 트랜잭션 로그에 기록한다.

상태 의미

상태 전이 (정상 경로):

Empty ──(파티션 등록)──▶ Ongoing ──(EndTxn commit)──▶ PrepareCommit ──(마커 기록 완료)──▶ CompleteCommit
                           │
                           └──(EndTxn abort / 타임아웃)──▶ PrepareAbort ──(마커 기록 완료)──▶ CompleteAbort

단계별 동작 흐름

아래 흐름은 전통적인 프로토콜 기준이다. 카프카 최근 버전에서는 KIP-890에 따라 일부 RPC(특히 AddPartitionsToTxn)의 호출 방식이 달라지고 있으므로, 정확한 요청 종류는 사용하는 버전의 문서를 확인해야 한다.

개념적 순서(등록 → 쓰기 → 커밋 결정 → 마커 기록)는 동일하다.

1) initTransactions()

1. 프로듀서가 아무 브로커에 FindCoordinator(타입: TRANSACTION) 요청을 보내 자신의 transactional.id를 담당하는 코디네이터를 찾는다.
2. 코디네이터에 InitProducerId 요청을 보낸다.
3. 코디네이터의 처리:
   • 처음 보는 transactional.id: 새 PID를 발급하고 에포크를 0으로 시작한다.
   • 이미 존재하는 transactional.id: 에포크를 증가시킨다. 이 시점부터 이전 에포크를 가진 프로듀서(좀비)의 요청은 모두 거부된다.
   • 이전 인스턴스가 남긴 미완료 트랜잭션이 있으면 중단 처리(또는 이미 커밋 결정된 경우 커밋 완료)한 뒤에야 응답한다.
4. 코디네이터는 transactional.id → PID, 에포크, 트랜잭션 타임아웃 매핑을 트랜잭션 로그에 기록한다.

이 호출은 프로듀서당 한 번만 수행하며, 재시작 시 이전 인스턴스를 정리하는 역할도 겸한다.

2) beginTransaction()

• 로컬 상태만 변경한다. 브로커로 요청을 보내지 않는다.
• 이후 send(), sendOffsetsToTransaction() 호출이 트랜잭션에 속하게 된다.

3) send() - 트랜잭션 내 쓰기

1. 이 트랜잭션에서 처음 쓰는 파티션이면, 프로듀서가 코디네이터에 해당 파티션을 등록한다(AddPartitionsToTxn). 코디네이터는 파티션 목록을 트랜잭션 로그에 기록하고 상태를 Ongoing으로 만든다.
2. 프로듀서가 해당 파티션의 리더 브로커에 Produce 요청을 보낸다. 요청에는 PID, 에포크, 시퀀스 번호, 트랜잭션 여부가 포함된다.
3. 리더는 이 레코드를 즉시 로그에 기록하고 복제한다. 커밋 전이라도 로그에는 이미 존재한다. 레코드에는 "트랜잭션에 속함" 표시가 붙는다.
4. 이 레코드를 read_uncommitted 컨슈머는 바로 읽을 수 있지만, read_committed 컨슈머는 트랜잭션이 끝날 때까지 읽지 못한다.
5. 해당 트랜잭션의 첫 레코드 오프셋이 파티션의 LSO 계산에 반영된다. 즉 트랜잭션이 열려 있는 동안 LSO는 그 첫 오프셋에서 멈춘다.

멱등성(PID + 시퀀스 번호)이 함께 동작하므로 트랜잭션 내 재시도 중복도 방지된다.

4) sendOffsetsToTransaction() - 오프셋을 트랜잭션에 포함

1. 프로듀서가 코디네이터에 컨슈머 그룹의 __consumer_offsets 파티션을 트랜잭션에 등록한다(AddOffsetsToTxn).
2. 프로듀서가 그룹 코디네이터에 TxnOffsetCommit 요청을 보낸다. 이 요청에는 컨슈머 그룹 메타데이터(그룹 ID, 세대 ID, 멤버 ID 등)가 포함된다.
3. 그룹 코디네이터는 세대 ID가 현재 세대와 맞는지 검사한다. 리밸런싱으로 세대가 바뀐 뒤 이전 세대의 인스턴스가 보낸 요청은 거부된다. 이것이 KIP-447 방식의 좀비 펜싱이다.
4. 오프셋은 보류(pending) 상태로 저장된다. 트랜잭션이 커밋되기 전에는 다른 컨슈머가 이 오프셋을 조회해도 반영되지 않는다.

결과적으로 "출력 레코드 쓰기"와 "입력 오프셋 커밋"이 같은 트랜잭션에 묶인다.

5) commitTransaction() - 2 Phase Commit

1단계: 커밋 결정 기록

1. 프로듀서가 코디네이터에 EndTxn(commit) 요청을 보낸다.
2. 코디네이터는 트랜잭션 로그에 PrepareCommit 을 기록한다.
3. 이 기록이 복제까지 완료되면 트랜잭션은 반드시 커밋된다. 이후 코디네이터가 장애로 교체되더라도 새 코디네이터가 로그를 읽어 커밋을 이어서 완료한다. 이 지점이 커밋 확정 시점(commit point)이다.
4. 코디네이터가 프로듀서에게 EndTxn 응답을 보낸다. 따라서 commitTransaction()이 반환되는 시점에는 커밋이 확정되어 있지만, 모든 파티션에 마커가 기록되었다는 의미는 아니다.

2단계: 마커 기록

1. 코디네이터가 트랜잭션에 참여한 모든 파티션의 리더(데이터 파티션 + 오프셋을 저장하는 __consumer_offsets 파티션)에 WriteTxnMarkers 요청을 보낸다.
2. 각 리더는 파티션 로그에 COMMIT 컨트롤 레코드를 추가한다.
3. 마커가 기록되면 그 파티션의 해당 트랜잭션은 종료된 것으로 처리되어 LSO가 앞으로 이동할 수 있게 된다. 보류 중이던 오프셋도 이 시점에 유효한 커밋 오프셋이 된다.
4. 모든 파티션에 마커가 기록되면 코디네이터는 트랜잭션 로그에 CompleteCommit 을 기록하고, 다음 트랜잭션을 받을 수 있는 Empty 상태가 된다.

6) abortTransaction()

커밋과 같은 구조이며 결정 기록이 PrepareAbort, 마커가 ABORT 컨트롤 레코드로 바뀐다. 중단된 트랜잭션의 데이터 레코드는 로그에서 삭제되지 않고 남아 있으며, read_committed 컨슈머가 걸러낸다. 보류 중이던 오프셋은 폐기된다.

4. 전체 시퀀스 (정상 커밋)

컨슈머 측 동작 (read_committed)

1. 브로커는 Fetch 응답에 LSO까지의 레코드만 포함한다. 진행 중인 트랜잭션의 레코드와 그 뒤에 쓰인 다른 레코드는 LSO 뒤에 있으므로 반환되지 않는다.
2. Fetch 응답에는 해당 범위에서 중단된 트랜잭션 목록(PID, 첫 오프셋)이 함께 담긴다.
3. 컨슈머 클라이언트는 이 목록과 컨트롤 레코드(ABORT 마커)를 이용해 중단된 트랜잭션의 데이터 레코드를 버린다. COMMIT된 레코드는 애플리케이션에 전달한다.
4. 컨트롤 레코드 자체는 애플리케이션에 전달되지 않지만 오프셋 번호는 차지한다. 그래서 오프셋이 연속적이지 않거나, 마지막 레코드의 오프셋 + 1이 다음 레코드의 오프셋과 다를 수 있다.
5. 지연 영향: 한 파티션에서 오래 열려 있는 트랜잭션이 하나만 있어도, 그 뒤에 쓰인 모든 메시지(다른 프로듀서의 것 포함)는 그 트랜잭션이 끝날 때까지 read_committed 컨슈머에게 보이지 않는다. 트랜잭션을 짧게 유지해야 하는 이유이다.

장애 시나리오

예외 처리

• 치명적(프로듀서를 닫아야 함)
  • ProducerFencedException, InvalidProducerEpochException, AuthorizationException, OutOfOrderSequenceException 등. 새 프로듀서 인스턴스로 다시 시작해야 한다.
• 복구 가능: 그 외 KafkaException 계열. abortTransaction()을 호출한 뒤 컨슈머 위치를 마지막 커밋 오프셋으로 되돌리고(seek) 처음부터 다시 처리한다.

중단 후 재시도할 때는 컨슈머의 읽기 위치를 마지막으로 커밋된 오프셋으로 되돌려야 한다. 그렇지 않으면 이미 poll()로 읽은 레코드가 처리되지 않고 건너뛰어진다.

관련 설정

보장 범위 정리

트랜잭션이 제공하는 것:

• 원자성: 한 트랜잭션에서 쓴 모든 파티션의 레코드와 오프셋 커밋이 함께 보이거나 함께 사라진다.
• 격리(읽기 측): read_committed 컨슈머는 커밋된 결과만 본다.
• 펜싱: 같은 논리적 프로듀서의 이전 인스턴스가 이후 쓰기를 할 수 없다.
• 내구성: 커밋 결정은 복제되는 트랜잭션 로그에 기록되어 브로커 장애 후에도 유지된다.

트랜잭션이 제공하지 않는 것:

• 카프카 밖의 상태(DB, 외부 API)에 대한 원자성
• 서로 다른 트랜잭션 간의 직렬화 보장 (잠금 개념 없음)
• 입력 토픽 소비와 무관한 중복 send() 호출 방지 (이것은 멱등적 프로듀서도 막지 못한다.)

1. Kafka에서 신뢰성이란?

Kafka는 대용량 이벤트 데이터를 안정적으로 전달하기 위한 분산 메시징 플랫폼이다

하지만 메시지를 Kafka에 저장한다고 해서 데이터 유실이나 중복 처리가 무조건 방지되는 것은 아니다

예를 들어 주문 서비스에서 주문 생성 이벤트를 Kafka에 전송했다고 가정했을 때 프로듀서가 메시지를 전송한 직후 네트워크 장애가 발생하면 다음과 같은 상황이 발생할 수 있다

• 브로커가 메시지를 저장하지 못했을 수 있다
• 브로커는 메시지를 저장했지만 프로듀서가 응답을 받지 못했을 수 있다
• 프로듀서가 재시도하면서 동일한 메시지가 중복 저장될 수 있다
• 컨슈머가 메시지를 처리한 뒤 오프셋을 커밋하기 전에 장애가 발생해 메시지를 다시 처리할 수 있다

이처럼 데이터 전달 과정에서는 여러 종류의 장애가 발생할 수 있다

그러므로 Kafka의 신뢰성을 이해하려면 프로듀서, 브로커, 컨슈머를 개별적으로 살펴보는 것뿐 아니라 전체 데이터 흐름을 함께 이해해야 한다.

Kafka의 신뢰성을 구성하는 핵심 요소

• 프로듀서가 메시지를 안전하게 전송하는 방법
• 브로커가 데이터를 복제하고 장애에 대응하는 방법
• 컨슈머가 메시지를 처리하고 오프셋을 관리하는 방법
• 재시도와 중복 처리 문제를 해결하는 방법

2. 신뢰성 있는 데이터 전달을 위한 고려 사항

Kafka에서 신뢰성을 설계할 때는 먼저 어떤 수준의 데이터 전달을 보장해야 하는지 정의해야 한다

일반적으로 데이터 전달 방식은 다음과 같이 구분할 수 있다

2.1 최대 한 번(At-most-once)

메시지를 최대 한 번 처리하는 방식

중복 처리를 방지하는 데 유리하지만, 장애 상황에서 메시지가 유실될 수 있다

예시

컨슈머가 메시지를 처리하기 전에 오프셋을 커밋하면, 처리 도중 장애가 발생하더라도 재시작 후 해당 메시지를 건너뛸 수 있다

데이터 유실보다 중복 처리가 더 큰 문제가 되는 상황에서는 고려할 수 있지만, 주문이나 결제처럼 데이터 유실이 허용되지 않는 시스템에서는 신중하게 적용해야 한다

2.2 최소 한 번(At-least-once)

메시지가 적어도 한 번 처리되도록 하는 방식

메시지 유실을 줄이는 데 유리하지만, 동일한 메시지가 여러 번 처리될 수 있다

예시

컨슈머가 데이터베이스에 주문을 저장한 뒤 오프셋을 커밋하기 전에 장애가 발생했다고 가정하고 재시작 후 Kafka는 해당 메시지가 아직 처리되지 않은 것으로 판단할 수 있으므로 같은 메시지를 다시 전달한다

이 경우 데이터베이스에는 동일한 주문이 두 번 저장될 수 있다

따라서 최소 한 번 전달을 사용하는 시스템에서는 중복 처리를 방지하는 별도의 설계가 필요하다

대표적으로 주문 ID를 고유 키로 사용하거나 데이터베이스의 고유 제약 조건을 활용할 수 있다

2.3 정확히 한 번(Exactly-once)

메시지가 중복 반영되지 않고 의도한 처리 결과가 한 번만 적용되도록 하는 방식

Kafka는 멱등적 프로듀서와 트랜잭션 등의 기능을 통해 특정 처리 흐름에서 정확히 한 번 의미 구조를 지원한다

다만 외부 데이터베이스나 외부 API 호출까지 자동으로 정확히 한 번 처리되는 것은 아니다

외부 시스템까지 포함하려면 트랜잭션 경계, 중복 제거, 멱등성 키 등의 추가적인 설계가 필요하다

세 가지 방식 중 무엇이 항상 가장 좋은 것은 아니다

데이터 유실, 중복 처리, 지연 시간 중 어떤 요소를 우선할 것인지에 따라 적절한 전략을 선택해야 한다

3. 브로커의 복제와 데이터 안정성

Kafka는 파티션의 복제본을 여러 브로커에 분산해 저장한다

특정 브로커에 장애가 발생하더라도 다른 복제본을 이용해 서비스를 지속할 수 있도록 하기 위해서다

예시

하나의 파티션에 복제 계수 3을 설정하면 다음과 같이 구성된다

• Leader: 프로듀서의 쓰기 요청을 처리하는 복제본
• Follower 1: 리더의 데이터를 복제하는 복제본
• Follower 2: 리더의 데이터를 복제하는 복제본

프로듀서는 일반적으로 리더에 메시지를 기록한다

팔로워는 리더의 데이터를 복제하며, 리더에 장애가 발생하면 조건을 충족하는 복제본이 새로운 리더로 선출될 수 있다

하지만 복제본이 여러 개 있다고 해서 데이터가 항상 안전한 것은 아니다

복제본이 충분히 동기화되지 않았거나 부적절한 설정으로 리더가 선출되면 데이터가 유실될 수 있다

3.1 ISR(In-Sync Replicas)

ISR은 리더와 동기화 상태를 유지하는 복제본의 집합

Kafka는 복제본의 동기화 상태를 고려해 장애 발생 시 리더를 선출한다

ISR에 포함된 복제본을 기반으로 리더를 선출하면 데이터 유실 위험을 줄일 수 있다

ISR에 포함되지 않은 복제본을 리더로 선출하는 것을 허용하는 것은 Option이다(Unclean Leader Eclection)

만약허용하면 서비스 가용성을 높일 수 있는 대신 데이터 유실 가능성이 커질 수 있다

3.2 replication.factor와 min.insync.replicas

두 설정은 함께 이해해야 한다

• replication.factor: 파티션 복제본의 총개수를 지정한다
• min.insync.replicas: 쓰기 요청을 성공으로 처리하기 위해 요구하는 최소 ISR 수를 지정한다

예시

복제 계수가 3이고 최소 ISR 수가 2라면, 세 복제본 중 최소 두 개가 ISR에 포함되어야 acks=all을 사용하는 쓰기 요청이 성공할 수 있다

브로커 한 대에 장애가 발생하더라도 나머지 두 복제본이 ISR을 유지한다면 쓰기를 계속할 수 있다

ISR이 하나만 남으면 데이터 안전성을 위해 쓰기 요청이 실패할 수 있다

이러한 설정은 장애 상황에서 무조건 쓰기를 허용하기보다 데이터 안정성을 우선하도록 만드는 장치다

3.3 Unclean Leader Election

Unclean Leader Election은 ISR에 포함되지 않은 복제본을 새로운 리더로 선출하는 방식이다

이 기능을 허용하면 ISR의 모든 복제본을 사용할 수 없는 상황에서도 서비스 복구가 가능할 수 있다

선택된 복제본에 최신 데이터가 없으면 일부 메시지가 유실될 수 있다

데이터 유실을 최소화하는 것이 중요한 시스템에서는 이 설정을 적용하는데 신중해야할 필요가 있다

4. 프로듀서의 신뢰성 설정

프로듀서는 Kafka에 메시지를 전송하는 역할을 담당한다

프로듀서의 설정에 따라 메시지 전송의 신뢰성과 지연 시간이 달라진다

4.1 acks 설정

acks는 프로듀서가 메시지 전송 성공을 판단하기 위해 기다리는 확인 수준을 결정한다

acks=all을 설정은 모든 Kafka 브로커의 응답을 기다리는 것이 아니다

> 현재 ISR에 속한 복제본들의 확인을 기준으로 한다

acks=all만 설정한다고 해서 충분한 복제 안정성이 자동으로 보장되는 것은 아니다

복제 계수, 최소 ISR 수, 리더 선출 정책 등을 함께 고려해야 한다

4.2 재시도(Retries)

프로듀서는 네트워크 오류나 일시적인 브로커 문제로 요청이 실패하면 재시도할 수 있다

재시도는 일시적인 장애로 인한 메시지 유실 가능성을 줄이는 데 도움이 된다

재시도에는 주의할 점이 있다

브로커가 메시지를 정상적으로 저장했지만 응답이 네트워크에서 유실되었다고 가정해 보자

프로듀서는 전송이 실패했다고 판단해 같은 메시지를 다시 보낼 수 있다

이 경우 중복 기록이 발생할 가능성이 있다

이러한 문제를 해결하기 위해 Kafka는 멱등적 프로듀서를 제공한다

4.3 멱등적 프로듀서(Idempotent Producer)

멱등적 프로듀서는 프로듀서의 재시도로 인해 동일한 메시지가 중복 기록되는 문제를 방지한다

enable.idempotence=true
acks=all

멱등성 기능은 프로듀서 ID와 시퀀스 번호 등을 활용해 재시도로 인한 중복 기록을 식별한다

* 멱등적 프로듀서는 애플리케이션이 동일한 주문 이벤트를 별도의 새로운 전송으로 두 번 생성하는 문제까지 자동으로 해결하지는 않는다

5. 컨슈머의 신뢰성과 오프셋 관리

컨슈머는 Kafka에서 메시지를 읽고 처리한다

이때 중요한 것은 메시지 처리와 오프셋 커밋의 순서다

오프셋은 컨슈머가 어디까지 읽었는지를 나타내는 위치 정보다

컨슈머가 오프셋을 커밋하면 재시작 후에도 해당 위치를 기준으로 소비를 이어갈 수 있다

5.1 오프셋을 먼저 커밋하면 발생하는 문제

컨슈머가 메시지를 읽은 직후 오프셋을 먼저 커밋하고, 이후 데이터베이스에 저장한다면

1. Kafka에서 메시지를 읽는다
2. 오프셋을 커밋한다
3. 데이터베이스 저장을 시도한다
4. 저장 과정에서 장애가 발생한다

이 경우 컨슈머가 재시작하면 이미 커밋된 위치 이후부터 메시지를 읽을 수 있다

결과적으로 데이터베이스에 반영되지 않은 메시지를 건너뛸 위험이 있다

5.2 처리 후 오프셋을 커밋하면 발생하는 문제

데이터베이스 저장을 먼저 완료하고 오프셋을 커밋한다면

1. Kafka에서 메시지를 읽는다
2. 데이터베이스에 저장한다
3. 오프셋을 커밋한다

이 방식은 처리 완료 전에 오프셋을 커밋하는 문제를 줄여준다

하지만 데이터베이스 저장이 성공한 뒤 오프셋 커밋 전에 장애가 발생하면 같은 메시지가 다시 처리될 수 있다

따라서 처리 후 커밋 방식은 일반적으로 최소 한 번 처리에 적합하지만, 중복 처리까지 방지하려면 추가적인 조치가 필요하다(멱등 처리가 필요하다)

5.3 자동 커밋과 수동 커밋

자동 커밋은 Kafka가 설정된 주기에 따라 오프셋을 커밋하도록 하는 방식이다

구현은 간단하지만 애플리케이션의 실제 처리 완료 시점과 커밋 시점이 일치하지 않을 수 있다

수동 커밋은 애플리케이션이 처리 완료 시점을 기준으로 오프셋 커밋을 직접 제어하는 방식이다

처리 순서와 오류 대응을 세밀하게 관리할 수 있지만, 커밋 실패나 재처리 상황을 직접 고려해야 한다

수동 커밋을 사용한다고 해서 중복 처리가 자동으로 사라지는 것은 아니다

중요한 것은 실제 처리 결과와 오프셋 상태 사이의 불일치를 어떻게 관리할 것인지다

6. 데이터 전달 신뢰성을 높이기 위한 설계

Kafka의 신뢰성을 높이려면 특정 설정 하나에 의존하기보다 전체 시스템의 동작을 함께 고려해야 한다

6.1 데이터 유실을 줄이는 구성

데이터 유실이 중요한 문제인 시스템에서는 다음 사항을 검토할 수 있다

• 충분한 복제 계수 설정
• 적절한 min.insync.replicas 설정
• 프로듀서의 acks=all 적용
• 멱등적 프로듀서 활성화
• 컨슈머의 처리 완료 후 오프셋 커밋
• 장애 상황에서의 복제 및 복구 테스트

이러한 설정은 데이터 유실 위험을 줄이는 데 도움이 되지만, 시스템 전체에서 데이터가 절대 유실되지 않는다는 의미는 아니다

6.2 중복 처리를 줄이는 구성

최소 한 번 전달 방식에서는 동일한 메시지가 다시 처리될 수 있으므로 중복 처리를 고려해야 한다

예시

주문 이벤트에 고유한 주문 ID를 부여하고 데이터베이스에서 해당 ID의 중복 저장을 방지할 수 있다

또는 처리 결과와 메시지 식별 정보를 함께 저장해 이미 처리한 이벤트인지 확인할 수 있다

이러한 방식은 Kafka의 메시지 전달 특성과 외부 시스템의 데이터 변경을 연결하는 데 유용하다

6.3 장애 상황을 가정한 테스트

설정이 올바르더라도 실제 장애 상황에서 예상한 대로 동작하는지 확인해야 한다

• 리더 브로커를 중단했을 때 새로운 리더가 정상적으로 선출되는가?
• ISR 수가 최소 요구치보다 적을 때 쓰기 요청이 어떻게 처리되는가?
• 프로듀서가 응답을 받지 못했을 때 재시도가 올바르게 작동하는가?
• 컨슈머가 데이터 처리 도중 종료되면 메시지가 어떻게 재처리되는가?
• 컨슈머 지연이나 복제 지연이 증가했을 때 이를 감지할 수 있는가?

장애 테스트를 통해 설정의 의미를 확인하고 운영 환경에서 발생할 수 있는 문제를 사전에 발견할 수 있다

1. Kafka의 데이터 저장 구조

Kafka는 Topic을 여러 Partition으로 나누어 저장하며, 각 Partition은 독립적인 로그를 유지한다

Topic: orders
 ├── Partition 0 → Broker A
 ├── Partition 1 → Broker B
 └── Partition 2 → Broker C

• Topic: 메시지를 논리적으로 구분하는 단위

• Partition: 데이터를 분산 저장하고 병렬 처리를 가능하게 하는 단위

• Log: 메시지가 순서대로 추가되는 기록

• Segment: 로그를 여러 파일로 나누어 관리하는 단위

• Offset: 파티션 내부에서 메시지의 위치를 식별하는 값

Kafka는 메시지를 소비했다고 바로 삭제하지 않고 보존 정책에 따라 유지하는 구조

2. 클러스터 멤버십

Kafka 클러스터에 참여하는 브로커의 상태와 구성을 관리하는 기능

ZooKeeper 기반 구조

• 브로커가 시작되면 고유 ID로 ZooKeeper에 등록하는 구조

• 브로커 등록에 임시 노드(Ephemeral Node)를 사용해 생존 상태를 관리하는 구조

• 브로커 종료나 세션 단절로 임시 노드가 사라지면 클러스터 구성 변경을 감지하는 구조

KRaft 기반 구조

• ZooKeeper 없이 Kafka 내부의 컨트롤러가 메타데이터를 관리하는 구조

• Raft 기반 합의 방식으로 메타데이터 로그를 복제하고 관리하는 구조

• 브로커 역할과 컨트롤러 역할을 하나의 노드에서 수행하거나 분리할 수 있는 구조

3. 컨트롤러

Kafka 클러스터 전체의 상태 변경을 조정한다

• 브로커의 추가 및 이탈에 대응하는 역할

• 파티션 리더 선출을 조정하는 역할

• 파티션 및 복제본의 상태 변경을 관리하는 역할

• 클러스터 메타데이터 변경을 조정하는 역할

컨트롤러가 메시지의 읽기와 쓰기를 직접 처리하는 것이 아니라 각 파티션의 리더 브로커가 일반적인 데이터 요청을 처리하는 구조

Controller Epoch

• 컨트롤러가 교체될 때 세대 번호를 통해 이전 컨트롤러의 오래된 명령을 구분하는 방식

• 이전 컨트롤러의 명령이 현재 클러스터 상태에 영향을 주는 문제를 방지하는 메커니즘

4. 복제(Replication)

파티션의 데이터를 여러 브로커에 복제하여 브로커 장애에 대응하는 메커니즘

Partition 0

Broker A        Broker B        Broker C
Leader          Follower        Follower
[0][1][2][3]    [0][1][2][3]    [0][1][2]

• Leader: Producer의 쓰기 요청을 처리하는 복제본

• Follower: 리더의 로그를 가져와 복제하는 복제본

• ISR: 리더와 동기화 상태를 유지하는 복제본 집합

• LEO: 개별 복제본의 로그 끝 위치

• HW: 일반 소비자가 읽을 수 있는 커밋 경계

복제 동작 과정

1. Producer가 리더에 메시지를 기록하도록 요청하는 과정

2. 리더가 자신의 로그에 메시지를 기록하는 과정

3. 팔로워가 리더의 로그를 복제하는 과정

4. 복제 상태에 따라 커밋 가능한 위치가 전진하는 과정

5. 리더 장애 시 컨트롤러가 적절한 복제본을 새 리더로 선출하는 과정

ISR에 속한다는 것은 Kafka가 정의한 복제 진행 및 생존 조건을 충족한다는 의미이며 모든 복제본의 상태가 매 순간 완전히 동일하다는 의미는 아니다

5. 요청 처리(Request Processing)

Kafka의 Producer와 Consumer는 브로커에 요청을 보내고 응답을 받는 방식으로 동작한다

Produce Request

• Producer가 파티션 리더에 메시지 기록을 요청하는 방식

• 여러 레코드를 배치로 묶어 네트워크 요청 횟수를 줄이는 방식

• 복제 및 확인 조건에 따라 요청 성공 여부가 결정되는 방식

Fetch Request

• Consumer가 브로커에서 메시지를 읽어 오는 방식

• Follower도 리더의 로그를 가져오기 위해 사용하는 방식

• 메시지를 배치 단위로 가져와 처리 효율을 높이는 방식

6. 물리적 저장소(Physical Storage)

Kafka가 디스크 기반 로그를 사용하면서 높은 처리량을 확보한다

순차 쓰기

• 로그 끝에 새로운 레코드를 추가하는 방식

• 순차적인 디스크 쓰기를 활용해 저장 효율을 높이는 방식

페이지 캐시

• 운영체제가 파일 데이터를 메모리에 캐싱하여 디스크 접근을 줄이는 방식

• Kafka가 JVM 힙뿐 아니라 운영체제의 페이지 캐시를 적극적으로 활용하는 구조

Zero-copy

• 파일 데이터를 네트워크로 전달할 때 메모리 복사와 사용자 공간 개입을 줄이는 최적화 방식

• 지원되는 환경과 전송 경로에서 sendfile 등의 기능을 활용하는 방식

Kafka의 높은 처리량은 순차 쓰기, 페이지 캐시, 배치 처리 및 Zero-copy 최적화가 함께 작용한 결과

7. 계층형 저장소(Tiered Storage)

로컬 디스크와 원격 저장소를 활용해 데이터 보관 비용과 저장 공간을 관리하는 구조이다

• 로컬 저장소: 최근 데이터에 빠르게 접근하기 위한 저장 영역

• 원격 저장소: 오래된 로그 세그먼트를 보관하기 위한 저장 영역

• 로컬 디스크의 저장 공간 부담을 줄이고 장기간 보관을 지원하는 구조

• 원격 저장소의 데이터는 로컬 데이터보다 접근 지연이 커질 수 있는 구조

계층형 저장소의 세부 기능과 지원 범위는 Kafka 버전에 따라 달라진다

8. 파티션 할당(Partition Allocation)

각 파티션의 리더와 복제본을 브로커에 분산 배치하는 과정이다

• 리더를 여러 브로커에 분산하여 요청 부하를 분산하는 목적

• 복제본을 여러 브로커에 배치하여 장애 대응 능력을 확보하는 목적

• 디스크 사용량과 네트워크 부하의 불균형을 줄이는 목적

파티션 수뿐 아니라 리더의 분산과 복제본 배치를 함께 고려해야 하는 구조

9. 파일 관리(File Management)

Kafka가 로그를 여러 세그먼트로 나누어 생성하고 관리하는 방식이다

• 세그먼트가 일정 크기나 시간 조건을 충족하면 새로운 세그먼트를 생성하는 방식

• 로그 데이터를 세그먼트 단위로 관리하여 삭제와 컴팩션을 수행하는 방식

• retention.ms로 시간 기준 보존 정책을 설정하는 방식

• retention.bytes로 용량 기준 보존 정책을 설정하는 방식

보존 조건을 충족한 레코드가 즉시 개별 삭제되는 것이 아니라 세그먼트 관리 과정에서 정리되는 구조이다

10. 파일 형식과 인덱스

Kafka가 레코드를 배치 단위로 저장하고 필요한 위치를 빠르게 찾는다

주요 파일

• Record Batch에 여러 레코드를 묶어 저장 및 전송하는 구조

• 배치 단위 압축으로 저장 공간과 네트워크 사용량을 줄이는 방식

• 인덱스를 활용해 특정 오프셋 주변의 로그 위치를 빠르게 찾는 방식

Kafka의 인덱스는 관계형 데이터베이스의 B-tree 인덱스와 동일한 구조가 아니라 로그 위치 탐색에 특화된 인덱스 구조이다

11. 로그 컴팩션(Log Compaction)

동일한 키를 가진 레코드 중 오래된 데이터를 정리하여 키별 최신 상태를 유지하는 기능이다

역할

Offset 0 → Key A, Value 100
Offset 1 → Key B, Value 200
Offset 2 → Key A, Value 150
Offset 3 → Key B, Value 250

컴팩션 이후 유지하려는 논리적 상태

Key A → Value 150
Key B → Value 250

• 상품별 최신 가격이나 사용자별 최신 상태를 유지하는 데 활용하는 기능

• 백그라운드에서 비동기적으로 수행되는 기능

• 컴팩션이 완료되기 전까지 동일 키의 여러 버전이 남을 수 있는 구조

Tombstone

• 키의 값을 null로 기록하여 해당 키의 삭제를 표시하는 레코드

• 컴팩션 과정에서 해당 키의 이전 레코드가 제거될 수 있도록 하는 기능

• Tombstone 자체도 보존 기간이 지나면 제거될 수 있는 구조

컴팩션과 집계의 차이

• 컴팩션: 동일 키의 오래된 레코드를 정리하여 최신 상태를 유지하는 기능

• 집계: 여러 레코드를 합산하거나 평균 및 개수를 계산하는 연산

1. 컨슈머와 컨슈머 그룹

컨슈머란

• 토픽을 구독(subscribe)하고, 파티션에서 메시지를 읽어 검증,저장,변환 등의 처리를 하는 클라이언트(소비자)
• 프로듀서가 쓰는 속도가 컨슈머가 처리하는 속도보다 빠르면 컨슈머가 뒤처진다(랙 증가)
• 이때 파티션과 컨슈머를 늘려 수평 확장 할 수 있다

컨슈머 그룹

• 같은 group.id를 가진 컨슈머들의 집합
• 그룹 내 컨슈머들이 토픽 파티션을 나눠서 읽는다
• 한 파티션은 그룹 내 하나의 컨슈머에만 할당됩니다
• 한 컨슈머는 여러 파티션을 읽을 수 있습니다
• 컨슈머 수 > 파티션 수이면 초과분은 아무것도 읽지 못한다
  • 파티션 수가 그룹 병렬성의 상한
  • 컨슈머 수 < 파티션 수 의 경우는 하나의 컨슈머가 한 개 이상의 파티션을 담당하게 된다
• 컨슈머는 (비교적 느릴 수 있는) 외부 작업(DB 쓰기, 무거운 계산 등)을 하는 경우가 많으므로, 처리 시간이 긴 컨슈머일수록 그룹 확장이 중요할 수 있다
• 서로 다른 애플리케이션은 서로 다른 group.id를 사용 한다
  • 각 그룹이 독립적으로 토픽 전체 메시지를 읽으며, 한 그룹의 소비가 다른 그룹에 영향을 주지 않는다
    • 그룹 A(2개 컨슈머), 그룹 B(4개 컨슈머)가 같은 토픽을 각각 읽음

2. 컨슈머 그룹과 파티션 리밸런스

리밸런스(Rebalance): 컨슈머에서 다른 컨슈머로 파티션이 할당 되는 것

리밸런스가 일어나는 경우

• 컨슈머가 그룹에 추가될 때
  • 기존 컨슈머가 읽던 파티션 일부를 새 컨슈머가 가져감
• 컨슈머가 종료 또는 크래시할 때
  • 해당 컨슈머의 파티션이 남은 컨슈머에게 재할당됨
• 토픽에 파티션이 추가되는 등 토픽 메타데이터가 바뀔 때

리밸런스는 컨슈머 그룹의 고가용성과 확장성을 가능하게 하지만, 리벨런싱 중에는 지연이 발생할 수 있어 주의가 필요하다

조급한(Eager) 리밸런스

1. 모든 컨슈머가 읽기를 멈추고 자신의 파티션 할당을 해제 한다
2. 그룹에 다시 참여(rejoin)하고, 새로 파티션을 할당 한다
3. 이 동안 전체 그룹이 읽지 못하는 stop-the-world가 발생 한다
4. 컨슈머가 많을수록 이 중단 시간이 길어질 수 있다
5. 재할당 결과가 이전과 같더라도 모든 파티션을 일단 내려놓고 다시 받는다

협력적(Cooperative, Incremental) 리밸런스

1. 컨슈머 그룹 리더가 다른 컨슈머들에게 재할당이 필요한 파티션만 알려 준다
2. 해당 파티션을 가진 컨슈머만 그 파티션의 읽기를 멈추고 할당을 해제 한다
3. 할당 해제된 파티션은 새 컨슈머에게 할당 된다
4. 점진적으로 진행되며, 나머지 파티션은 계속 읽기가 가능해 전체 중단이 없습니다
5. 컨슈머가 많은 큰 그룹에서 유리하다

활성화 방법 쓸 것

리밸런스 동작 방식

• 컨슈머는 그룹 코디네이터(group coordinator) 역할을 하는 브로커에 하트비트(heartbeat) 를 보내 멤버십과 할당된 파티션 할당을 유지 한다
• 하트비트는 컨슈머의 백그라운드 스레드에서 전송되며, 일정 시간 하트비트가 없으면 코디네이터가 그 컨슈머를 죽은 것으로 간주하고 리밸런스 시작한다
• 컨슈머가 정상 종료(close())하면 코디네이터에 그룹 탈퇴를 알려, 세션 타임아웃을 기다리지 않고 즉시 리밸런스가 일어난다
• 파티션 할당 과정
  1. 컨슈머가 처음 그룹에 참여하려 할 때 코디네이터에 JoinGroup 요청
  2. 그룹에 처음 참여한 컨슈머가 그룹 리더가 된다
  3. 리더는 코디네이터로부터 그룹 멤버 목록을 받아 PartitionAssignor 구현체를 사용해 파티션 할당을 계산한다
  4. 리더가 할당 결과를 코디네이터에 전달하고, 코디네이터가 각 컨슈머에게 자기 할당 정보만 전달한다

  - 각 컨슈머는 자기 할당만 알고, 전체 할당은 리더만 알고 있다
  

  1. 이 과정은 리밸런스가 일어날 때마다 반복된다

정적 그룹 멤버십 (Static Group Membership)

• 기본적으로 컨슈머의 그룹 멤버십은 일시적이다 컨슈머가 그룹을 떠나면 파티션이 회수되고, 재참여하면 새 멤버 ID와 새 파티션이 할당된다
• group.instance.id를 지정하면 정적 멤버가 된다
  • 컨슈머가 종료되어도 session.timeout.ms 이내에 같은 group.instance.id로 다시 참여하면 리밸런스 없이 이전 파티션을 그대로 돌려받는다
  • 그룹 코디네이터가 그룹 상태를 캐시하므로, 다시 참여하는 컨슈머는 이전 할당 정보를 받는다
  • session.timeout.ms를 넘기면 리밸런스가 일어나고 파티션은 다른 컨슈머에게 넘어간다
• 유용한 경우: 각 컨슈머가 로컬 상태나 캐시를 유지하는 애플리케이션(재구축 비용이 큰 경우)
• 주의점
  • group.instance.id는 그룹 내에서 고유해야 한다 같은 ID로 두 컨슈머가 뜨면 나중에 시작한 쪽이 오류를 받는다
  • 컨슈머가 죽었는데 재시작이 늦으면, 그동안 그 파티션은 아무도 읽지 않아 지연이 커진다
  • 따라서 session.timeout.ms는 "재시작을 기다려줄 시간"과 "장애 감지 속도" 사이에서 균형을 맞춰 정한다

3. 컨슈머 생성과 토픽 구독

필수 속성

속성 설명

생성 예제

java

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092");
props.put("group.id", "CountryCounter");
props.put("key.deserializer",
    "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer",
    "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

토픽 구독

java

// 토픽 목록 구독
consumer.subscribe(Collections.singletonList("customerCountries"));

// 정규식 구독
consumer.subscribe(Pattern.compile("test.*"));

• 정규식 구독은 나중에 생성되는 토픽도 이름이 일치하면 자동으로 구독하고 리밸런스가 일어난다
• 여러 시스템의 데이터를 한 컨슈머가 읽을 때 유용하다
• 정규식 구독 시 컨슈머(또는 브로커 쪽)는 주기적으로 전체 토픽, 파티션 목록을 요청한다 토픽과 파티션이 매우 많으면 컨슈머 그룹이 많은 클러스터에서 브로커 네트워크 부하가 커질 수 있고, 클라이언트 쪽에서 목록을 필터링하는 비용도 있다
• 정규식 구독은 컨슈머에 토픽 목록 조회 권한(ACL의 describe 등)이 필요하다 특정 토픽에만 권한이 있는 환경에서는 정규식 구독이 실패할 수 있다

4. 폴링 루프

기본 구조

java

Duration timeout = Duration.ofMillis(100);

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(timeout);
        for (ConsumerRecord<String, String> record : records) {
            log.debug("topic = {}, partition = {}, offset = {}, customer = {}, country = {}",
                record.topic(), record.partition(), record.offset(),
                record.key(), record.value());
            // 처리 로직
        }
    }
} finally {
    consumer.close();
}

poll()이 하는 일

• 컨슈머 API의 핵심이다 처음 호출 시 다음을 수행한다
  • 그룹 코디네이터를 찾고, 컨슈머 그룹에 참여하고, 파티션을 할당받는다
• 이후 호출에서는 리밸런스, 하트비트 확인, 데이터 가져오기를 처리한다
• 반환 값은 ConsumerRecords로, 레코드마다 토픽, 파티션, 오프셋, 키, 값 정보를 담고 있다
• poll() 인자(timeout)는 컨슈머 버퍼에 데이터가 없을 때 블록할 최대 시간이다 0이면 즉시 반환하고, 데이터가 있으면 timeout과 관계없이 즉시 반환한다
• 하트비트는 별도 스레드가 보내지만, poll()을 오래 호출하지 않으면 max.poll.interval.ms 초과로 컨슈머가 그룹에서 제외된다
• 루프 안에서 오래 걸리는 처리(외부 호출, 대량 계산)를 하면 poll() 간격이 길어져 리밸런스가 발생할 수 있다 해결책
  • max.poll.records 감소
  • max.poll.interval.ms 증가
  • 처리를 별도 스레드로 분리

스레드 안전성

• 하나의 스레드에 하나의 컨슈머가 원칙이다
• 같은 그룹의 컨슈머 여러 개를 한 스레드에서 돌리거나, 한 컨슈머를 여러 스레드가 공유하는 것은 지원하지 않는다
  (KafkaConsumer는 스레드 안전하지 않으며, 동시에 접근하면 ConcurrentModificationException이 발생한다)
• 같은 애플리케이션에서 같은 그룹의 컨슈머를 여러 개 돌리려면 각각 별도 스레드에서 실행한다
  (ExecutorService에 컨슈머 하나씩 제출)
• wakeup()은 다른 스레드에서 호출해도 안전한 유일한 메서드
• 이벤트 처리 병렬화가 필요하면 컨슈머 하나가 poll()로 데이터를 가져오고, 처리는 스레드 풀에 넘기는 패턴이 있다 이 경우 커밋 시점과 처리 완료 시점을 맞추는 데 주의해야 한다
• 컨슈머 그룹을 스레드/프로세스에 걸쳐 확장하는 방법이 필요하면 Kafka Streams 등 프레임워크 사용도 고려한다

5. 컨슈머 설정

데이터 가져오기 관련

설정 기본값 설명

생존(그룹 멤버십) 관련

설정 기본값 설명

• session.timeout.ms를 낮추면: 죽은 컨슈머를 빨리 감지하지만, 처리 지연이나 GC pause 등으로 불필요한 리밸런스가 늘 수 있다
• session.timeout.ms를 높이면: 불필요한 리밸런스는 줄지만, 실제로 죽은 컨슈머 감지가 느려진다

그 외 설정

설정 기본값 설명

파티션 할당 전략 (partition.assignment.strategy)

전략 동작

• 커스텀 할당자는 PartitionAssignor 인터페이스를 구현하여 지정할 수 있다

6. 커밋과 오프셋

개념

• 카프카는 다른 JMS 계열 큐와 달리 컨슈머의 ack를 받지 않는다 대신 컨슈머가 파티션의 현재 읽은 위치(오프셋)를 기록한다
• 오프셋을 업데이트하는 동작을 커밋(commit)이라 하며, 컨슈머는 오프셋을 내부 토픽 __consumer_offsets에 메시지로 보낸다(파티션별로 커밋된 오프셋)
• 컨슈머가 크래시하거나 그룹에 새 컨슈머가 들어와 리밸런스가 일어나면, 각 컨슈머는 마지막으로 커밋된 오프셋부터 읽기를 이어간다
• 커밋 오프셋과 실제 처리 위치가 다르면 문제가 생긴다
  • 커밋된 오프셋이 클라이언트가 실제 처리한 마지막 오프셋보다 작으면 → 그 사이 메시지가 중복 처리된다
  • 커밋된 오프셋이 클라이언트가 실제 처리한 마지막 오프셋보다 크면 → 그 사이 메시지가 유실된다(처리되지 않은 채 건너뜀)
• 따라서 오프셋 관리 방식이 애플리케이션에 큰 영향을 준다

자동 커밋

• enable.auto.commit=true이면 auto.commit.interval.ms(기본 5초)마다 컨슈머가 poll()이 반환한 가장 마지막 오프셋을 커밋한다
• 커밋은 poll() 호출 안에서, 마지막 커밋 이후 interval이 지났는지 확인해 수행한다
• 장점: 간단하다
• 단점: 커밋 후 3초 뒤 크래시가 나면 리밸런스 후 3초 분량 메시지가 중복 처리된다 커밋 간격을 줄여도 중복 윈도만 줄일 뿐 없앨 수는 없다
• 주의: 다음 poll() 전에 반환된 모든 레코드 처리를 끝내야 한다(다음 poll 시점에 커밋되므로, 처리 중 예외가 나면 안전성이 깨질 수 있다) close() 시에도 자동 커밋된다

현재 오프셋 수동 커밋: commitSync()

java

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(timeout);
    for (ConsumerRecord<String, String> record : records) {
        // 처리
    }
    try {
        consumer.commitSync();
    } catch (CommitFailedException e) {
        log.error("commit failed", e);
    }
}

• enable.auto.commit=false로 설정한다
• 가장 최근 poll()이 반환한 마지막 오프셋을 커밋한다 따라서 poll()이 반환한 모든 레코드를 처리한 후 호출해야 한다(그렇지 않으면 유실 가능)
• 복구 불가능한 오류가 아니면 성공할 때까지 재시도한다 복구 불가능하면 예외를 던진다(CommitFailedException 등)
• 브로커 응답이 올 때까지 블록되므로 처리량이 떨어진다 커밋 빈도를 줄이면(배치 여러 개마다) 완화되지만 중복 가능성이 늘어난다

비동기 커밋: commitAsync()

java

consumer.commitAsync();

// 콜백 사용
consumer.commitAsync(new OffsetCommitCallback() {
    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {
        if (e != null)
            log.error("Commit failed for offsets {}", offsets, e);
    }
});

• 브로커 응답을 기다리지 않고 요청만 보내므로 처리량이 높다
• 재시도하지 않는다 재시도 시 순서 문제가 생기기 때문이다
  • 예: 오프셋 2000 커밋이 일시 실패 → 그동안 3000 커밋이 성공 → 2000이 재시도로 성공하면 커밋 오프셋이 뒤로 돌아가 중복 처리를 유발한다
• 재시도를 직접 구현하려면 단조 증가하는 시퀀스 번호를 두고, 콜백에서 재시도 시점에 시퀀스가 최신일 때만 재시도한다(이미 더 나중 커밋이 있다면 재시도하지 않는다) 다만 이 패턴은 복잡하므로 아래 조합 방식을 더 자주 쓴다
• 콜백은 다음 poll()이나 commitSync() 호출 중에 실행된다

동기 + 비동기 조합 (권장 패턴)

java

try {
    while (!closing) {
        ConsumerRecords<String, String> records = consumer.poll(timeout);
        for (ConsumerRecord<String, String> record : records) {
            // 처리
        }
        consumer.commitAsync();      // 평상시: 빠르게
    }
    consumer.commitSync();           // 종료 직전: 확실하게
} catch (Exception e) {
    log.error("Unexpected error", e);
} finally {
    consumer.close();
}

• 평상시의 일시적 커밋 실패는 다음 커밋이 대체하므로 무시해도 된다
• 하지만 마지막 커밋(종료, 리밸런스 직전)은 다음 커밋이 없으므로 commitSync()로 성공을 보장한다

특정 오프셋 커밋

• commitSync()/commitAsync()는 기본적으로 마지막 poll이 반환한 최신 오프셋을 커밋하므로, 큰 배치 처리 중간에 커밋하고 싶을 때는 오프셋 맵을 직접 전달한다

java

private Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
int count = 0;

// ... 루프 안에서 레코드 처리 후
currentOffsets.put(
    new TopicPartition(record.topic(), record.partition()),
    new OffsetAndMetadata(record.offset() + 1, "no metadata"));   // 다음에 읽을 오프셋
if (count % 1000 == 0)
    consumer.commitAsync(currentOffsets, null);
count++;

• 커밋하는 값은 마지막으로 처리한 오프셋 + 1(= 다음에 읽을 오프셋)이다
• 컨슈머가 여러 파티션을 읽는 경우 파티션별로 추적해야 한다

7. 리밸런스 리스너

컨슈머 API는 subscribe() 호출 시 ConsumerRebalanceListener를 전달해 리밸런스 전후에 코드를 실행할 수 있게 한다

메서드 호출 시점 용도

예제: 리밸런스 직전에 처리한 오프셋 커밋

java

private Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();

private class HandleRebalance implements ConsumerRebalanceListener {
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) { }

    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        System.out.println("Lost partitions in rebalance. Committing current offsets: " + currentOffsets);
        consumer.commitSync(currentOffsets);   // 여기서는 동기 커밋
    }
}

consumer.subscribe(topics, new HandleRebalance());

• 리밸런스 중에는 commitSync()로 확실하게 커밋해야 한다(비동기는 리밸런스가 끝나고 나서야 처리될 수 있다)
• 리스너 콜백은 컨슈머 스레드에서 poll() 도중에 호출되므로, 처리가 길어지면 다시 max.poll.interval.ms 문제가 생길 수 있다

8. 특정 오프셋의 레코드 읽기

poll()은 마지막 커밋 오프셋부터 읽지만, 다른 위치에서 읽고 싶을 때 다음 메서드를 사용한다

• seekToBeginning(partitions): 파티션의 맨 처음부터
• seekToEnd(partitions): 파티션의 맨 끝(최신)부터
• seek(partition, offset): 특정 오프셋부터
• offsetsForTimes(Map<TopicPartition, Long>): 특정 타임스탬프에 해당하는 오프셋을 조회 예를 들어 "한 시간 전부터 다시 처리"할 때 사용한다

java

Long oneHourEarlier = Instant.now().atZone(ZoneId.systemDefault()).minusHours(1).toEpochSecond();
Map<TopicPartition, Long> partitionTimestampMap = consumer.assignment()
    .stream()
    .collect(Collectors.toMap(tp -> tp, tp -> oneHourEarlier));
Map<TopicPartition, OffsetAndTimestamp> offsetMap = consumer.offsetsForTimes(partitionTimestampMap);

for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : offsetMap.entrySet()) {
    consumer.seek(entry.getKey(), entry.getValue().offset());
}

정확히 한 번(exactly-once) 처리

• 카프카 커밋 대신 결과와 오프셋을 같은 트랜잭션으로 외부 저장소(DB)에 저장하면 중복 없이 한 번만 처리하는 효과를 낼 수 있다

java

public class SaveOffsetsOnRebalance implements ConsumerRebalanceListener {
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        commitDBTransaction();                       // 결과+오프셋을 함께 커밋
    }
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        for (TopicPartition partition : partitions)
            consumer.seek(partition, getOffsetFromDB(partition));   // DB의 저장 위치로 이동
    }
}

consumer.subscribe(topics, new SaveOffsetsOnRebalance());
consumer.poll(Duration.ofMillis(0));                 // 파티션 할당 유도
for (TopicPartition partition : consumer.assignment())
    consumer.seek(partition, getOffsetFromDB(partition));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
        storeRecordInDB(record);
        storeOffsetInDB(record.topic(), record.partition(), record.offset());
    }
    commitDBTransaction();
}

• 재시작과 리밸런스 후에는 항상 DB에 저장된 오프셋으로 seek()해야 한다

9. 안전하게 종료하기

• 폴링 루프는 무한 루프이므로, 종료하려면 다른 스레드(예: shutdown hook)에서 consumer.wakeup()을 호출한다
  • wakeup()은 다른 스레드에서 호출해도 안전한 유일한 컨슈머 메서드다
  • 컨슈머가 poll()에서 대기 중이면 즉시 WakeupException을 던지며 빠져나온다 처리 중이었다면 다음 poll() 호출 시 예외가 발생한다
• 메인 스레드에서 WakeupException을 잡아 루프를 빠져나오며, 예외 자체는 별도 처리가 필요 없다
• 종료 전에 반드시 consumer.close()를 호출한다
  • 필요하면 오프셋을 커밋하고,
  • 그룹 코디네이터에 컨슈머 탈퇴를 알려 세션 타임아웃을 기다리지 않고 즉시 리밸런스가 일어나게 한다

java

Runtime.getRuntime().addShutdownHook(new Thread() {
    public void run() {
        System.out.println("Starting exit...");
        consumer.wakeup();
        try {
            mainThread.join();       // 메인 스레드가 정리를 마칠 때까지 대기
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
});

try {
    // ... 컨슈머 구독 및 폴링 루프
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(timeout);
        // 처리 및 commitAsync()
    }
} catch (WakeupException e) {
    // 무시: 종료 신호
} finally {
    consumer.commitSync();
    consumer.close();
    System.out.println("Closed consumer and we are done");
}

10. 역직렬화(Deserializer)

• 프로듀서의 시리얼라이저와 반드시 짝이 맞아야 한다
• 종류: StringDeserializer, IntegerDeserializer, ByteArrayDeserializer 등

Custom Deserializer

• Deserializer<T> 인터페이스(configure(), deserialize(), close())를 구현한다
• 단점
  • 프로듀서와 컨슈머의 직렬화 형식이 강하게 결합된다 필드가 추가, 변경되면 두 쪽 코드를 동시에 바꾸고 호환성을 신경 써야 한다(이전 데이터와 새 데이터가 한 토픽에 섞여 있을 때 특히 어렵다)
  • 팀 간에 같은 코드를 공유해야 하고 유지보수가 어렵다
• 따라서 일반적으로 JSON, Thrift, Protobuf, Avro 같은 범용 포맷과 직렬화 프레임워크를 쓴다

Avro Deserializer + 스키마 레지스트리

• Avro는 스키마 변경에 유연하다 스키마를 별도 저장소(스키마 레지스트리)에 두고, 메시지에는 스키마 ID만 담는다 컨슈머는 그 ID로 스키마를 조회해 역직렬화한다
• 스키마가 변경 되어도(하위/상위 호환 규칙 내에서) 컨슈머가 이전 데이터와 새 데이터를 모두 읽을 수 있다

java

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092");
props.put("group.id", "CountryCounter");
props.put("key.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
props.put("specific.avro.reader", "true");
props.put("schema.registry.url", schemaUrl);

String topic = "customerContacts";
KafkaConsumer<String, Customer> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList(topic));

while (true) {
    ConsumerRecords<String, Customer> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, Customer> record : records) {
        System.out.println("current customer name is: " + record.value().getName());
    }
    consumer.commitSync();
}

• schema.registry.url: 스키마 레지스트리 주소
• specific.avro.reader=true: 생성된 특정 클래스(Customer)로 역직렬화한다 지정하지 않으면 GenericRecord로 반환된다
• 스키마 레지스트리는 Avro 외에 Protobuf, JSON Schema도 지원한다

11. Standalone 컨슈머

컨슈머 그룹과 리밸런스 없이 특정 파티션만 직접 읽고 싶은 경우다

• 예: 특정 토픽의 특정 파티션을 읽는 프로세스가 항상 하나만 있으면 될 때

java

List<PartitionInfo> partitionInfos = consumer.partitionsFor("topic");

if (partitionInfos != null) {
    List<TopicPartition> partitions = new ArrayList<>();
    for (PartitionInfo partition : partitionInfos)
        partitions.add(new TopicPartition(partition.topic(), partition.partition()));
    consumer.assign(partitions);

    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> record : records) {
            // 처리
        }
        consumer.commitSync();
    }
}

• subscribe() 대신 assign()으로 파티션을 직접 지정한다 그룹 관리, 리밸런스가 없다
• subscribe()와 assign()은 함께 사용할 수 없다
• 오프셋 커밋은 여전히 가능하며, 커밋할 때는 group.id를 지정해 두는 것이 일반적이다
• 새 파티션이 추가되어도 자동으로 감지되지 않는다 주기적으로 partitionsFor()를 호출하거나, 파티션이 추가되면 컨슈머를 다시 시작해야 한다

카프카 프로듀서란?

프로듀서는 어플리케이션에서 생성한 레코드를 카프카 브로커로 전송하는 역할을 담당 한다

카프카 프로듀서 흐름

Produer Record

브로커로 보내기 위한 정보를 담은 객체

• 토픽
• 파티션 (option)
• 키 (option)
• 밸류

// 1. 파티션까지 직접 지정하는 경우
ProducerRecord<String, String> record = new ProducerRecord<>("토픽A", 0, "키값", "메시지내용");

// 토픽명 파티션 키 밸류
// 2. 파티션 없이 키만 지정 (파티셔너가 파티션 결정)
ProducerRecord<String, String> record = new ProducerRecord<>("토픽A", "키값", "메시지내용");

// 3. 파티션도 키도 없는 경우
ProducerRecord<String, String> record = new ProducerRecord<>("토픽A", "메시지내용");

카프카 프로듀서 메시지 전송 흐름

• ProducerRecord 생성 (토픽, 파티션, 키, 값 등 포함)
• 직렬화(Serializer): 키와 값을 바이트 배열로 변환
• 파티셔너(Partitioner): 어느 파티션으로 보낼지 결정 (키가 있으면 해시 기반, 없으면 라운드로빈 등)
• 레코드 배치(RecordAccumulator): 같은 파티션으로 가는 레코드를 배치로 묶음
• 전송 스레드가 배치를 브로커로 전송
• 성공 시 RecordMetadata 반환, 실패 시 예외 처리

프로듀서 생성

• bootstrap.servers: 브로커 목록
  • 브로커 동작 이슈 대비하여 최소 2개 이상 권장
• key.serializer, value.serializer: 직렬화 클래스 지정
  • ByteArraySerializer
  • StringSerializer
  • IntegerSerializer

메시지 전송 방식

파이어 앤 포켓

• 메시지를 전송한 후 결과를 확인 안함

동기적 전송

• 메시지를 전송한 후 반환 되는 Future 객체의 get 메서드를 호출하여 결과를 수신할 때 까지 대기하고 결과 수신 후 다음 처리를 진행

비동기적 전송

• 메시지를 전송 후 브로커로부터 응답을 수신 받을 시점에 callback 함수를 호출하여 별도 비즈니스 로직을 처리

프로듀서 설정

• client.id : 프로듀서와 애플리캐이션을 구분하기 위한 논리적 식별자
• acks: 메시지 전달 신뢰성 수준 (0, 1, all/-1)
  • 0 : 메시지가 성공적으로 전달되었다고 간주하고 브로커의 응답을 기다리지 않음
  • 1 : 리더 레플리카가 메시지를 받는 순간 브로커로부터 성공했다는 응답을 받음
  • all/-1 : 메시지가 모든 인-싱크 레플리카에 전달된 뒤에 브로커로부터 성공했다는 응답을 받음
• buffer.memory: 전송 대기 중인 메시지를 위한 버퍼 크기
• compression.type: 압축 방식 (snappy, gzip, lz4, zstd)
• retries / retry.backoff.ms: 재시도 관련 설정
• batch.size, linger.ms: 배치 크기와 대기 시간 (처리량 vs 지연시간 트레이드오프)
• max.in.flight.requests.per.connection: 순서 보장과 관련된 설정
• max.request.size: 최대 요청 크기

delivery.timeout.ms >= linger.ms + retry.backoff.ms + request.timeout.ms

max.block.ms : send를 호출 했을 때, 버퍼가 꽉 차있거나 메타데이터를 아직 못 가져오는 경우 최대 대기 시간
linger.ms : 메시지를 바로 보내지 않고, 배치로 묶기 위해 추가로 기다리는 시간
retry.backoff.ms : 전송이 실패했을 때 재시도하기 전 대기하는 시간
request.timeout.ms : 실제로 브로커에 요청을 보낸 후, 브로커의 응답을 얼마나 기다릴지 정하는 시간

시리얼라이저

기본 시리얼라이저

커스텀 시리얼라이저

Serializer인터페이스를 구현하여 Custom 시리얼라이저를 만들 수 있다

public class Customer {
    private int customerID;
    private String customerName;
    // getter, setter 생략
}

public class CustomerSerializer implements Serializer<Customer> {
    @Override
    public byte[] serialize(String topic, Customer data) {
        // customerID(4바이트) + customerName 길이(4바이트) + customerName(바이트)
        // 형태로 직접 바이트 배열을 조립
        ...
    }
}

커스텀 시리얼라지어의 문제점

• 스키마 변경에 취약함
  • 예: Customer 객체에 email 필드를 나중에 추가하면, 새 버전으로 보낸 메시지를 예전 버전의 컨슈머(디시리얼라이저)가 읽으면 파싱이 깨지거나 에러 발생
  • 프로듀서와 컨슈머 간의 스키마 호환성을 스스로 관리해야 하는 부담이 생김
• 유지보수 어려움
  • 필드가 추가/삭제될 때마다 시리얼라이저와 디시리얼라이저 코드를 양쪽 다 수정해야 함
  • 여러 팀, 여러 서비스가 같은 토픽을 사용하는 경우 동기화가 매우 어려워짐
• 재사용성 부족
  • 객체마다 매번 새로운 시리얼라이저 작성 필요

Avro

아파치에서 만든 직렬화 시스템
데이터 구조를 스키마로 별도 정의하고, 스키마를 기반으로 직렬화/역직렬화를 수행

1. 스키마와 데이터 분리
   • Avro는 데이터를 쓸 때 사용한 스키마와 읽을 때 사용하는 스키마가 완전히 같지 않아도 호환되도록 설계
2. 스키마 진화(Schema Evolution) 지원
   • 필드 추가/삭제 시에도 하위 호환성(backward compatibility) 또는 상위 호환성(forward compatibility)을 유지 가능
3. 스키마 레지스트리(Schema Registry) 연동
   • 실무에서는 보통 Avro를 Confluent Schema Registry와 함께 사용
   • 동작 방식:
     1. 프로듀서가 메시지를 보낼 때, 스키마를 레지스트리에 등록 (또는 이미 등록된 스키마 ID 조회)
     2. 메시지 안에는 스키마 전체가 아니라 스키마 ID만 포함해서 전송 → 메시지 크기 절약
     3. 컨슈머는 메시지에 포함된 스키마 ID로 레지스트리에서 해당 스키마를 조회해 역직렬화
   • 이 구조 덕분에 프로듀서와 컨슈머가 서로 다른 버전의 스키마를 사용해도 자동으로 호환성이 체크 보장

예시

[프로듀서]
Customer 객체 → Avro 스키마 기반 직렬화 → 바이트 + 스키마ID
        ↓
   Schema Registry에 스키마 등록/조회
        ↓
   카프카 브로커로 전송

[컨슈머]
바이트 + 스키마ID 수신
        ↓
   Schema Registry에서 스키마ID로 스키마 조회
        ↓
Avro 역직렬화 → Customer 객체 복원

파티션

파티셔너(Partitioner)의 역할

프로듀서가 메시지를 보낼 때, 어느 토픽의 몇 번 파티션에 저장할지 결정하는 컴포넌트

시리얼라이저 (바이트 변환)
      ↓
파티셔너 ← 여기서 파티션 결정
      ↓
배치 큐 (해당 파티션으로)

• 같은 파티션 내에서는 메시지 순서가 보장

키가 있는 경우 - 해시 기반 파티셔닝

동작 방식

• 기본 파티셔너(DefaultPartitioner)는 키의 해시값을 파티션 개수로 나눈 나머지를 이용해 파티션 결정

partition = hash(key) % 파티션개수

특징

• 같은 키는 항상 같은 파티션으로 감 → 이게 핵심 포인트
• 예: 키가 "user123"인 메시지는 항상 같은 파티션으로 전송됨
• 컨슈머 입장에서 봤을 때 같은 키를 가진 메시지들끼리는 순서가 보장됨

  주의할 점

• 파티션 개수가 변경되면 해시 결과가 달라짐 → 기존에 매핑되던 파티션과 달라질 수 있음
• 그래서 토픽의 파티션 수는 운영 중 함부로 늘리지 않는 것이 권장됨 (특히 키 기반 파티셔닝을 사용할 때)

키가 없는 경우 - 라운드로빈 / Sticky 방식

키가 없을 때의 동작

• 키가 없으면 해시 계산을 할 대상이 없으므로, 파티션에 고르게 분산시키는 방식 사용

Sticky 방식으로 바뀌었나

• 순수 라운드로빈은 매번 파티션을 바꾸다 보니, 배치가 작은 상태로 여러 파티션에 흩어져 전송되는 비효율 발생
  • 배치 효율 저하 → 처리량 저하
• Sticky Partitioner는 같은 파티션에 메시지를 모아 배치 크기를 키워서 전송 효율(처리량)을 높이는 것이 목적
• 장기적으로 보면 전체 파티션에 고르게 분산되는 건 동일하되, 배치 단위로 묶어서 보내는 효율성 개선

커스텀 파티셔너 (Custom Partitioner)

기본 해시 기반 방식으로는 해결이 안 되는 비즈니스 로직상의 특수한 분배 요구가 있을 때 사용

• Partitioner 인터페이스 구현

헤터, 인터셉터, 쿼터

헤더 (Headers)

카프카 메시지는 키와 밸류 외에 헤더(Header)라는 별도의 키-값 쌍 목록을 가질 수 있다.
메시지 본문(밸류)을 건드리지 않고 부가적인 메타데이터를 함께 전달하고 싶을 때 사용한다.

예시

ProducerRecord<String, String> record = 
    new ProducerRecord<>("토픽A", "키값", "메시지내용");

record.headers().add("traceId", "abc-123".getBytes(StandardCharsets.UTF_8));
record.headers().add("source", "order-service".getBytes(StandardCharsets.UTF_8));

producer.send(record);

인터셉터 (ProducerInterceptor)

메시지가 전송되기 직전 또는 전송 완료(성공/실패) 직후에 공통 로직을 자동으로 끼워 넣을 수 있는 훅(hook)이다.
애플리케이션 코드를 수정하지 않고도 프로듀서 동작에 공통 처리를 추가할 수 있다.

• ProducerInterceptor<K, V> 인터페이스 구현
  • onSend: 시리얼라이저, 파티셔너를 거치기 전 send를 호출한 직후
  • onAcknowledgement: 브로커로부터 응답(성공/실패)을 받은 직후
  • close: 프로듀서 종료 시

public class CountingProducerInterceptor implements ProducerInterceptor<String, String> {

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        // 전송 전: 헤더에 타임스탬프 추가
        record.headers().add("sentAt", 
            String.valueOf(System.currentTimeMillis()).getBytes());
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        // 전송 후: 성공/실패 카운트
        if (exception != null) {
            System.out.println("전송 실패: " + exception.getMessage());
        } else {
            System.out.println("전송 성공, 오프셋: " + metadata.offset());
        }
    }

    @Override
    public void close() {}

    @Override
    public void configure(Map<String, ?> configs) {}
}

쿼터 (Quotas)

카프카 브로커가 특정 클라이언트(프로듀서/컨슈머)의 처리량을 제한하는 기능
하나의 클라이언트가 브로커 자원을 독점해서 다른 클라이언트에게 영향을 주는 것을 방지하기 위한 장치

쿼터 종류

• 프로듀서 쿼터: 특정 클라이언트가 브로커에 쓸 수 있는 초당 바이트 수 제한
• 컨슈머 쿼터: 특정 클라이언트가 브로커에서 읽을 수 있는 초당 바이트 수 제한
• 요청 쿼터: 브로커 요청 처리 시간(CPU 사용률) 기준 제한

동작 방식

• 클라이언트가 할당된 쿼터를 초과하면, 브로커는 응답을 지연(throttle)시켜서 클라이언트가 자연스럽게 속도를 줄이도록 유도
• 요청을 거부하는 게 아니라 응답 지연을 통해 속도 조절

필요 이유

• 특정 프로듀서가 갑자기 대량의 메시지를 쏟아내면, 브로커의 네트워크/디스크 자원을 독점하게 되어 다른 프로듀서, 컨슈머의 성능이 저하될 수 있음
• 이를 막기 위해 브로커 관리자가 클라이언트별로 초당 처리 가능한 바이트 수(byte rate)를 제한

해시 클러스터에서의 해시 몰림

• 1 ~ 100, 101 ~ 200, 201 ~ 300 키 값이 한 구간에 몰리면 하나의 서버에만 부하가 받게 된다.
• 클러스터 내에 해시 서버가 추가 또는 삭제되는 경우 리밸런싱 작업이 전체적으로 일어나면서 데이터의 이동과 변경이 발생하게 되고 그 과정에서 사용성이 떨어질 수 있다

해시 클러스터 내에서 안정적인 해시 운영을 위한 방법으로 다양한 방법이 있지만 안정해시라는 개념을 통해 해시 클러스터 내에 해시 데이터를 안정적으로 관리한다.

안정 해시의 컨셉

• 안정 해시의 컨셉은 링이다
• 링에 의해 한방향으로 가까운 노드에 해시 값을 저장 또는 삭제한다
• 노드가 추가되면 링 위에 새로운 노드가 위치하고 한방향으로 가까운 노드만 리밸런싱을 하게 되어 다른 노드들은 영향을 받지 않게 된다

안정 해시의 문제점

• 안정 해시도 결국 쏠림 현상이 발생할 수 있다
• 이벤트성 트래픽에서는 특정 데이터에 몰릴 수 있는 문제가 생길 수 있다

안정 해시 쏠림 현상 개선

• 가상 노드를 통해 링 위에 가상 노드를 다른 노드들 사이에 배치한다
• 가상 노드의 수를 늘릴 수 록 노드 하나가 담당하게 되는 파티션의 크기가 작아지기 때문에 데이터가 균등하게 분포하게 된다

설계

• CAP : AP(가용성 + 분단 허용성)
  • 일관성을 유지함에 따라 분단 허용성을 지킬 수 없음
  • 결과적 일관성을 통한 일관성 유지

1. 링형 아키텍처 적용(양 끝을 연결한다는 것으로 논리적으로 링형)
   • 해시 키에 인접한 노드를 찾아감
2. Virtual Node 적용
   • 가상 노드를 링 위에 인접하지 않도록 배치하여 적재되는 데이터의 분포를 균등하게 함
3. 적재되는 데이터를 인접 노드에 함께 저장하여 장애 내성을 갖춤
4. 해시키 범위 별 참여하고 있는 Lookup 테이블을 관리하는 방식 적용
5. 데이터 저장시 저장 성공을 위한 성공 Node 수 관리

아키텍처 구성

처리 흐름

get

sequenceDiagram
    autonumber
    participant Client as 클라이언트
    participant Coord as LB
    participant Table as Lookup 테이블
    participant NodeA as Node A 
    participant NodeB as Node B 
    participant NodeC as Node C 

    Client->>Coord: GET(Key)
    activate Coord

    Coord->>Table: hash 값이 속한 담당 범위 조회
    activate Table

    Table-->>Coord: Primary = NodeA<br/>Replica Set = [NodeB, NodeC] (링 상 다음 N-1개 노드)
    deactivate Table

    Coord->>NodeA: GET(Key) 요청 [Primary]
    activate NodeA

    alt NodeA 정상 응답
        NodeA-->>Coord: Value 반환
    else NodeA 장애 / 타임아웃
        deactivate NodeA
        Coord->>NodeB: GET(Key) 요청 [인접 Node]
        activate NodeB
        alt NodeB 정상 응답
            NodeB-->>Coord: Value 반환
        else NodeB도 장애
            deactivate NodeB
            Coord->>NodeC: GET(Key) 요청 [다음 인접 Node]
            activate NodeC
            NodeC-->>Coord: Value 반환
            deactivate NodeC
        end
    end

    Coord-->>Client: Key : Value 응답
    deactivate Coord

put

sequenceDiagram
    autonumber
    participant Client as 클라이언트
    participant Coord as LB
    participant Table as Lookup 테이블
    participant NodeA as Node A
    participant NodeB as Node B
    participant NodeC as Node C

    Client->>Coord: PUT(Key, Value)
    activate Coord

    Coord->>Table: hash 값이 속한 담당 범위 조회
    activate Table
    Table-->>Coord: NodeA<br/>인접 노드 = [NodeB, NodeC] (N-1개 노드)
    deactivate Table

    Coord->>NodeA: PUT(Key, Value) 저장 요청 [Primary]
    activate NodeA
    NodeA-->>Coord: ACK (저장 성공)
    deactivate NodeA

    Coord-->>Client: PUT 성공 응답
    deactivate Coord

    par 인접 노드 비동기 복제
        Coord->>NodeB: PUT(Key, Value) 복제 [인접 Node]
        activate NodeB
        NodeB-->>Coord: ACK (복제 완료)
        deactivate NodeB
    and
        Coord->>NodeC: PUT(Key, Value) 복제 [다음 인접 Node]
        activate NodeC
        NodeC-->>Coord: ACK (복제 완료)
        deactivate NodeC
    end

인접 노드에 대한 복제 시 비동기 워커를 통한 복제
이때 몇개의 노드에 복제되었을 때 저장이 완료되게 할지에 대한 설정 필요

Cluster에 서버 추가

sequenceDiagram
    autonumber
    participant New as 신규 서버
    participant Coord as LB
    participant Table as Lookup 테이블
    participant Succ as 기존 인접 서버 (승계자)

    New->>Coord: 클러스터 참여 요청 (Join)
    activate Coord

    Coord->>Coord: 신규 서버의 가상 서버 해시 위치 계산

    Coord->>Table: 가상 서버 위치를 링에 삽입
    activate Table
    Table->>Table: 삽입 위치 기준 시계방향으로<br/>가장 가까운 기존 가상 서버 탐색<br/>→ 분할될 범위 산정
    Table-->>Coord: 이관 대상 범위 + 기존 담당 서버 반환
    deactivate Table

    Coord->>Succ: 분할 범위에 해당하는 Key 이관 요청
    activate Succ
    Succ->>New: 해당 범위 Key:Value 복사
    activate New
    New-->>Succ: 이관 완료 ACK
    deactivate New
    Succ-->>Coord: 마이그레이션 완료 보고

    Coord->>Table: Lookup 테이블 갱신
    activate Table
    Table-->>Coord: 갱신 완료
    deactivate Table

    Coord->>Succ: 이관 완료된 Key 삭제
    Succ-->>Coord: 삭제 완료
    deactivate Succ

    Coord-->>New: 서버 추가 완료 응답
    deactivate Coord

Cluster에 서버 삭제

sequenceDiagram
    autonumber
    participant Leave as 제거 대상 서버
    participant Coord as LB
    participant Table as Lookup 테이블
    participant Succ as 인접 서버 (승계자)

    Leave->>Coord: 클러스터 제거 요청
    activate Coord

    Coord->>Table: 이탈 서버의 가상 서버 위치 조회
    activate Table
    Table->>Table: 링 상에서 시계방향으로<br/>가장 가까운 다음 가상 서버 탐색<br/>→ 승계할 서버 결정
    Table-->>Coord: 이관 대상 범위 + 승계 서버 반환
    deactivate Table

    Coord->>Leave: 보유 중인 Key 전체 이관 요청
    activate Leave
    Leave->>Succ: 담당 범위 Key:Value 복사
    activate Succ
    Succ-->>Leave: 이관 완료 ACK
    deactivate Succ
    Leave-->>Coord: 마이그레이션 완료 보고
    deactivate Leave

    Coord->>Table: Lookup 테이블 갱신
    activate Table
    Table-->>Coord: 갱신 완료
    deactivate Table

    Coord-->>Leave: 클러스터 제외 및 종료 승인
    deactivate Coord

이 책은 애자일에 대한 책으로 가치있는 것을 빠르게 보여주고 피드백 받으면서 목표를 달성하기 위한 수단과 방법을 소개하는 책이다.

크게 계획하고 나중에 한 번에 보여주는 것이 아니라, 작게 쪼개서 지금 보여주는 것.

자주 틀리고 빠르게 수정하는 것이 한 번에 완벽하게 맞추려는 것보다 훨씬 낫다는 이야기다.

이 책을 보고 든 이미지는 아래와 같다.

출발점에서 목표까지 일직선으로 달리는 것이 아니라, 방향을 확인하며 조금씩 꺾어가는 화살표이다.

중요한 건 목표에 가까워지고 있느냐다.

우리가 애자일한 방법을 추구해야하는 이유는 아래의 워터폴의 문제를 들여다 보면 공감할 수 있다.

처음에 완벽하게 계획했다고 생각했던 것이 끝에 다다를 즘에 틀렸다는 걸 알게 된다.

하지만 그때는 이미 되돌아가기엔 너무 늦는다.

물론 애자일이 항상 옳은 것은 아니다.

요구사항이 처음부터 명확하게 고정된 프로젝트라면 굳이 애자일일 필요가 없을 것이다.

하지만 대부분의 소프트웨어 개발은 그렇지 않다.

요구사항은 바뀌고, 사용자가 원하는 것도 써봐야 안다.

그 불확실한 환경에서 피드백 루프를 통해 방향을 조정할 수 있다는 것, 그것이 애자일의 분명한 강점이다.

애자일에 대한 AI의 영향에 대한 생각

애자일의 핵심은 단순하다. 가치를 빠르고 자주 전달하는 것이다.

짧은 주기로 만들어서 보여주고, 피드백을 받아 방향을 수정한다.

그리고 이 사이클을 빠르게 돌릴수록 좋다.

AI는 이 사이클을 실질적으로 빠르게 만들어준다.

반복적인 코드 작성, 테스트 생성, 버그 수정에 개발자가 시간을 많이 쏟던 영역들은

AI에 의해 더욱 빠르게 처리된다.

결과적으로 더 자주 배포할 수 있고, 피드백을 더 빨리 받을 수 있고, 방향을 더 일찍 수정할 수 있다.

애자일이 원하는 사이클이 AI로 인해 비로소 완성되어가고 있는 모습이다.

애자일의 조력자 AI

AI가 애자일에 가져다주는 것들은 결국 하나의 방향으로 수렴한다.

사람이 해야 할 일에 더 집중할 수 있게 해준다는 것.

애자일에서 사람만이 할 수 있는 일이 있다.

무엇을 만들지 판단하는 것, 가치를 정의하는 것, 피드백을 해석하는 것, 팀이 같은 방향을 바라보게 하는 것.

이것들은 AI가 대신할 수 없다.

그리고 아이러니하게도, AI가 반복 작업을 빠르게 처리해줄수록 사람은 이 일에 더 많은 시간을 쏟을 수 있다.

"가치를 빠르고 자주 전달하는 것" 빠른 것은 수단이고 가치는 목적이다.

AI는 그 수단을 강력하게 만들어줬다.

경계해야할 것은?

AI가 빠르게 실행해줄수록 속도 자체가 목적이 되는 순간이 온다.

단순하게 빠르게 라는 것에만 매몰되면 애자일이 말하는 가치는 옅어지게 된다.

빠른 것은 수단이고 가치는 목적이다.

AI는 그 수단을 강력하게 만들어줬다.

목적을 잊지 않는 건 여전히 사람의 몫이다.

그렇다면 우리는 무엇을 해야 하는가

AI 시대에 개발자에게 필요한 것은 더 빠르게 쓰는 능력이 아니다.

멈추고 생각하는 능력, 가치를 판단하는 능력, 방향을 잡는 능력.

아이러니하게도 AI가 발전할수록 이 능력의 가치는 더 높아진다.

AI가 빠르게 달려줄수록, 어디로 달릴지 결정하는 사람의 역할은 더 중요해진다.

그것이 AI 시대에 애자일이 작동하는 방식일 것이다.

[References]

• https://ronjeffries.com/articles/nature

Spring Batch에서 JpaPagingItemReader 같은 Paging 기반 Reader는 처음에는 꽤 합리적인 선택처럼 보인다.

한 번에 많은 데이터를 메모리에 올리지 않아도 되고, 일정 크기로 잘라 읽는 방식이어서 대량 처리에도 잘 어울려 보이기 때문이다.

그런데 실제 운영 환경에서는 이런 현상을 자주 만난다.

• 처음 몇 페이지는 빠르다.
• 중간 페이지부터 점점 느려진다.
• 뒤 페이지로 갈수록 같은 page size인데도 처리 시간이 더 길어진다.
• 전체 데이터가 늘수록 배치 시간이 비정상적으로 증가한다.

이러한 현상의 핵심 원인은 대부분 offset 기반 pagination의 구조적 비용에 있다.

문제는 "읽는 건수"보다 "버리는 건 수"에 있다

많은 사람이 Paging Reader를 이렇게 이해한다.

한 번에 1000건씩 읽으니까, 매 페이지 비용도 비슷하겠지.

겉으로 보면 맞는 말처럼 보인다.

실제로 애플리케이션이 받아오는 건수는 매번 limit 1000으로 동일하다.

하지만 DB가 그 1000건을 어떻게 찾아서 반환하는지까지 보면 이야기가 달라진다.

예를 들어 아래 같은 쿼리를 생각해보자.

select id, amount, created_at
from payment
where status = 'READY'
order by id asc
limit 1000 offset 0;

첫 페이지는 비교적 단순하다. 앞에서 버릴 데이터가 없기 때문이다.

그런데 500번째 페이지를 읽으면 쿼리는 이렇게 바뀐다.

select id, amount, created_at
from payment
where status = 'READY'
order by id asc
limit 1000 offset 499000;

마지막 구간에 가까워지면 이런 식이 된다.

select id, amount, created_at
from payment
where status = 'READY'
order by id asc
limit 1000 offset 900000;

애플리케이션은 여전히 1000건만 받는다.

하지만 DB는 그 1000건을 주기 전에 앞의 900000건을 지나가야 할 수도 있다.

즉, Paging Reader가 뒤 페이지로 갈수록 느려지는 이유는 더 많이 가져와서가 아니라, 더 많이 버려야 하기 때문이다.

Paging Reader는 실제로 어떻게 동작하는가

Spring Batch의 Paging Reader는 개념적으로 다음 흐름으로 동작한다.

1. page 0 조회
2. page 1 조회
3. page 2 조회
4. ...
5. 마지막 page까지 반복

예를 들어 page size가 1000이면 내부적으로는 이런 조회가 이어진다.

-- 1 page
limit 1000 offset 0

-- 2 page
limit 1000 offset 1000

-- 3 page
limit 1000 offset 2000

-- 4 page
limit 1000 offset 3000

이 구조는 웹 화면에서 몇 페이지 정도 넘겨보는 정도라면 큰 문제가 아닐 수 있다.

하지만 배치는 다르다.

배치는 보통 첫 페이지부터 마지막 페이지까지 전부 읽는다.

즉, 큰 offset이 "가끔" 등장하는 것이 아니라, 언젠가 반드시 등장한다.

그래서 배치에서 Paging Reader의 성능 문제는 예외 상황이 아니라 구조적으로 누적되는 문제다.

offset 기반 pagination의 비용 모델

offset pagination의 핵심은 다음 두 숫자다.

• limit: 최종적으로 반환할 건수
• offset: 반환 전에 건너뛸 건수

많은 경우 limit는 작고 고정되어 있다. 예를 들어 1000건이다. 반면 offset은 페이지가 뒤로 갈수록 계속 커진다.

이때 비용을 지배하는 것은 대개 limit보다 offset이다.

예를 들어 총 100만 건이 있고, page size가 1000이라고 해보자.

• 1페이지: offset 0
• 100페이지: offset 99000
• 500페이지: offset 499000
• 1000페이지: offset 999000

반환 건수는 매번 1000건으로 같지만, DB가 도달해야 하는 위치는 점점 멀어진다.

이 때문에 뒤 페이지는 앞 페이지보다 느릴 가능성이 높다.

각 페이지의 반환 건수는 같지만, 각 페이지의 도달 비용은 같지 않다.

DB는 offset을 보고 "순간이동"하지 못한다

Paging을 사용하는 경우 보통 이렇게 생각을 하게 된다.

offset 900000이면 DB가 그냥 900001번째부터 바로 읽으면 되는 것 아닌가?

이렇게 생각하기 쉽지만, 실제 DB는 그렇게 단순하게 움직이지 않는다.

DB는 쿼리 조건과 정렬 조건을 만족하는 결과 집합을 만들고, 그 결과 중 앞의 offset개를 건너뛴 뒤 limit개를 반환해야 한다.

즉, offset 900000 limit 1000은 논리적으로 다음과 같은 의미에 가깝다.

1. 조건에 맞는 후보를 찾는다.
2. 정렬 기준에 맞춰 순서를 정한다.
3. 앞의 900000건을 지난다.
4. 그 다음 1000건을 반환한다.

DB 엔진과 실행 계획에 따라 세부 동작은 다르지만, 핵심은 같다.

큰 offset은 "바로 건너뛰는 숫자"가 아니라, 시스템이 감당해야 하는 작업량의 일부다.

정렬이 들어가면 왜 더 부담스러워질 수 있을까

Paging 쿼리는 거의 항상 order by와 함께 사용된다.

그런데 이 정렬은 생각보다 비용에 큰 영향을 준다.

예를 들어 아래 쿼리를 보자.

select id, amount, created_at
from payment
where status = 'READY'
order by created_at asc
limit 1000 offset 900000;

여기서 DB는 단순히 900001번째 행을 찾는 것이 아니라:

• status = 'READY' 조건을 만족하는 대상을 찾고
• created_at asc 순서를 맞추고
• 그 순서에서 앞의 900000건을 건너뛴 뒤
• 다음 1000건을 반환해야 한다

만약 where 조건과 order by를 함께 잘 받쳐주는 인덱스가 없다면, 부담은 더 커진다.

• 더 많은 row 접근이 필요할 수 있다.
• 정렬을 위한 추가 작업이 필요할 수 있다.
• 임시 정렬 영역이나 filesort 성격의 비용이 발생할 수 있다.

뒤 페이지 성능 저하는 단순히 offset 숫자만의 문제가 아니라, 조건 + 정렬 + offset이 결합되며 커지는 경우가 많다.

인덱스가 있어도 왜 문제가 완전히 사라지지 않을까

이 주제를 쓰면 반드시 나오는 질문이 있다.

인덱스만 잘 잡으면 해결되는 것 아닌가?

인덱스는 분명 중요하다.

그리고 실제로 성능을 상당히 개선해줄 수 있다.

하지만 여기서 중요한 것은 인덱스가 offset pagination을 근본적으로 바꾸지는 못한다는 점이다.

인덱스가 잘 잡혀 있으면 다음이 좋아질 수 있다.

• 조건 필터링 비용 감소
• 정렬 비용 감소
• Full scan 회피

하지만 인덱스가 있어도 큰 offset에서는 여전히 이런 부담이 남는다.

• 인덱스 상에서 많은 엔트리를 따라가야 할 수 있다.
• 원하는 위치에 도달하기 전까지 여러 row를 지나가야 한다.
• 뒤 페이지일수록 누적 탐색 비용이 커진다.

인덱스는 offset paging을 덜 나쁘게 만들 수는 있어도, 뒤 페이지로 갈수록 느려지는 구조 자체를 없애지는 못한다.

배치에서 이 문제가 더 위험한 이유

웹 서비스의 목록 조회라면 사용자가 1페이지나 2페이지 정도만 보고 떠나는 경우가 많다.

심지어 100페이지 이후를 실제로 보는 사용자는 거의 없을 수도 있다.

하지만 배치는 다르다.

• 첫 페이지부터 마지막 페이지까지 모두 읽는다.
• 중간 페이지에서 멈추지 않는다.
• 매일, 혹은 주기적으로 반복된다.
• 데이터가 늘수록 마지막 offset도 같이 커진다.

웹에서는 offset 문제가 부분적으로 드러날 수 있지만, 배치에서는 그 비용이 항상 끝까지 누적된다.

예를 들어 오늘 10만 건이던 테이블이 몇 달 뒤 500만 건이 되면, page size가 그대로여도 뒤쪽 페이지 비용은 훨씬 무거워진다.

그래서 초기에는 별문제가 없어 보이던 배치가 운영 몇 달 후 갑자기 느려지는 경우가 생긴다.

실무에서는 이 패턴이 꽤 흔하다.

처음에는 충분히 빨랐는데, 데이터가 쌓이자 어느 날부터 새벽 배치가 윈도우 안에 끝나지 않는다.

이때 원인은 코드 복잡도보다도, 대개 offset 기반 순회 구조의 누적 비용에 있는 경우가 많다.

숫자로 보면 왜 후반부가 더 느린지 더 직관적이다

총 100만 건, page size 1000이라고 해보자.

각 페이지는 아래 정도의 skip 비용을 가진다.

• 1페이지: 0건 skip
• 10페이지: 9000건 skip
• 100페이지: 99000건 skip
• 500페이지: 499000건 skip
• 1000페이지: 999000건 skip

마지막 페이지는 1000건을 받기 위해 앞의 거의 100만 건에 가까운 위치를 지나가야 할 수 있다.

이 관점에서 보면 page size가 같다는 사실은 큰 의미가 없다.

중요한 것은 최종적으로 몇 건을 반환하느냐가 아니라, 거기 도달하기 위해 얼마나 많은 데이터를 지나가야 하느냐다.

Spring Batch 코드로 보면 왜 더 헷갈리기 쉬운가

Spring Batch 코드만 보면 Paging Reader는 아주 단순해 보인다.

@Bean
fun paymentPagingReader(entityManagerFactory: EntityManagerFactory): JpaPagingItemReader<Payment> {
    val reader = JpaPagingItemReader<Payment>()
    reader.setName("paymentPagingReader")
    reader.setEntityManagerFactory(entityManagerFactory)
    reader.setPageSize(1000)
    reader.setQueryString(
        """
        select p
        from Payment p
        where p.status = :status
        order by p.id asc
        """.trimIndent()
    )
    reader.setParameterValues(
        mapOf("status" to PaymentStatus.READY)
    )
    return reader
}

애플리케이션 코드 관점에서는 그저 "1000건씩 잘 읽는 Reader"로 보인다.

그래서 많은 사람이 page size, chunk size, thread 수 같은 애플리케이션 설정만 튜닝하려고 한다.

하지만 실제 병목은 Reader 바깥, 즉 DB가 뒤 페이지에 도달하기 위해 치르는 offset 비용에 있을 수 있다.

그래서 이 문제는 Spring Batch 설정값만 바꿔서는 해결되지 않는 경우가 많다.

page size를 키우면 해결될까

이것도 자주 시도하는 대응이다.

• 1000건이 느리면 5000건으로 키워본다.
• 5000건이 느리면 10000건으로 키워본다.

물론 page 수가 줄어들기 때문에 전체 쿼리 횟수는 감소할 수 있다.

그래서 일정 수준의 개선은 있을 수 있다.

왜냐하면 page size를 키워도 여전히 offset 기반이면, 뒤 페이지는 여전히 큰 offset을 가진다.

단지 페이지 개수가 줄어들 뿐이다.

게다가 page size를 너무 키우면 다음 문제가 생긴다.

• 한 번에 읽는 row 수 증가
• 메모리 사용량 증가
• 처리 시간 증가
• 트랜잭션 부담 증가

page size 조정은 튜닝 포인트일 수는 있어도, 뒤 페이지 성능 저하의 근본 원인을 없애지는 못한다.

해결 방향: offset이 아니라 기준점으로 읽는다

이 문제를 근본적으로 줄이려면 페이지 번호를 기준으로 읽는 대신, 마지막으로 읽은 기준점을 기준으로 다음 데이터를 가져오는 방식이 더 적합하다.

흔히 keyset pagination 또는 seek 방식이라고 부른다.

예를 들어 아래 같은 쿼리다.

select id, amount, created_at
from payment
where status = 'READY'
  and id > :lastId
order by id asc
limit 1000;

이 방식은 offset 900000처럼 큰 숫자를 가지고 앞 데이터를 버리지 않는다.

대신 마지막으로 읽은 id 다음부터 바로 이어서 읽는다.

• offset 방식: 몇 번째부터 읽을지 계산한다.
• keyset 방식: 어디서부터 이어 읽을지 지정한다.

배치에서는 후자가 훨씬 잘 맞는다.

배치는 순차적으로 끝까지 읽는 작업이기 때문이다.

Kotlin 예시: key 기준으로 다음 페이지 읽기

fun findNextPage(lastId: Long, pageSize: Int): List<Payment> {
    return entityManager.createQuery(
        """
        select p
        from Payment p
        where p.status = :status
          and p.id > :lastId
        order by p.id asc
        """.trimIndent(),
        Payment::class.java
    )
        .setParameter("status", PaymentStatus.READY)
        .setParameter("lastId", lastId)
        .setMaxResults(pageSize)
        .resultList
}

이 방식의 장점은 분명하다.

• 뒤 페이지로 갈수록 offset이 커지지 않는다.
• 마지막 기준점 이후만 읽기 때문에 순차 처리에 적합하다.
• 대량 데이터에서 후반부 성능 저하를 줄이기 쉽다.

물론 이 방식도 정렬 기준과 인덱스 설계가 중요하다.

하지만 적어도 "앞의 수십만 건을 버리고 다음 1000건을 읽는" 구조는 피할 수 있다.

마무리

Paging Reader가 뒤 페이지로 갈수록 느려지는 이유는 단순히 데이터가 많아서가 아니다.

offset 기반 조회가 뒤쪽으로 갈수록 더 많은 row를 건너뛰어야 하는 구조이기 때문이다.

하지만 Paging Reader를 사용하는 것 자체가 문제는 아니다.

데이터 양이 크지 않고, 조회 범위가 제한적이며, 뒤 페이지까지 깊게 순회하지 않는 상황이라면 Paging 방식은 여전히 단순하고 실용적인 선택이 될 수 있다.

결국 중요한 것은 Reader의 이름이 아니라 데이터의 양, 조회 패턴, 정렬 방식, 그리고 배치가 끝까지 어떤 방식으로 순회하는가다.

데이터가 계속 커지고 끝 페이지까지 모두 읽어야 하는 배치라면 offset 기반 Paging이 점점 불리해질 수 있고, 반대로 특정 구간만 가볍게 읽는 작업이라면 충분히 적합할 수도 있다.

배치에서는 "익숙한 Reader를 쓰는가"보다 현재 데이터 규모와 운영 상황에 맞는 Reader를 선택하고 있는가를 먼저 봐야 한다.

Paging Reader도 그 선택지 중 하나일 뿐이고, 상황에 따라서는 Cursor Reader나 keyset 방식이 더 적합할 수 있다.