Kafka 재시도가 정상 파티션까지 멈출 때
목차
소비자 프로세스가 살아 있고 브로커와도 연결돼 있는데 정상 문서 처리가 진행되지 않는다. 잘못된 메시지가 있는 파티션만 느릴 것으로 예상했지만, 다른 파티션의 메시지도 기다린다. 두 파티션이 같은 실행 슬롯을 쓰고 실패한 작업이 재시도하는 동안 그 슬롯을 붙잡고 있다면 가능한 현상이다.
Kafka의 파티션은 메시지가 순서대로 쌓이는 로그다. 각 메시지의 파티션 내부 위치를 **오프셋(offset)**이라고 한다. 여기서는 처리가 반복해서 실패하는 파티션 0과 정상 메시지를 가진 파티션 1을 가정한다. 먼저 실행 자리가 막히는 원인을 보고, 그 자리를 분리한 뒤에도 남는 완료 기록과 오프셋 커밋 문제를 따라간다. 브로커를 띄운 성능 실험이 아니라 소비자 안의 상태와 실행 순서를 고정한 모델이다.
파티션을 나누어도 자원은 공유될 수 있다
파티션이 분리돼 있다는 사실은 소비자 애플리케이션의 스레드 풀·세마포어·메모리까지 분리한다는 뜻이 아니다. 실행 슬롯 두 개를 먼저 온 작업에 주는 소비자를 생각해 보자. 파티션 0의 오프셋 10과 11에 있는 메시지가 둘 다 실패한 채 슬롯을 보유하면, 파티션 1의 오프셋 20에 있는 메시지는 시작할 수 없다.
실행 슬롯 1: 파티션 0의 오프셋 10을 재시도
실행 슬롯 2: 파티션 0의 오프셋 11을 재시도
실행 대기: 파티션 1의 오프셋 20재시도 간격을 1초에서 10초로 늘려도 이 구조에서 정상 파티션의 작업은 여전히 기다린다. 대기 시간을 늘린 것과 실행 자리를 반환한 것은 다르다. 재시도를 예약한 뒤 비싼 실행 슬롯을 반환할 수는 있지만, 미완료 메시지와 재시도 시각을 보관하는 책임은 남는다.
같은 코드와 입력으로 절대 성공하지 않는 실패는 반복 실행 전에 식별한다. 필수 필드 누락과 일시적인 저장소 연결 오류를 구분하는 이유다. 다만 미래 스키마를 지원하지 못한 입력은 코드 배포 뒤에는 처리할 수 있으므로, 지금의 반복 시도로 바뀌는 조건과 향후 재처리 조건을 나누어 기록한다.
파티션별 제한이 바꾸는 것을 작게 실험한다
실행 파일을 python3 partition_frontier.py로 실행하면 슬롯 배정, 안전한 커밋 위치, 옛 할당 결과의 무시를 순서대로 확인한다. 슬롯 배정의 입력에서 각 쌍은 (파티션 번호, 오프셋)이다. 예를 들어 (0, 10)은 파티션 0의 오프셋 10에 있는 메시지를 뜻한다.
배정 함수의 전체 구현은 다음과 같다. 전역 실행 슬롯은 두 개이고, 파티션별 제한에 걸린 후보는 건너뛰어 다른 파티션의 후보를 확인한다.
from collections import Counter
def admit(records, per_partition_limit=None):
running = []
counts = Counter()
for partition, offset in records:
if len(running) == 2:
break
if per_partition_limit is not None and counts[partition] >= per_partition_limit:
continue
running.append((partition, offset))
counts[partition] += 1
return running
records = [(0, 10), (0, 11), (1, 20)]
shared = admit(records)
isolated = admit(records, per_partition_limit=1)
assert shared == [(0, 10), (0, 11)]
assert isolated == [(0, 10), (1, 20)]
print("shared slots:", shared)
print("per-partition admission:", isolated)shared slots: [(0, 10), (0, 11)]
per-partition admission: [(0, 10), (1, 20)]한 파티션이 동시에 차지할 수 있는 슬롯을 하나로 줄이면 이 입력에서는 정상 파티션의 메시지가 들어갈 자리가 남는다. 이것만으로 일반적인 공정성이 증명되지는 않는다. 모델은 정상 파티션의 메시지가 이미 후보 목록에 있다는 가정이다. 실제 소비자는 다른 파티션의 후보도 계속 보게 하고, 한 파티션의 대기열만 끝없이 검사하지 않는 스케줄링이 필요하다.
아직 실행하지 않은 파티션 0의 오프셋 11에 있는 메시지를 버려도 된다는 뜻은 아니다. 이미 가져온 레코드는 제한된 버퍼에 유지하거나 다시 가져올 수 있는 위치를 보존해야 한다. 특정 파티션의 가져오기를 pause()하는 방법은 앞으로 반환될 레코드를 제어한다. 이미 poll()로 받은 메시지 본문의 메모리를 되돌려 주지는 않는다. Kafka Java 4.1 클라이언트는 pause와 resume으로 할당된 파티션의 후속 조회를 제어한다. KafkaConsumer의 consumption flow control
실행 완료 순서와 재시작 위치는 다르다
실행 슬롯을 분리해도 커밋을 잘못하면 처리하지 않은 메시지를 건너뛴다. Kafka의 커밋 값은 해당 파티션에서 다음에 읽을 위치다. 병렬 작업의 가장 큰 완료 offset에 1을 더하는 규칙은 앞의 작업이 덜 끝났을 때 안전하지 않다.
파티션 0에서 실제로 전달받은 오프셋이 [10, 12, 15]라고 하자. 레코드 사이에 빈 offset이 있을 수 있으므로 숫자가 연속인지보다 전달받은 순서가 중요하다.
| 완료 집합 | 건너뛰면 안 되는 작업 | 보수적으로 커밋할 위치 |
|---|---|---|
{12, 15} | 10 | 기존 위치 10 유지 |
{10, 15} | 12 | 11 |
{10, 12, 15} | 없음 | 16 |
두 번째 행의 11은 레코드 11을 처리했다는 뜻이 아니다. 10까지 완료했으며 그다음부터 다시 가져와도 된다는 위치다. 다음 유효 레코드가 12라면 재시작 후 12를 받게 된다. 마지막 행에서만 16까지 진행할 수 있다.
def frontier(delivered, done, committed):
result = committed
for offset in delivered:
if offset not in done:
break
result = offset + 1
return result이 함수는 한 파티션의 현재 할당에서 아직 커밋하지 않은 전달 목록을 순서대로 받는 계산 모델이다. 실제 구현은 커밋이 확정된 앞부분을 제거하고 이미 확정한 위치를 후퇴시키지 않는다. 모든 worker가 각자 commit을 호출하게 하지 않고, 완료 통지를 받은 한 소유자가 파티션별 진행을 계산하면 순서와 상태를 관리하기 쉽다. Java 소비자의 스레드 안전성이나 재할당 콜백은 사용하는 클라이언트 계약에 맞춰야 한다. KafkaConsumer의 수동 커밋과 멀티스레드 처리 설명
끝난 작업도 메모리를 붙잡는다
같은 파티션에서 오프셋 10의 메시지는 계속 실패하고, 뒤의 오프셋 12와 15의 메시지는 처리를 마쳤다고 하자. 뒤의 두 작업은 실행 슬롯을 반환했지만, 완료 기록은 앞의 작업이 끝날 때까지 보관해야 한다. 뒤쪽 작업을 계속 받아들이면 실행 중 작업 수는 작아도 완료 대기 기록이 계속 늘어난다.
따라서 최소한 세 종류의 자원을 따로 센다. 실제 실행 수, 실행·재시도·커밋 대기를 합친 미확정 기록 수, 그 기록이 보유하는 메시지 본문 바이트다. 단순화한 메모리 모델은 다음과 같다.
소비자 보유 메모리 ≈ 가져온 메시지 본문 + 실행 중 임시 메모리 + 완료 대기 결과문제가 생긴 파티션에 완료 대기 기록이 충분히 쌓이면 그 파티션에서 메시지를 가져오는 속도를 늦춘다. 그동안 다른 파티션의 처리는 계속 진행할 수 있어야 한다. 파티션별 한도만 두면 할당된 파티션 수에 비례해 합계가 커지므로 전역 예산도 둔다. 이때 전역 예산이 이미 한 파티션으로 가득 찼다면 다른 파티션의 여유를 보장할 수 없다. 파티션별 최대 점유량과 전체 스케줄링을 함께 정해야 하는 이유다.
재시도 대기 중에도 메시지를 가져오는 소비자 루프 자체를 긴 sleep으로 멈추는 방식은 별도 문제를 만든다. poll() 호출 간격과 소비자 그룹의 생존 조건을 지켜야 하므로 재시도 시각은 작업별 상태로 관리하고, 실제 호출 간격이 사용하는 소비자 설정을 만족하는지 관측한다. 모델 예제는 생존 신호(heartbeat)나 그룹 재할당을 검증하지 않는다.
재할당 후 옛 worker의 완료가 돌아온다면
파티션 0을 맡았을 때 할당 세대를 7로 기록했다고 하자. 오래 걸린 worker가 돌아오기 전에 그 파티션의 담당에서 해제됐다가 다시 맡으면 세대가 8일 수 있다. 이때 7의 완료 통지를 현재 처리의 완료로 합치면 새 소유권의 진행 위치를 잘못 바꾼다.
모델은 결과에 (파티션, offset, 할당 세대)를 묶고 현재 세대와 다르면 진행 상태에 반영하지 않는다. 검사 결과는 old assignment result: ignored다. 이것은 애플리케이션의 오래된 완료 통지 방어를 보여 줄 뿐, Kafka 그룹 프로토콜이나 커밋 요청 자체의 안전성을 대신하지 않는다.
외부 저장소에 이미 수행한 쓰기도 이 비교로 사라지지는 않는다. 파티션 소유권을 잃은 worker가 늦게 저장할 수 있으므로 저장소의 멱등성·버전 조건이 필요하다. 완료 통지를 무시하는 일과 외부 효과를 차단하는 일은 다른 경계다.
잘못된 입력을 넘기는 순간에도 완료 계약이 필요하다
복구할 수 없는 입력에서 파티션을 멈추면 순서를 보존하지만 뒤의 정상 메시지가 기다린다. 실패 메시지를 별도로 보관하는 **DLQ(Dead Letter Queue)**로 넘기고 진행하면 원래 순서대로 업무 결과가 생긴다는 보장은 포기한다. 어느 선택이든 원본을 보존하고 운영자가 어떻게 재처리할지 결정할 수 있어야 한다.
DLQ에 기록하기 전에 원본 offset을 커밋하면 그 사이 종료됐을 때 메시지가 사라진다. 기록 뒤 커밋 전에 종료되면 DLQ에 중복될 수 있다. 두 기록이 Kafka 안에 있다면 DLQ 쓰기와 소비 offset 변경을 Kafka 트랜잭션으로 묶는 선택지가 있다. 별도 DB나 객체 저장소를 DLQ로 쓰면 같은 트랜잭션 보장이 자동으로 이어지지 않는다. Kafka의 delivery semantics와 transactions
앞에서 완료된 구간까지만 커밋하는 계산은 무엇을 “끝났다”고 인정할지 결정한 뒤에야 의미가 있다. 정상 결과가 저장됐거나, 선택한 격리 정책에 따라 실패 기록이 내구성 있게 확정된 상태여야 한다. 로그를 한 줄 남겼다는 이유로 완료 집합에 넣으면, 커밋 계산은 정확해도 데이터 보존 계약은 깨진다.
새 소비자를 검증할 때는 한 파티션의 잘못된 입력을 계속 실패시키면서 다른 파티션을 진행시켜 본다. 그다음 문제가 생긴 파티션의 뒤쪽 완료 순서를 뒤집고, 마지막으로 저장 직후와 커밋 직전에 중단을 넣는다. 이 세 조건을 분리하면 실행 공정성, 안전한 재시작 위치, 외부 효과의 중복을 각각 확인할 수 있다.