[Kafka] 메시지 추적 및 상태 유지

개요

Kafka가 의도치 않게 종료되었다면 이전까지 읽은 데이터를 어떻게 추적할 수 있을까? 생각을 해보면 인입된 메시지 정보는 이전 개요에서 말했듯 로컬 스토리지에서 관리가 되기 때문에 브로커 서버가 내려가기 전 까지 인입된 메시지는 모두 보존되어 있을 것이고, 당연 어디까지 읽었는지 추적하는 무언가가 존재하기 때문에 데이터 영속성을 보장할 수 있을 것이다.

 

 

메시지 지속성

Kafka는 모든 메시지를 로컬 디스크에 저장하므로 파티션당 하나 이상의 로그 세그먼트 파일이 존재한다. 따라 서버가 재시작되어도 데이터가 유지된다. 이는 인덱스 파일로 관리가 되며 타임스탬프 기반 검색을 위한 검색이 가능하다.

/var/lib/kafka/data/my-topic-0/
  ├── 00000000000000000000.log      # 데이터 파일
  ├── 00000000000000000000.index    # 오프셋 인덱스
  ├── 00000000000000000000.timeindex # 시간 인덱스
  ├── 00000000000012345678.log      # 다음 세그먼트
  └── ...

 

 

소비자 오프셋 추적

Kafka는 각 소비자 그룹이 각 파티션에서 어디까지 읽었는지(오프셋)를 추적한다.

 

• 내부 토픽: __consumer_offsets 특수 토픽에 오프셋 정보 저장
• 주기적 커밋: 소비자가 주기적으로 현재 오프셋을 커밋(기록)
• 지속성: 이 오프셋 정보도 디스크에 저장되므로 서버 재시작 후에도 유지

# 소비자 그룹의 오프셋 확인 예시
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group my-consumer-group

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
my-consumer-group my-topic       0          157             157             0
my-consumer-group my-topic       1          152             162             10

 

 

 

재시작 시나리오

Kafka 서버(브로커)를 내렸다가 다시 올리면 다음과 같은 과정이 진행된다.

 

1. 메시지 복원

 

• 디스크에서 로그 파일 로드
• 모든 토픽/파티션 메시지 복원
• 토픽 메타데이터 복원

 

 

2. 오프셋 복원

 

• __consumer_offsets 토픽 복원
• 모든 소비자 그룹의 오프셋 정보 로드

 

 

3. 소비자 재연결 시

 

• 소비자는 마지막으로 커밋한 오프셋부터 다시 읽기 시작
• 아직 읽지 않은 메시지는 그대로 남아있음
• 이미 읽었지만 커밋하지 않은 메시지는 다시 읽힐 수 있음

 

 

소비자 오프셋 관리 방식

소비자 오프셋 관리 방식에 따라 재시작 후 동작이 달라질 수 있다.

 

1. 자동 커밋 (기본 설정)

properties.put("enable.auto.commit", "true");
properties.put("auto.commit.interval.ms", "5000");  // 5초마다 자동 커밋

• 일정 간격으로 오프셋 자동 커밋
• 간편하지만 중복 처리 가능성 있음

 

2. 수동 커밋

 

// 메시지 처리 후 명시적 커밋
consumer.poll(Duration.ofMillis(100)).forEach(record -> {
    processRecord(record);
    consumer.commitSync(
        Collections.singletonMap(
            new TopicPartition(record.topic(), record.partition()),
            new OffsetAndMetadata(record.offset() + 1)
        )
    );
});

• 개발자가 명시적으로 오프셋 커밋
• 정확한 제어 가능, 정확히 한 번 처리 구현 가능

 

실제 예시 시나리오

1. 시나리오 1: 재시작 후 미처리 메시지

1. 소비자가 오프셋 100까지 읽고 커밋
2. 현재 파티션에는 오프셋 150까지 메시지 존재
3. Kafka 브로커 재시작
4. 소비자 재연결
5. 소비자는 오프셋 101부터 읽기 시작 (미처리 메시지 처리)

 

2. 시나리오 2: 커밋되지 않은 메시지

1. 소비자가 오프셋 100까지 읽었지만 오프셋 80까지만 커밋
2. Kafka 브로커 재시작
3. 소비자 재연결
4. 소비자는 오프셋 81부터 다시 읽기 시작 (일부 메시지 재처리)

 

 

추가 구성 옵션

1. 오프셋 자동 리셋 정책

// 소비자 구성 - 오프셋이 없거나 유효하지 않을 때 동작 설정
properties.put("auto.offset.reset", "earliest");  // 가장 오래된 메시지부터 시작
// 또는
properties.put("auto.offset.reset", "latest");    // 가장 최근 메시지부터 시작

 

2. 오프셋 보존 기간

# 브로커 설정 - 오프셋 정보 유지 기간
offsets.retention.minutes=10080  # 7일 (기본값)

 

 

볼륨 관리

메시지가 축적된다면 로컬 스토리지에 대한 용량 부담도 함께 커지게 될 것이다. 결국 해당 메시지들을 읽은 오프셋까지 혹은 일부 메시지들에 대해 스냅샷을 찍어 별도의 네트워크 스토리지나 하드 디스크에 백업을 해두어야 할 것이다.

 

만약 이렇게 이미 읽은 오프셋에 이하의 데이터가 볼륨에서 제거되게 된다면 오프셋은 어떻게 될까? 0으로 초기화가 되거나 삭제된 데이터의 오프셋 만큼 감소하게 될까? 우선 이미 읽은 오프셋과 볼륨 데이터의 관계를 알아보자

 

1. 읽은 오프셋과 메시지 데이터

• 다시 읽히지는 않지만 (커밋된 오프셋 이후부터 읽기 시작)
• 볼륨에는 여전히 존재한다. (보존 정책에 따라 유지)

Kafka는 데이터 소비와 데이터 보존을 별개로 취급하여 소비자가 읽었다고 해서 메시지가 삭제되지 않고, 메시지는 보존 정책(시간/크기)에 따라서만 삭제된다.

 

2. 볼륨 데이터가 제거된 경우의 오프셋 동작

볼륨에서 오래된 데이터가 제거된 경우(보존 정책에 의해), 소비자 오프셋은 0으로 초기화 되지 않는다, 다만 다음과 같은 상황이 발생한다.

오프셋 불일치 상황:

소비자가 마지막으로 커밋한 오프셋: 100
보존 정책으로 인해 오프셋 80까지의 메시지가 디스크에서 삭제됨
현재 볼륨에는 오프셋 81부터 존재


소비자 재연결 시 동작:

소비자는 오프셋 101부터 읽으려고 시도
이는 여전히 유효한 위치이므로 정상적으로 계속 소비


요청한 오프셋이 없는 경우:

만약 보존 정책으로 인해 소비자가 요청한 오프셋이 이미 디스크에서 삭제된 경우
소비자의 auto.offset.reset 설정에 따라 동작이 결정됨:

earliest: 현존하는 가장 오래된 오프셋부터 시작 (예: 81)
latest: 가장 최신 오프셋부터 시작

 

3. 실제 시나리오 예시

시나리오 A: 소비자 오프셋이 유효한 경우

1. 소비자가 오프셋 5000까지 읽고 종료
2. 보존 정책으로 오프셋 0-2999의 메시지 삭제됨
3. 소비자 재시작
4. 소비자는 오프셋 5001부터 읽기 시작 (정상)

 

시나리오 B: 소비자 오프셋이 유효하지 않은 경우

1. 소비자가 오프셋 1500까지 읽고 오랫동안 비활성화
2. 보존 정책으로 오프셋 0-2999의 메시지 삭제됨
3. 소비자 재시작 - 오프셋 1501부터 읽으려 함
4. 해당 오프셋이 없으므로 auto.offset.reset 정책 적용
   - earliest: 오프셋 3000부터 읽기 시작 (가장 오래된 가용 오프셋)
   - latest: 현재 최신 오프셋부터 읽기 시작