신뢰성 있는 데이터 전달, Exactly at-once semantics

신뢰성 있는 데이터 전달

카프카가 보장하는 신뢰성

• 카프카는 파티션 안의 메시지들 간의 순서를 보장
• 클라이언트가 쓴 메시지는 모든 in-sync replica 의 파티션에 쓰여진 뒤에 커밋 된 것으로 간주
  • 프로듀서는 acks 설정에 따라 메시지가 커밋된 다음 응답이 올지, 리더에만 쓰여졌을 때 응답이 올지, 네트워크 전송된 다음 응답이 올지 결정할 수 있음
• 커밋된 메시지는 최소1개의 작동 가능한 레플리카가 남아있는 한 유실되지 않음
• 컨슈머는 커밋 된 메시지만 읽을 수 있음

복제

• 모든 이벤트들은 리더 레플리카에 쓰여지며, 대체로 리더 레플리카에서 읽혀짐
• 다른 레플리카들은 리더 레플리카와 동기화를 맞추며 최신 이벤트를 복사해 감
• 리더 레플리카가 죽을 경우, 인 싱크 레플리카 중 하나가 리더가 됨

다음 아래 조건에 따라 in-sync replica 가 됨

• 주키퍼와의 활성 세션이 있는 경우
• 최근 10초 사이 리더로부터 메시지 읽어온 경우
• 최근 10초 사이 리더로부터 읽어온 메시지가 가장 최근 메시지일 경우
  • 최근 10초 사이 랙이 없는 경우

동기화가 늦은 in-sync 레플리카는 프로듀서와 컨슈머 둘다 느리게 만들 수 있음

• 프로듀서는 모든 in-sync 레플리카가 메시지를 받을 때까지 기다려야 함
• 컨슈머는 커밋된 메시지만 읽을 수 있기 때문

신뢰성과 관련된 브로커 설정

복제 팩터

토픽 단위 설정(replication.factor ), 브로커 단위 설정(default.replication.factor )

복제 팩터가 N 일 경우 N-1 개의 브로커가 작동 불능이 되더라도 토픽의 데이터를 읽거나 쓸 수 있음

BUT 가용성과 하드웨어 사용량 사이의 trade-off 를 고려해야 함.

파티션의 레플리카들이 서로 다른 랙에 분산되어 저장되게 하여(borker.rack 매개변수에 랙 이름을 잡아줌) 가용성을 높이는게 좋음

언클린 리더 선출

unclean.leader.election.enable 로 설정.

클린 리더란 파티션 리더가 작동 불능일 경우 인 싱크 레플리카 중 하나가 리더가 되는 것을 의미 BUT 작동 불능인 리더 외에 out-of-sync 레플리카 밖에 없다면 다음과 같은 문제 발생함

• out-of-sync 레플리카가 리더가 될 수 없다면, 해당 파티션은 해당 리더가 복구될 때까지 오프라인 상태가 됨. 가용성이 줄어듦
• out-of-sync 레플리카가 리더가 될 수 있으면 (⇒ 언클린 리더 선출이 가능하게 설정할 경우) 새 리더가 동기화를 못한 데이터에 대해서 예전 리더에 쓰여졌던 데이터들이 유실되며, 컨슈머는 예전 리더의 메시지 + 새 리더의 메시지가 뒤섞인 채로 읽게 됨

out-of-sync 레플리카가 리더가 될 수 있도록 허용할 경우 데이터 유실 및 일관성이 깨질 수 있음

최소 인-싱크 레플리카

min.isync.replicas 로 설정.

커밋된 데이터를 2개 이상의 레플리카에 쓰고자 한다면 in-sync 레플리카의 최소값을 높게 잡아줘야 함

만약 3개의 레플리카 모두가 in-sync 레플리카이고, min.isync.replicas 가 2로 설정되었을 때 3개 중 2개가 작동 불능 상태가 되면, 해당 브로커는 더이상 쓰기 요청을 받지 않고(프로듀서는 NotEnoughReplicasException 을 받게 됨) 해당 레플리카는 읽기 전용이 됨 (컨슈머는 데이터를 읽을 수 있음)

위 설정값은 언클린 리더 선출이 발생하면 사라질 데이터를 읽거나 쓰는 상황을 방지할 수 있음

레플리카를 인-싱크 상태로 유지

zookeeper.session.timeout.ms : 카프카 브로커가 주키퍼로 하트비트 전송을 멈출 수 있는 최대 시간을 정의함. 이 간격 안에만 하트비트를 보내면 브로커는 죽었다고 판단되지 않음.

가비지 수집, 네트워크 상황과 같은 무작위적인 변동에 영향 받지 않을 만큼 높게 작동이 멈춘 브로커가 적시에 탐시될 수 있을 만큼 낮게 설정하여야 함.

replica.lag.time.max.ms : 이 시간 이상으로 리더로부터 데이터를 읽어오지 못하거나 리더에 쓰여진 최신 메시지를 따라잡지 못할 경우 out-of-sync 레플리카가 됨. 이 값은 메시지가 모든 in-sync 레플리카에 도착해서 컨슈머가 읽을 수 있게 하는 시간과 연관되어 있으므로, 컨슈머의 최대 지연에도 영향 줄 수 있음

디스크에 저장

카프카는 메시지 받은 레플리카 수에 의존할 뿐 디스크에 저장되지 않은 메시지에도 응답함 디스크로 플러시 하는 시점은 세그먼트를 교체할때와 재시작 직전에만 메시지를 디스크로 플러시하며, 그 외에는 페이지 캐시 기능을 이용함

flush.messages 설정값 이용해서 디스크에 저장되지 않은 최대 메시지 수를, flush.ms 는 얼마나 자주 디스크에 메시지를 저장하는지조절함

신뢰성 있는 시스템에서 프로듀서 사용하기

ack 설정

acks=0 : 프로듀서가 네트워크로 메시지를 전송한 시점에서 메시지가 카프카에 성공적으로 쓰여진 것으로 간주. 지연은 낮지만 종단 지연이 개선되진 않음(컨슈머는 커밋된 메시지만 읽을 수 있기 때문)

acks=1 : 리더가 메시지를 받아서 파티션 데이터 파일에 쓴 직후 응답 또는 에러를 받음. 팔로워로 복제되기 전 리더가 크래쉬 날 경우, 불완전 복제 파티션이 발생할 수 있음.

acks=all : 리더 + 모든 인 싱크 레플리카가 메시지 받을 때까지 기다렸다가 응답하거나 에러를 보낸다는 것 의미. min.isync.replicas 설정과 더불어 응답이 오기 전까지 얼마나 많은 레플리카에 메시지 복사될 건지 조절할 수 있음. 완전히 커밋될 때까지 프로듀서는 메시지를 재전송함, 제일 안정적이지만 프로듀서 지연이 길어질 수 있음.

재시도 설정

LEADER_NOT_AVAILABLE 같은 재시도 가능한(retriable error) 일 경우, 프로듀서가 메시지 재전송을 하도록 설정함

• 재시도 수를 기본 설정값(MAX_INT ) 으로 두고, 메시지 전송을 포기할 때까지 대기할 수 있는 시간(delivery.timeout.ms ) 을 최대값으로 잡아 그 시간동안 메시지 전송을 계속 시도하도록 함
• BUT 재시도 할 경우 메시지가 중복될 수 있음, 이 경우 enable.idempotence=true 설정을 통해 브로커가 재시도로 인해 중복된 메시지를 건너뛸 수 있게 함

추가적으로 타임아웃, 브로커 전송하기 전에 발생한 에러 등등은 적절한 에러 핸들러를 통해 해결하자

신뢰성 있는 시스템에서 컨슈머 사용하기

오프셋을 커밋 하는 이유: 읽고 있는 파티션에 대해 어디까지 읽었는지를 기록해 둬야 해당 컨슈머나 다른 컨슈머가 재시작 한 뒤에도 그에 이어서 작업을 할 수 있음

컨슈머가 메시지를 누락할 수 있는 경우는 읽기는 완료 했으나 처리는 완료되지 않은 메시지를 커밋했을 경우임

컨슈머 설정

• group.id : 같은 컨슈머 그룹에 속하는 두 개의 컨슈머가 한 토픽을 구독할 경우 각각은 서로 다른 파티션만의 메시지를 읽게 됨
• auto.offset.reset : 커밋된 오프셋이 없을 때, 컨슈머가 브로커에 없는 오프셋 요청할 때 컨슈머가 어떤 오프셋을 읽어올지 정의. 파티션의 맨 앞의 오프셋 읽어오는 earliest (이 경우 중복이 발생할 수 있지만 유실은 발생하지 않음), 파티션 맨 뒤 오프셋 읽어오는 latest (중복은 없으나 일부 메시지는 누락될 수 있음)
• aenable.auto.commit : 자동 오프셋 커밋 기능은 처리하지 않은 오프셋을 실수로 커밋하는 것을 방지해주지만 중복 처리를 제어할 수는 없다 (처리 도중에 컨슈머 멈추면 컨슈머 재시작 시 메시지가 중복 처리됨)
• auto.commit.interval.ms : 자동 커밋 주기 설정

명시적 오프셋 커밋

• 메시지 처리 먼저, 오프셋 커밋은 나중에 하라: 자동 오프셋 커밋 설정을 사용하거나, 폴링 루프의 끝에서 오프셋을 커밋하거나, 아니면 루프 안에서 일정한 주기로 오프셋을 커밋할 것
• 커밋 빈도는 성능과 크래시 발생 시 중복 개수 사이의 트레이드 오프: 커밋 작업은 성능 오버헤드를 수반함, 커밋 주기가 잦게 되면 크래시 발생 시 중복 개수는 줄어들지만 오버헤드는 증가함
• 정확한 시점에 정확한 오프셋을 커밋
• 컨슈머는 재시도를 해야할 수 있음. 레코드를 처리한 뒤 일부 레코드는 처리가 완료되지 않을 수 있음
  • 마지막으로 성공한 레코드의 오프셋을 커밋하고, 나중에 처리해야 할 레코드는 버퍼에 저장 후 pause() 메서드를 호출해서 추가적인 poll() 호출이 데이터를 리턴하지 않도록 함. 버퍼에 저장된 레코드를 처리함
  • 또는 처리가 완료되지 않은 메시지를 재시도 전용 별도의 토픽에 쓴 뒤 진행하는 것. DLQ 와 비슷
• 컨슈머는 상태를 유지해야 될 수 있음 - poll() 호출 간 상태를 유지해야 하는 경우가 있음. 각 poll 에서 읽은 값을 result 토픽에 써야 하는 경우 .. 이 경우 트랜잭션 기능을 사용하자

모니터링

프로덕션 환경에서 모니터링

JMX 지표 중 레코드별 에러율(error-rate ), 재시도율( retry-rate )을 볼 것

컨슈머 쪽에서 볼 지표는 컨슈머가 브로커 내 파티션에 커밋된 최신 메시지로부터 얼마나 뒤떨어져 있는지 나타내는 consumer lag 을 볼 것 카프카 브로커가 클라이언트에게 보내는 에러 응답률 지표 볼 것 (브로커가 보낸 응답이 태그 형식으로 되어있음)

Exactly-once semantics

멱등적 프로듀서

멱등적(idempotent) : 동일한 작업을 여러 번 수행해도 한 번 실행한 것과 결과가 같은 것

카프카에서 프로듀서가 메시지 재전송을 시도하면 (프로듀서 입장에서는 메시지 전송이 안되었다고 판단해 재전송 했으나, 브로커는 메시지를 성공적으로 복제한 후 크래쉬 난 상태에서 다시 복구되었을 경우, 동일한 메시지가 중복될 수 있음) 메시지 중복이 발생할 수 있음

카프카의 멱등적 프로듀서 기능은 이러한 중복을 탐지하고 처리할 수 있게 함

프로듀서에 enabled.idempotence=true 설정 추가

• 전송되는 각 레코드 배치에는 프로듀서 ID와 배치 내 첫 메시지의 시퀀스 넘버 포함
• 브로커는 레코드 배치의 시퀀스 넘버를 검증해서 메시지 중복을 방지
• 장애 바랭하더라도 각 파티션에 쓰여지는 메시지 순서는 보장

작동 원리

멱등적 프로듀서 기능을 키면 모든 메시지는 고유한 프로듀서 ID 와 시퀀스 넘버를 가지게 됨

대상 토픽 파티션 + producer id + sequence number 를 합치면 각 메시지의 고유한 식별자가 됨

해당 브로커는 할당된 모든 파티션에 쓰여진 마지막 5개 메시지들을 추적하기 위해 이 고유 식별자를 사용함. 추적되야 하는 시퀀스 수를 제어하고 싶으면 max.in.flights.requests.per.connection 설정값을 5 이하로 조정할 것.

브로커가 이전에 받은 메시지를 다시 받을 경우 에러를 발생시킴

프로듀서에서는 record-error-rate 지표값 확인하여 에러 확인할 수 있고, 브로커일 경우 RequestMetrics 유형의 ErrorsPerSec 지표값에 기록됨

브로커가 예상보다 높은 시퀀스 넘버를 받게 되면 (2번, 3번 .. 다음에 27번 받으면), out of order sequence number 에러를 발생시킴

• 프로듀서 재시작: 프로듀서가 재시작 될 경우 멱등적 프로듀서 기능이 켜져 있을 경우 프로듀서는 초기화 과정에서 카프카 브로커로부터 프로듀서 ID 를 생성받음. (멱등적 프로듀서 기능을 켜지 않을 경우, 프로듀서는 새로운 ID를 브로커로부터 할당받게 되며 이 경우 메시지의 중복을 탐지하지 못함)
• 브로커 장애 : 리더는 새 메시지가 쓰여질 때마다 인 메모리 프로듀서 상태에 저장된 최근 5개의 시퀀스 넘버를 업데이트하고, 팔로워 레플리카는 리더로부터 메시지 복제할 때마다 자체적인 인 메모리 버퍼를 업데이트 함.예전 리더가 다시 돌아온 경우, 인 메모리 프로듀서 상태는 메모리에 저장되어 있지 않지만 프로듀서 상태에 대해 저장된 스냅샷 파일에서 최신 상태를 읽어오고 현재 리더로부터 복제한 레코드를 사용해서 프로듀서 상태를 업데이트하여 최신 상태를 복구함
• 각 프로듀서 상태에 저장된 스냅샷이 업데이트 되지 않았다면 각 파티션 최신 세그먼트의 메시지를 사용해서 복구하고, 복구 작업이 완료되면 새로운 스냅삿 파일이 완성됨
• 브로커 장애가 발생할 경우 팔로워가 리더가 된 시점에 팔로워 레플리카는 이 인 메모리 버퍼를 통해 새로 쓰여진 메시지의 유효성 검증을 재개할 수 있음.

멱등적 프로듀서의 한계

멱등적 프로듀서가 방지할 수 있는 메시지 중복은 ONLY 프로듀서의 내부 로직으로 인한 재시도가 발생하는 경우 생기는 중복만을 방지할 수 있음

의미적으로 동일한 메시지를 producer.send()를 두 번 호출한다고 해서 메시지가 중복되는 것을 막아주지는 않는다

즉 멱등적 프로듀서는 프로듀서 자체의 재시도 매커니즘(프로듀서, 브로커, 네트워크 에러로 인해 발생하는) 으로 인해 발생하는 중복만 방지해줌

트랜잭션

원본 토픽으로부터 이벤트 읽어서 처리 한 후 결과를 토픽에 쓰고 해당 메시지를 커밋하는 경우, 각 메시지에 대해 결과가 정확히 한 번만 쓰여지게 하고 싶다면 다음과 같은 문제가 발생할 수 있음

• 애플리케이션 크래시 : 출력 토픽에는 썼는데 입력 오프셋이 커밋되기 전 애플리케이션이 크래시 난다면 파티션에 새로 할당된 컨슈머는 마지막으로 커밋된 오프셋에서부터 크래시가 난 시점까지 처리된 레코드들을 중복해서 처리할 것이며 출력 토픽에 다시 쓰여질 것
• 좀비 애플리케이션에 의해 발생하는 재처리: 컨슈머가 크래시나고, 다른 컨슈머가 할당되어 처리 중일 경우, 기존 컨슈머가 다시 작동하게 되면, 자기가 죽은 것으로 판정된 것으로 알기 전까지 마지막으로 읽어왔던 레코드 배치 처리하고 결과를 출력 토픽에 쓰게 될 수 있음. 따라서 중복된 결과가 발생할 수 있음.

원자적 다수 파티션 쓰기

정확히 한 번은 읽기 , 처리, 쓰기 작업이 원자적(all or nothing) 으로 이루어짐

이를 위해 카프카 트랜잭션은 원자적 다수 파티션 쓰기 도입

결과는 출력 토픽에, 오프셋은 _consumer_offsets 토픽에 쓰여진다는 점을 이용

원자적 다수 파티션 쓰기 는 출력 레코드가 토픽에 커밋되었을 경우 입력 레코드의 오프셋도 해당 컨슈머에 커밋되었음을 보장함

원자적 다수 파티션 쓰기를 수행하려면 트랜잭션적 프로듀서 를 사용

• transactional.id 설정이 잡혀 있고, initTransactions() 를 호출해서 초기화
• 재시작을 해도 transactional.id 는 유지, 재시작 후에도 동일한 프로듀서를 식별
• 브로커는 transactional.id , producer.id 간 대응 관계 유지

좀비 인스턴스가 중복 메시지를 생성하는 것을 방지하기 위해, 좀비 펜싱 필요

• epoch 를 사용해 initTransactions 를 호출하면 transactional.id 에 해당하는 에포크 값 증가
• 만약 transactional.id 는 같지만 현재 epoch 보다 낮은 프로듀서가 메시지 전송, 트랜잭션 커밋 등 요청 보낼 경우 FencedProducer 에러 발생하면서 거부됨

컨슈머의 경우도 isolation.level 설정 값을 설정할 수 있음

• read_comitted: 커밋된 트랜잭션에 속한 메시지나 처음부터 트랜잭션에 속하지 않는 메시지 리턴. 아직 진행중인 트랜잭션이 처음으로 시작된 시점(LSO; Last Stable Offset) 이후에 쓰여진 메시지는 리턴되지 않음 (⇒ 이 메시지들은 트랜잭션이 커밋되거나 중단되거나 혹은 transaction.timeout.ms 설정값이 지나 브로커가 트랜잭션을 중단될 때까지 리턴되지 않음, 트랜잭션이 오랫동안 닫히지 않으면 종단 지연이 길어짐)
• read_uncomitted: 진행중이거나 중단된 트랜잭션에 속하는 것들 포함하여 모든 레코드가 리턴

트랜잭션으로 해결할 수 없는 문제

• 위 기능은 카프카에 쓰여지는 레코드의 all or nothing 을 보장하는 것, 기타 api 호출이나 파일 쓰기 등에 대한 원자성을 보장하는 것은 아니다
• 트랜잭션에서 외부 DB에 결과를 쓰고 오프셋을 커밋하도록 하는 매커니즘은 없음. 이 경우 transactional outbox pattern 을 사용할 것
• 한 클러스터에서 다른 클러스터로 복사할 때 정확히 한 번은 보장할 수 있으나, 복사 과정에서 트랜잭션 속성이나 보장은 유실됨. 데이터를 읽어오는 카프카 컨슈머 입장에서 트랜잭션의 모든 데이터를 읽어왔는지는 알 수도 없음.
• 발행/구독 패턴에서 read_committed 모드인 컨슈머는 중단된 트랜잭션에 속한 레코드를 보지 못할 것이고, 오프셋 커밋 로직에 따라 메시지를 중복해서 처리할 수 있음

트랜잭션 사용법

• 카프카 스트림즈에서 exactly-once 보장 활성화

KafkaProducer producer = createKafkaProducer(
  “bootstrap.servers”, “localhost:9092”,
  “transactional.id”, “my-transactional-id”); //transactional.id 생성

producer.initTransactions(); //초기화 작업 실행
//트랜잭션 ID 등록하고, 동일한 트랜잭션 ID 갖는 프로듀서들이 좀비로 인식될 수 있게
//에포크 값 증가

KafkaConsumer consumer = createKafkaConsumer(
  “bootstrap.servers”, “localhost:9092”,
  “group.id”, “my-group-id”,
  "isolation.level", "read_committed"); //read_committed: LSO 이전의 데이터만 읽어옴

consumer.subscribe(singleton(“inputTopic”));

while (true) {
  ConsumerRecords records = consumer.poll(Long.MAX_VALUE);
 //이 시점부터 트랜잭션이 종료되는 시점까지 쓰여진 레코드들이 모두
 //하나의 원자적 트랜잭션의 일부임을 보장
  producer.beginTransaction();
  for (ConsumerRecord record : records)
    producer.send(producerRecord(“outputTopic”, record));
  //현재 오프셋 커밋
  producer.sendOffsetsToTransaction(currentOffsets(consumer), group);  
  //트랜잭션 완료
  producer.commitTransaction();
}

트랜잭션 ID 와 펜싱

~ver2.5 까지, 트랜잭션 ID를 파티션에 정적으로 대응시켜 펜싱을 보장했음

즉 새 프로듀서가 들어올 경우 이 프로듀서의 트랜잭션 ID와 크래시났던 전 프로듀서의 트랜잭션 ID 가 다르기 때문에 펜싱되지 않게 되고, 좀비 프로듀서가 살아있게 됨

즉 동일한 파티션에 쓰기 작업을 할 때 언제나 동일한 트랜잭션 ID 가 쓰일 거라는 보장이 없음

이후 트랜잭션 ID와 컨슈머 그룹 메타데이터를 함께 사용하는 펜싱 도입

트랜잭션에 컨슈머 그룹 정보를 포함하여 현재 요청한 트랜잭션이 최신 세대의 컨슈머 그룹에서 온 것이 명백하면 작업을 진행함

트랜잭션의 작동 원리

카프카는 transaction_state 라는 이름의 내부 토픽을 사용해 트랜잭션 로그를 씀

1. initTransaction() 호출하여 트랜잭션 프로듀서 등록
   • 이 요청은 트랜잭션 코디네이터 역할을 맡을 브로커로 보내짐
   • 트랜잭션 코디네이터는 각 트랜잭션 ID에 해당하는 트랜잭션 로그 파티션의 리더 브로커가 맡음
   • 새 트랜잭션 아이디를 등록하거나, 기존 트랜잭션 아이디의 에포크 값을 증가시킴
2. beginTransaction() 호출하여 현재 진행중인 트랜잭션이 존재함을 알림. 프로듀서가 레코드 전송 할 때마다 브로커에 AddPartitonToTxn 요청을 보내 이 프로듀서에 진행중인 트랜잭션이 있으며 레코드가 추가되는 파티션들이 트랜잭션의 일부임을 알림. 이 정보는 트랜잭션 로그에 기록
3. 쓰기 작업이 완료되고 sendOffsetsToTransaction 호출하면 트랜잭션 코디네이터로 오프셋, 컨슈머 그룹 ID 가 포함된 요청이 전송되고 트랜잭션 코디네이터는 컨슈머 그룹 ID로 컨슈머 그룹 코디네이터를 찾은 뒤 오프셋을 커밋
4. commitTransaction(), abortTransaction() 호출하면 트랜잭션 코디네이터에 EndTxn 요청 전송 트랜잭션 코디네이터는 커밋/중단 시도를 기록한 후 트랜잭션에 포함된 모든 파티션에 커밋 마커 를 쓴 다음 트랜잭션 로그에 커밋이 성공적으로 완료되었음을 기록 (크래시 날 경우 새로운 트랜잭션 코디네이터가 선출되어 작업을 마무리함)
5. transaction.timeout.ms 에 설정된 기간 내에 커밋 혹은 중단되지 않는다면 트랜잭션 코디네이터가 자동으로 트랜잭션을 중단함

트랜잭션 성능

트랜잭션 초기화, 커밋 요청은 동기적으로 작동함

BUT 트랜잭션 오버헤드는 트랜잭션에 포함된 메시지의 수와는 무관하기 때문에 트랜잭션마다 많은 수의 메시지를 처리하는 것이 오버헤드를 줄일 수 있고, 동기적으로 실행되는 단계의 수도 줄어들게 됨

컨슈머 쪽에서 read_committed 모드로 레코드를 읽어올 때 아직 완료되지 않은 트랜잭션의 레코드들이 리턴되지 않으므로 트랜잭션을 적용했을 때 트랜잭션 커밋 사이의 간격이 길어질 수록 종단 지연이 길어질 수 있음

그러나 컨슈머는 아직 리턴되지 않은 메시지를 버퍼링할 필요가 없으므로, 트랜잭션 데이터를 읽을때 컨슈머 쪽에서 해줘야 할 추가적인 작업은 없음

누락보다 중복이 낫다? 읽은 오프셋을 레디스에 써서 중복을 감지함