Kafka 심화) '정확히 한 번' 의미 구조

이전 챕터에서 신뢰성 보장을 제어할 수 있게 해주는 설정 매개변수들과 모범사례를 보았다. 여기서 우리는 '최소 한번' 전달에 초점을 맞췄다. 

 

멱등적 프로듀서

간혹 재시도는 메시지 중복을 발생시킨다. 가령 다음과 같은 시나리오에서..

- 파티션 리더가 프로듀서로부터 레코드를 받아서 팔로워들에게 성공적으로 복제한다.

- 프로듀서에게 응답을 보내기 전, 파티션 리더가 있는 브로커에 크래시가 발생한다. 

- 프로듀서 입장에서는 응답을 받지 못한 채 타임아웃이 발생하고, 메시지를 재전송한다.

- 재전송된 메시지가 새 리더에 도착한다. 하지만 이 메시지는 이미 저장되어 있다. 

 

이러한 케이스를 대응하기 위해서 우리는 멱등적 프로듀서를 사용한다. 

 

멱등적 프로듀서의 작동 원리

멱등적 프로듀서 기능을 켜면 모든 메시지는 고유한 프로듀서 ID와 시퀀스 넘버를 가지게 된다. 해당 브로커에 할당된 모든 파티션들에 쓰여진 마지막 5개 메시지들을 추적하기 위해 이 고유 식별자를 사용한다. 파티션별로 추적되어야 하는 시퀀스 넘버의 수를 제한하고 싶다면 프로듀서의 max.in.flights.requests.per.connection 설정값이 5 이하로 잡혀 있어야한다. (기본값 5)

 

브로커가 예전에 받은 적이 있는 메시지를 받게 될 경우, 적절한 에러를 발생시킴으로써 중복 메시지를 거부한다. 

 

과연 멱등적 프로듀서가 어떻게 처리하는지를 생각해보는 것은 의미가 있다. 프로듀서 재시작과 브로커 장애, 두 경우를 생각해보자. 

 

1) 프로듀서 재시작 Case

프로듀서에 장애가 발생할 경우, 보통 새 프로듀서를 생성해서 장애가 난 프로듀서를 대체한다. 여기서 중요한 점은 프로듀서가 시작될 때 멱등적 프로듀서 기능이 켜져 있을 경우, 프로듀서는 초기화 과정에서 카프카 브로커로부터 프로듀서 ID를 생성받는 다는 점이다. 트랜젝션 기능을 켜지 않았을 경우, 프로듀서를 초기화할 때마다 완전히 새로운 ID가 생성된다. 즉, 프로듀서에 장애가 발생해서 대신 투입된 새 프로듀서가 기존 프로듀서가 이미 전송한 메시지를 다시 전송할 경우, 브로커는 메시지에 중복이 발생했음을 알아차리지 못한다. 

 

2) 브로커 장애 Case

하지만 브로커 3 입장에서, 어느 시퀀스 넘버까지 쓰여졌는지 어떻게 알고 중복 메시지를 걸러내는가? 리더는 새 메시지가 쓰여질 때마다 인-메모리 프로듀서 상태에 저장된 최근 5개의 시퀀스 넘버를 업데이트한다. 팔로워 레플리카는 리더로부터 새로운 메시지를 복제할 때마다 자체적인 인-메모리 버퍼를 업데이트 한다. 하지만, 여기서 예전 리더가 다시 돌아온다면 어떤 일이 벌어질까? 브로커는 종료되거나 새 세그먼트가 생성될 때마다 프로듀서 상태에 대한 스냅샷을 파일 형태로 저장한다. 브로커가 시작되면 일단 파일에서 최신 상태를 읽어온다. 그러고 나서 현재 리더로부터 복제한 레코드를 사용해서 프로듀서 상태를 업데이트함으로써 최신 상태를 복구한다. 그래서 이 브로커가 다시 리더를 맡을 준비가 될 시점에는 최신 시퀀스 넘버를 가지고 있게 한다.

 

멱등적 프로듀서의 한계

카프카의 멱등적 프로듀서는 프로듀서의 내부 로직으로 인한 재시도가 발생할 경우 생기는 중복만을 방지한다. 동일한 메시지를 가지고 producer,send()를 두번 호출하면 멱등적 프로듀서가 개입하지 않는 만큼 중복된 메시지가 생기게 된다. 프로듀서 입장에서는 전송된 레코드 두개가 실제로는 동일한 레코드인지 확인할 방법이 없기 때문이다. 프로듀서 예외를 잡아서 애플리케이션이 직접 재시도 하는 것보다는 프로듀서에 탑재된 재시도 메커니즘을 사용하는 것이 언제나 더 낫다. 멱등적 프로듀서는 이 패턴을 더 편리하게 만들어 준다.

 

멱등적 프로듀서 사용법

이 부분은 쉽다. 프로듀서 설정에 enable.idempotence=true 를 추가해주면 끝이다. 만약 프로듀서에 acks=all 설정이 이미 잡혀 있다면, 성능에는 차이가 없을 것이다. 멱등적 프로듀서 기능을 활성화 시키면 다음과 같은 것들이 바뀐다. 

프로듀서 ID를 받아오기 위해 프로듀서 시동 과정에서 API를 하나 더 호출한다. 

전송되는 각각의 레코드 배치에는 프로듀서 ID와 배치 내 첫 메시지의 시퀀스 넘버가 포함된다. 이 새 필드들은 각 메시지 배치에 96 비트를 추가한다. 

브로커들은 모든 프로듀서 인스턴스에서 들어온 레코드 배치의 시퀀스 넘버를 집중해서 메시지 중복을 방지한다. 

 

트랜잭션 활용 사례

금융 애플리케이션은 '정확히 한번' 기능이 정확한 직접 결과를 보장하는데 쓰이는 복잡한 스트림 처리 애플리케이션의 전형적인 예다. 하지만, 카프카 스트림즈 애플리케이션이 '정확히 한 번' 보장을 제공하도록 설정하는 것이 상당히 단순한 만큼 챗본과 같이 더 흔한 활용 사례에서도 이 기능이 활용되는 것을 볼 수 있다. 

 

트랜젝션이 해결하는 문제

1) 애플리케이션 크래시로 인한 재처리

애플리케이션은 두 가지를 해야한다. 즉, 하나는 결과를 출력 토픽에 쓰는 것이고 또 하나는 우리가 읽어 온 메시지의 오프셋을 커밋하는 것이다. 만약 출력 토픽에는 이미 썼는데 입력 오프셋은 커밋되기 전에 애플리케이션이 크래시 나면 어떻게 될까?

 

2) 좀비 애플리케이션에 의해 발생하는 재처리

만약 애플리케이션이 카프카로부터 레코드 배치를 읽어온 직후 뭔가를 하기 전에 멈추거나, 카프카로의 연결이 끊어진다면 어떻게 될까? 앞에서 살펴본 상황과 비슷하게, 하트비트가 끊어지면서 애플리케이션은 죽은 것으로 간주될 것이며, 해당 컨슈머에 할당되어 있던 파티션들은 컨슈머 그룹 내 다른 컨슈머들에게 재할당될 것이다. 파티션을 재할당 받은 컨슈머가 레코드 배치를 다시 읽어서 처리하고, 출력 토픽에 결과를 쓰고 작업을 계속한다. 그 사이, 멈췄던 애플리케이션의 첫번재 인스턴스가 다시 작동할 수 있다. 

 

트랜젝션은 어떻게 '정확히 한번'을 보장하는가?

카프카 트랜젝션은 원자적 다수 파티션 쓰기 기능을 도입했다. 트랜젝션을 사용해서 워자적 다수 파티션 쓰기를 수행하려면 트랜잭션적 프로듀서를 사용해야한다. 카프카 브로커는 transactional.id에서 producer.id로의 대응 관계를 유지하고 있다가 만약 이미 있는 transactional.id 프로듀서가 initTransactions()를 다시 호출하면 새로운 랜덤 값이 아닌 이전에 쓰던 producer.id값을 할당해준다.

 

한 클러스터에서 다른 클러스터로 데이터 복제

하지만 이것이 트랜젝션의 운자성을 보장하지는 않는다. 만약 애플리케이션이 여러개의 레코드와 오프셋을 트랜젝션으로 쓰고, 미러메이커 2.0이 이 레코드들을 다른 카프카 클러스터에 복사한다면 복사 과정에서 트랜젝션 속성이나 보장같은 것은 유실된다. 데이터를 읽어왔는지 알 수 도 없고 보장할 수도 없는 것이다. 

 

트랜젝션 사용법

트랜젝션은 브로커 기능이다. 트랜젝션 기능을 사용하는 가장 일반적이고도 권장되는 방법은 카프카 스트림즈에서 exactly-once 보장을 활성화 하는 것이다. 

 

트랜젝션 ID와 펜싱

프로듀서가 사용할 트랜젝션 ID를 선택하는 것은 중요한 뿐 아니라 보기보다 조금 더 어려운 일이다. 트랜젝션 ID를 잘못 할당해 줄 경우 애플리케이션에 에러가 발생하거나 '정확히 한번' 보장을 준수할 수 없게 될 수 도 있다. 

 

버전 2.5까지, 펜싱을 보장하는 유링방 방법은 트랜젝션 ID를 파티션에 정적으로 대응시켜 보는 것 뿐이었다. 아파치 카프카 2.5에서 소개된 KIP-447 은 펜싱을 수행하는 두번째 방법, 즉 트랜젝션 ID와 컨슈머 그룹 메타데이터를 함께 사용하는 펜싱을 도입하였다. 

 

트랜잭션 성능

트랜젝션 초기화와 커밋 요청은 동기적으로 작동하기 때문에 성공적으로 완료되거나, 실패하거나, 타임아웃되거나 할 때까지 어떤 데이터도 전송되지 않는다. 그렇기 때문에 오버헤드는 더 증가한다. 프로듀서에 있어서 트랜젝션 오버헤드는 트랜젝션에 포함된 메시지의 수와는 무관하다는 점을 명심하라.