코딩 보관소

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. 컨슈머와 컨슈머 그룹

컨슈머란

• 토픽을 구독(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 방식이 더 적합할 수 있다.

들어가며

개발을 하다보면 "정렬된 상태를 유지해야 하는" 상황을 꽤 자주 마주친다.

이런 문제를 처음 마주쳤을 때 "DB에서 ORDER BY 하면 되지 않나?"라는 생각이 들게 되는데,

트레픽이 많은 경우에 매번 테이블을 조회하여 정렬된 데이터를 가져오는 것은 부담이 될 수 있다.

Redis의 Sorted Set(ZSet)을 활용하면 이러한 문제에서 조금 자유로울 수 있다.

데이터를 넣는 순간부터 항상 정렬된 상태를 유지하여 따로 정렬할 필요가 없다.

ZSet이란 무엇인가

ZSet은 각 멤버(member)에 score라는 숫자 값을 부여하고, 항상 score 기준으로 정렬된 상태를 유지하는 자료구조다. 일반 Set에 "순서"라는 차원이 하나 더 추가된 것이라고 봐도 된다.

ZSet이 항상 보장하는 두 가지 성질이 있다.

유일성 — 같은 멤버는 하나만 존재한다. 이미 있는 멤버를 추가하면 score만 갱신

정렬 — 멤버는 score 오름차순으로 정렬된다. score가 동일하면 멤버 이름을 사전순으로 비교해 순서를 결정

단순해 보이지만, 이 두 성질을 삽입, 삭제가 발생할 때 마다 이를 위지하느 것은 꽤 부담이되어 보인다.

어떻게 이를 해결했을까?

내부 자료구조

ZSet의 성능 비결은 내부에서 두 가지 자료구조를 동시에 유지한다는 데 있다. 그 전에, 데이터 크기에 따라 구조 자체가 달라진다는 점을 먼저 이해하면 좋다.

인코딩 전략 — 작을 때와 클 때

ZSet은 데이터 규모에 따라 내부의 방식을 자동 전환한다.

멤버 수 ≤ 128  AND  각 멤버 크기 ≤ 64 bytes
  → listpack  (연속된 메모리 블록, 오버헤드 최소화)

그 외
  → skiplist + hashtable  (빠른 탐색과 조회를 위한 구조)

데이터가 적을 때는 단순한 연속 메모리 구조인 listpack을 써서 메모리를 절약한다.

그러다 임계값을 넘는 순간 자동으로 skiplist 기반 구조로 전환된다.

이 임계값은 zset-max-listpack-entries와 zset-max-listpack-value 설정으로 조정할 수 있다.

실제로 운영 환경에서 ZSet을 쓰다 보면 어느 시점부터 메모리 사용량이 갑자기 늘어나는 것처럼 보이는 경우

대부분 이 인코딩 전환이 일어난 시점이다.

Skip List — 정렬과 범위 조회 담당

임계값을 넘어서면 ZSet은 Skip List로 전환된다.

Skip List는 연결 리스트에 "지름길" 역할을 하는 포인터를 여러 레벨로 추가한 자료구조다.

레벨 4 │ ──────────────────────────────────── [100]
레벨 3 │ ─────────── [30] ────────────────── [100]
레벨 2 │ ──── [10] ─ [30] ──── [60] ───────── [100]
레벨 1 │ [5]─[10]─[20]─[30]─[40]─[60]─[80]─[100]
                       ↑ score 오름차순

탐색할 때는 레벨이 높은 곳에서 시작해 목표 범위를 빠르게 좁힌다.

노드의 레벨은 삽입 시 확률적으로 결정되는데, Redis는 레벨업 확률로 25%를 사용한다.

이 덕분에 균형 트리(AVL, Red-Black Tree)와 비슷한 O(log N) 성능을 내면서도 구현이 훨씬 단순하다.

Skip List는 범위 기반 탐색에 특히 강하다.

"score가 100 이상 500 이하인 멤버 전부"를 가져오는 ZRANGEBYSCORE 같은 명령이 빠른 이유다.

연결 리스트 특성상 범위 내 다음 노드를 순서대로 따라가면 되기 때문이다.

span 필드

Redis의 Skip List는 각 포인터에 span이라는 값을 저장한다.

span은 그 포인터가 건너뛰는 노드 수를 의미한다. 포인터를 따라 이동하면서 span 값을 누적하면,

별도의 계산 없이 탐색과 동시에 rank(순위)를 구할 수 있다.

이 덕분에 ZRANK나 ZREVRANK가 O(log N)에 동작할 수 있다.

Hash Table — score 단건 조회 담당

Skip List는 범위 조회에 강하지만, "alice의 score가 정확히 얼마냐"는 질문에는 O(log N)이 걸린다.

전체 리스트를 탐색해야 하기 때문이다.

이걸 해결하기 위해 Redis는 member → score 매핑을 담은 Hash Table을 병행해서 유지한다.

Hash Table 덕분에 ZSCORE 명령은 O(1)에 답을 준다.

Hash Table │ { "alice": 1500.0, "bob": 1320.0, "carol": 1750.0 }
Skip List  │ bob(1320) → alice(1500) → carol(1750)

두 구조는 쓰기(삽입·삭제·갱신)가 일어날 때마다 함께 업데이트 된다.

쓰기 비용이 조금 늘어나지만, 읽기 패턴이 다양한 ZSet의 특성에 적합한 트레이드오프로 볼 수 있다.

결과적으로

범위 조회는 Skip List가, score 단건 조회는 Hash Table이 담당한다.

/* check if the element is too large or the list
 * becomes too long *before* executing zzlInsert. */
if (zzlLength(zobj->ptr)+1 > server.zset_max_listpack_entries ||
    sdslen(ele) > server.zset_max_listpack_value ||
    !lpSafeToAdd(zobj->ptr, sdslen(ele)))
{
    zsetConvertAndExpand(zobj, OBJ_ENCODING_SKIPLIST, zsetLength(zobj) + 1);
} else {
    zobj->ptr = zzlInsert(zobj->ptr,ele,score);
    if (newscore) *newscore = score;
    *out_flags |= ZADD_OUT_ADDED;
    return 1;
}

위 소스는 실제 레디스 소스를 일부 가져온 것인데, 특정 임계 치를 넘어가게 되면 Zset의 구조를 전환하는 코드를 확인할 수 있다(zsetConvertAndExpand)

마치며

ZSet은 정렬된 데이터를 빠르게 가져올 때 쉽게 사용하는데, 그 안에는 Skip List와 Hash Table을 역할에 따라 나눠 쓰는 꽤 정교한 설계가 들어 있다.

동작 구조를 살펴보니 해당 자료구조에 대한 이해도를 높일 수 있었고, 이를 통해 개발이나 운영 때 이러한 전환 구조에 따른 변화를 고려하고 참고 할 수 있을 것 같다.

참고

1) https://redis.io/docs/latest/develop/data-types/sorted-sets

2) https://github.com/redis/redis/blob/unstable/src/t_zset.c

대기열은 왜 필요한가?

트래픽이 갑자기 몰리면 서버는 쉽게 무너진다.

티케팅, 선착순 쿠폰, 오픈런 이벤트 — 공통점은 하나다. 평소보다 몇십 배의 요청이 짧은 시간에 집중된다는 것이다.

이 문제를 해결하는 방법이 대기열(Waiting Queue)이다.

서버는 슈퍼맨이 아니다

Spring Boot 서버가 요청을 받으면 Tomcat 스레드 풀에서 스레드 하나를 꺼내 처리한다.

기본 설정은 최대 200개. 즉, 동시에 201번째 요청이 들어오는 순간부터는 앞선 요청이 끝나기를 기다려야 한다.

DB도 마찬가지다.

커넥션 풀(HikariCP)은 기본 최대 10개의 커넥션을 유지한다.

11번째 요청이 DB를 쓰려는 순간, 커넥션이 반환될 때까지 블로킹된다.

평소에는 요청 하나가 수십 ms 안에 끝나기 때문에 스레드와 커넥션이 빠르게 반환된다.

그런데 트래픽이 몇십 배로 폭증하면 이야기는 달라진다.

대기열이 없으면 어떤 일이 생기나

스레드 풀이 가득 찬다

요청이 폭발적으로 들어오면 200개의 Tomcat 스레드가 모두 점유된다.

이후 요청은 Tomcat 내부 accept queue에서 대기하다가, 그마저 넘치면 Connection refused로 거절된다.

[요청 폭주]
→ Tomcat 스레드 200개 전부 점유
→ 새 요청: 대기 or Connection refused
→ 응답 시간 급증

커넥션 풀이 고갈된다

스레드들이 각자 DB 커넥션을 잡고 처리 중이다.

커넥션이 10개인데 스레드가 200개면, 190개의 스레드는 커넥션을 기다리며 블로킹된다.

기다리는 시간이 길어질수록 스레드는 더 오래 묶이고, 새 요청은 더 오래 기다리는 악순환이 시작된다.

[DB 커넥션 대기]
→ 스레드가 커넥션을 못 잡고 블로킹
→ 스레드 점유 시간 증가
→ 결국 전체 스레드 고착 (thread starvation)

타임아웃 폭탄이 터진다

HikariCP의 기본 connectionTimeout은 30초다. 30초 동안 커넥션을 못 받으면 예외를 던진다.

수백 개의 요청이 동시에 30초를 기다렸다가 동시에 실패하면, 클라이언트의 재시도가 또 다른 폭주를 만든다.

상황은 오히려 더 악화된다.

연쇄 장애(Cascade Failure)로 번진다

문제는 한 서버에서 끝나지 않는다. MSA 환경에서는 서비스들이 서로를 호출한다.

주문 서비스가 느려지면, 연관된 다른 서비스도 영향을 받아 연쇄 장애로 번질 수 있다.

대기열이 하는 일

트래픽을 흡수한다

순간 폭주하는 요청을 일단 줄에 넣어두고, 서버가 소화할 수 있는 속도로 천천히 내보낸다.

서버 입장에서는 항상 일정한 속도의 요청만 처리하면 된다.

많은 인원이 동시에 몰려도, 서버는 정해진 인원만 처리한다. 나머지는 줄에서 기다린다.

순서를 보장한다

먼저 온 사람이 먼저 처리된다.

도착 시각을 기준으로 줄이 세워지기 때문에, 대기열에 진입한 후로는 집인 순서에 따라 공정하게 처리된다.

대기열 없이 수만 건의 요청이 동시에 DB에 도달하면 처리 순서는 사실상 랜덤이다.

먼저 클릭했어도 나중에 처리될 수 있다.

시스템을 지킨다

서버가 감당할 수 있는 범위를 초과하는 요청은 즉시 처리하지 않는다.

무리하게 받다가 전체가 다운되는 것보다, 정해진 속도로 처리하고 나머지는 기다리게 하는 것이 낫다.

대기열이 유입량 자체를 조절하기 때문에 스레드 풀, 커넥션 풀 등 자원의 고갈로 인한 문제를 예발 할 수 있다.

대기열의 흐름

사용자 입장에서는 이렇게 느껴진다.

1. 버튼 클릭 → "현재 347번째입니다. 예상 대기 시간 2초"
2. 주기적으로 순위 확인 → 숫자가 줄어든다
3. 내 차례 → 입장권(토큰) 발급
4. 입장권을 들고 → 실제 서비스 이용
5. 처리 완료 → 입장권 자동 소멸

입장권에는 유효 시간이 있다. 일정 시간 안에 사용하지 않으면 만료된다.

줄을 서다가 자리를 비운 사람이 자원을 점유하지 않도록 하기 위해서다.

대기열 X vs 대기열 O

마무리

대기열은 불편한 것이 아니라 서비스를 지키는 안전장치다.

대기열 없이 트래픽 폭주를 그냥 받으면 서비스 자체가 다운된다.

반면 대기열이 있으면 늦더라도 "내 차례는 온다"는 신뢰를 줄 수 있다.

병목은 "동시에 너무 많이 들어오는 것"에서 시작된다.

대기열은 그 입구를 조절하는 기술이다.