이번 시간에는 Kafka에서 메시지가 언제 어떤 단위로 지워지는지와 보관 기간보다 늦은 컨슈머가 어떻게 되는지 확인해 보며 배워가는 시간입니다.
예를 들어서 컨슈머가 메시지를 다 읽었는데도 토픽 크기가 줄지 않는다고 해볼게요. 반대로 며칠 멈춰 있던 컨슈머를 다시 띄웠더니, 읽지 않은 메시지 일부가 처리되지 않고 지나가기도 해요.
큐에서는 컨슈머가 메시지를 가져가면 그 메시지가 사라지죠. 그런데 Kafka의 토픽은 로그라서, 컨슈머가 읽어도 메시지가 그대로 남아요. 지우는 일은 토픽의 보관 정책이 따로 맡습니다.
실습 준비
이 글의 명령은 모두 실습 코드의 kafka-retention-offset/ 디렉터리에서 실행합니다.
compose.yaml: Kafka 4.3 단일 노드 컨테이너 정의
Docker로 브로커와 컨트롤러를 겸하는 Kafka 4.3 한 대를 띄울게요(compose.yaml). 모든 명령은 컨테이너 안의 Kafka 도구로 실행합니다.
services:
kafka:
image: apache/kafka:4.3.1
environment:
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,EXTERNAL://localhost:9441
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS: 5000
KAFKA_HEAP_OPTS: -Xmx512m -Xms256m
ports:
- "9441:9094"
KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS는 브로커가 지울 세그먼트를 찾는 주기(log.retention.check.interval.ms)예요.
기본값은 5분이라서 실습에서 삭제를 바로 보기 어려워요. 그래서 5초로 줄였습니다. 이 값은 토픽 설정에는 없고 브로커 설정으로만 바꿀 수 있어요.
컨테이너를 띄우고, 자주 쓰는 명령을 셸 함수로 만들어 둘게요. 아래 블록은 파일로 만들지 말고 실습 디렉터리에서 연 터미널에 그대로 붙여 넣어 주세요. 이후 명령은 모두 이 터미널에서 실행하고, 터미널을 새로 열면 다시 붙여 넣으시면 됩니다.
docker compose up -d
kafka() { docker compose exec -T kafka /opt/kafka/bin/"$@"; }
offsets() { echo "start=$(kafka kafka-get-offsets.sh --bootstrap-server kafka:9092 --topic "$1" --time earliest | cut -d: -f3) end=$(kafka kafka-get-offsets.sh --bootstrap-server kafka:9092 --topic "$1" --time latest | cut -d: -f3)"; }
segs() { docker compose exec -T kafka ls /tmp/kafka-logs/"$1" | grep '\.log' | tr '\n' ' '; echo; }
describe() { kafka kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe "$@" 2>/dev/null | awk 'NF {print $1, $3, $4, $5, $6}' | column -t; }
offsets: 파티션 0의 가장 앞 offset(start)과 다음 메시지가 들어갈 offset(end)을 출력합니다.start를 log start offset,end를 log end offset이라고 부릅니다.segs: 파티션 디렉터리에 있는 세그먼트 파일 목록을 출력합니다.describe: 컨슈머 그룹의 commit 위치를 필요한 열만 출력합니다.
실습 1: 읽어도 메시지는 그대로 남습니다
파티션 1개짜리 orders 토픽에 메시지 5개를 넣고 offset 범위를 볼게요.
kafka kafka-topics.sh --bootstrap-server kafka:9092 --create --topic orders --partitions 1
seq 1 5 | sed 's/^/주문/' | kafka kafka-console-producer.sh --bootstrap-server kafka:9092 --topic orders
offsets orders
Created topic orders.
start=0 end=5
billing 그룹으로 5개를 모두 읽은 뒤 다시 봅니다.
kafka kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic orders --group billing --from-beginning --max-messages 5 2>/dev/null
offsets orders
주문1
주문2
주문3
주문4
주문5
start=0 end=5
다 읽었는데도 범위는 그대로 0부터 5까지예요. 컨슈머는 메시지를 가져갈 뿐, 지울 권한이 없거든요.
실습 2: 시간이 지나면 세그먼트 단위로 지워집니다
메시지는 토픽의 보관 정책에 걸렸을 때 지워지고, 컨슈머가 읽은 시점과는 관계가 없어요. 관련 토픽 설정은 이렇습니다.
retention.ms: 기본값 604800000(7일). 이 시간이 지난 데이터를 지웁니다.retention.bytes: 기본값 -1(제한 없음). 파티션 하나의 크기가 이 값을 넘으면 오래된 데이터부터 지웁니다.segment.bytes: 기본값 1073741824(1GB). 파티션을 이 크기의 세그먼트 파일로 나눠 씁니다.segment.ms: 기본값 604800000(7일). 세그먼트가 이 크기에 닿지 않아도 이 시간이 지나면 새 세그먼트로 넘어갑니다.
실제로 지우는 단위는 메시지 하나가 아니라 세그먼트 파일이에요. 세그먼트 파일 이름은 그 세그먼트의 첫 offset입니다.
보관 기간을 60초, 세그먼트 전환 시간을 10초로 줄인 events 토픽을 만들어 볼게요.
kafka kafka-topics.sh --bootstrap-server kafka:9092 --create --topic events --partitions 1 --config retention.ms=60000 --config segment.ms=10000
Created topic events.
메시지 5개씩 세 번, 12초 간격으로 보냅니다. 앞 세그먼트가 10초를 넘긴 뒤 새 메시지가 들어오니까, 세그먼트가 세 개로 나뉘어요.
for b in 0 1 2; do
seq $((b*5)) $((b*5+4)) | sed 's/^/event-/' | kafka kafka-console-producer.sh --bootstrap-server kafka:9092 --topic events
echo "$(date +%T) offset $((b*5))~$((b*5+4)) 보냄"
[ $b -lt 2 ] && sleep 12
done
15:08:01 offset 0~4 보냄
15:08:17 offset 5~9 보냄
15:08:34 offset 10~14 보냄
offset 범위와 세그먼트 파일을 몇 초 간격으로 확인해 볼게요. log start offset이 15가 되면 멈춥니다.
while :; do
o=$(offsets events); echo "$(date +%T) $o $(segs events-0)"
[ "${o%% *}" = "start=15" ] && break
sleep 8
done
15:08:44 start=0 end=15 00000000000000000000.log 00000000000000000005.log 00000000000000000010.log
15:09:05 start=0 end=15 00000000000000000000.log.deleted 00000000000000000005.log 00000000000000000010.log
15:09:25 start=10 end=15 00000000000000000000.log.deleted 00000000000000000005.log.deleted 00000000000000000010.log
15:09:46 start=15 end=15 00000000000000000000.log.deleted 00000000000000000005.log.deleted 00000000000000000010.log.deleted 00000000000000000015.log
출력을 보면 offset 0~4를 담은 세그먼트는, 마지막 메시지를 보낸 15:08:01에서 60초가 지난 뒤 삭제 표시(.deleted)가 붙었어요.
Kafka는 세그먼트 안에서 가장 늦은 메시지의 타임스탬프를 기준으로 보관 기간을 따져요. 그래서 세그먼트 안의 메시지가 모두 60초를 넘겨야 그 세그먼트를 지웁니다. 세그먼트 하나가 크면, 그만큼 오래된 메시지가 보관 기간보다 오래 남는 거죠.
세그먼트가 지워질 때마다 log start offset이 0에서 10, 15로 올라갔어요. 컨슈머가 처음부터 읽겠다고 해도 이제 15보다 앞은 읽을 수 없습니다.
.deleted 파일은 이미 로그에서 빠진 파일이고, file.delete.delay.ms(기본 60초)가 지난 뒤 디스크에서 사라져요.
15:09:25에 10.log는 만들어진 지 50초가 넘었지만 여전히 활성 세그먼트였어요. segment.ms가 지나도 새 메시지가 들어와야 세그먼트가 바뀌기 때문이에요. 그런데 이 세그먼트의 메시지까지 보관 기간을 넘기자, Kafka는 빈 세그먼트 15.log를 새로 만들고 10.log를 지웠어요.
실습 3: 너무 늦은 컨슈머는 처음부터 읽지 못합니다
컨슈머가 멈춰 있는 동안 보관 기간이 지나면, 아직 읽지 않은 메시지도 지워져요.
events와 같은 설정으로 payments 토픽을 만들고 결제 메시지 10개를 보내 볼게요. 그리고 세 그룹이 각각 4개씩만 읽고 멈춥니다.
max.poll.records=1은 한 번에 1개씩 가져오게 하는 설정이에요. 콘솔 컨슈머가 4개를 받은 시점까지만 commit하도록 넣었습니다.
kafka kafka-topics.sh --bootstrap-server kafka:9092 --create --topic payments --partitions 1 --config retention.ms=60000 --config segment.ms=10000
seq 0 9 | sed 's/^/pay-/' | kafka kafka-console-producer.sh --bootstrap-server kafka:9092 --topic payments
echo "$(date +%T) offset 0~9 보냄"
for g in g-earliest g-latest g-none; do
kafka kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic payments --group $g --from-beginning --max-messages 4 --command-property max.poll.records=1 2>/dev/null | tr '\n' ' '; echo "<- $g"
done
describe --group g-earliest --group g-latest --group g-none
Created topic payments.
15:10:40 offset 0~9 보냄
pay-0 pay-1 pay-2 pay-3 <- g-earliest
pay-0 pay-1 pay-2 pay-3 <- g-latest
pay-0 pay-1 pay-2 pay-3 <- g-none
GROUP PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
g-earliest 0 4 10 6
GROUP PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
g-latest 0 4 10 6
GROUP PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
g-none 0 4 10 6
세 그룹 모두 4를 commit했고, offset 4~9는 아직 읽지 않았어요.
컨슈머가 멈춘 동안 새 메시지 5개가 더 들어오게 하고, 첫 세그먼트가 보관 기간을 넘길 때까지 기다려 볼게요.
sleep 40
seq 10 14 | sed 's/^/pay-/' | kafka kafka-console-producer.sh --bootstrap-server kafka:9092 --topic payments
echo "$(date +%T) offset 10~14 보냄"
until [ "$(offsets payments)" = "start=10 end=15" ]; do sleep 2; done
echo "$(date +%T) $(offsets payments)"
describe --group g-earliest --group g-latest --group g-none
15:11:45 offset 10~14 보냄
15:11:54 start=10 end=15
GROUP PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
g-earliest 0 4 15 11
GROUP PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
g-latest 0 4 15 11
GROUP PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
g-none 0 4 15 11
세 그룹의 commit 위치는 4인데, log start offset은 10이에요. 그룹이 다음에 읽으려는 offset 4는 이미 지워진 거죠.
그런데 describe 출력은 이 상태를 따로 알려 주지 않고, LAG도 여전히 11로 계산해요.
이렇게 저장된 위치가 파티션 범위를 벗어나면, 컨슈머는 auto.offset.reset 설정에 따라 새 위치를 정해요. 그룹에 commit된 위치가 아예 없을 때도 같은 설정을 씁니다.
latest: 기본값. 파티션의 끝(log end offset)으로 옮깁니다.earliest: 파티션에 남아 있는 가장 앞(log start offset)으로 옮깁니다.none: 위치를 옮기지 않고 컨슈머에 예외를 던집니다.
세 그룹을 각 설정으로 다시 띄워 볼게요.
kafka kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic payments --group g-earliest --command-property auto.offset.reset=earliest --timeout-ms 5000 2>/dev/null
kafka kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic payments --group g-latest --command-property auto.offset.reset=latest --timeout-ms 5000 2>/dev/null
kafka kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic payments --group g-none --command-property auto.offset.reset=none --timeout-ms 5000 2>&1 | grep OffsetOutOfRange
describe --group g-earliest --group g-latest --group g-none
pay-10
pay-11
pay-12
pay-13
pay-14
org.apache.kafka.clients.consumer.OffsetOutOfRangeException: Fetch position FetchPosition{offset=4, offsetEpoch=Optional[0], currentLeader=LeaderAndEpoch{leader=Optional[kafka:9092 (id: 1 rack: null isFenced: false)], epoch=0}} is out of range for partition payments-0
GROUP PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
g-earliest 0 15 15 0
GROUP PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
g-latest 0 15 15 0
GROUP PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
g-none 0 4 15 11
그룹별 결과는 이렇습니다.
g-earliest: 남아 있는 10부터 읽었습니다. 지워진pay-4부터pay-9까지 6개는 한 번도 처리하지 못했습니다.g-latest: 아무것도 읽지 않고 끝(15)으로 옮겼습니다. 지워진 6개에 더해, 아직 남아 있던pay-10부터pay-14까지 5개도 건너뛰어 11개를 놓쳤습니다.g-none:OffsetOutOfRangeException을 내고 멈췄습니다. commit 위치는 4 그대로입니다.
earliest와 latest에서는 컨슈머가 오류 없이 계속 돌고, 이후 LAG는 0이 돼요. 그래서 모니터링만 봐서는 메시지를 놓친 흔적이 남지 않습니다.
어느 설정을 골라도 이미 지워진 pay-4부터 pay-9까지는 Kafka에서 다시 읽을 수 없어요. 설정은 그다음 읽을 위치만 정해 줍니다.
정리
범위: Kafka 4.3(apache/kafka:4.3.1) 단일 브로커, log.retention.check.interval.ms 5초, 시간 실습 토픽은 retention.ms 60초와 segment.ms 10초.
billing그룹이 메시지 5개를 다 읽은 뒤에도 offset 범위는 0~5 그대로였어요. 토픽 크기는 컨슈머가 아니라 보관 정책으로 관리하세요.events토픽은 세그먼트 안의 마지막 메시지가 60초를 넘긴 뒤에야 세그먼트째 지워졌고, log start offset이 0에서 10, 15로 올라갔어요. 메시지는 보관 기간보다 최대 세그먼트 하나만큼 더 오래 남는다고 계산해 주세요.- 멈춰 있던 세 그룹의 commit 위치는 4인데 log start offset은 10이었고, describe 출력은 경고 없이 LAG 11만 보여 줬어요. commit 위치와
kafka-get-offsets.sh --time earliest를 따로 확인하세요. - 다시 띄우자
earliest는 지워진 6개를,latest는 남아 있던 5개까지 11개를 놓친 채 LAG 0이 됐고,none만OffsetOutOfRangeException으로 멈췄어요. 놓친 메시지를 알아야 하는 컨슈머는none으로 두고 위치를 사람이 정하세요.