카프카 핵심 가이드 5 : 멱등성

8. '정확히 한 번' 의미 구조

---

 멱등적 프로듀서

 멱등성 의미 구조가 아닌 '최소 한 번' 의미 구조를 가지도록 프로듀서를 설정한다면, 프로듀서가 메세지 전송을 재시도함으로써 메세지가 최소 한 번 이상 도착할 수 있는 불확실성이 존재하게 된다. 이렇게 재시도는 메세지 중복으 발생시킬 수 있다.

 가장 고전적인 경우리면, 이런 경우가 있겠다.

1. 파티션 리더가 프로듀서로부터 레코드를 받아서 팔로워들에게 성공적으로 복제한다.
2. 프로듀서에게 응답을 보내기 전, 파티션 리더가 있는 브로커에 크래시가 발생한다.
3. 프로듀서 입장에서는 응답을 받지 못한 채 타임아웃이 발생하고, 메세지를 재전송한다.
4. 재전송된 메세지가 새 리더에 도착한다. 하지만 이 메세지는 이미 저장되어 있다(결과적으로, 중복이 발생한다).

 어떤 애플리케이션에서는 중복이 크게 문제되지 않는다. 하지만 다른 애플리케이션에서는 재고가 맞지 않는다든가, 재무재표가 잘못된다는가, 우산 하나를 주문한 고객에게 두 개를 배송해 준다든가하는 문제가 발생할 수 있다.

 카프카의 멱등적 프로듀서 기능은 자동으로 이러한 중복을 탐지하고 처리함으로써 이 문제를 해결한다.

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

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

 브로커가 예전에 받은 적이 있는 메세지를 받게 될 경우, 적절한 에러를 발생시킴으로써 중복 메세지를 거부한다. 이 에러는 프로듀서에 로깅도 되고 지푯값에도 반영이 되지만, 예외가 발생하는 것은 아니기 떄문에 사용자에게 정보를 보내지는 않는다. 프로듀서 클라이언트에서는 record-error-rate 지푯값을 확인함으로써 에러를 확인할 수 있다. 브로커의 경우 RequestMetrics 유형의 ErrosPerSec 지푯값에 기록된다(RequestMetrics에는 유형별 에러 수가 기록된다).

 만약 브로커가 예쌍보다 높은 시퀀스 넘버를 받게 된다면 어떻게 될까? 브로커는 2번 메세지 다음에 3번 메세지가 올 것을 예상하지만, 대신 27번 메세지가 오면 어떻게 될까? 이러한 경우, 브로커는 'out of order sequence number' 에러를 발생시킨다. 하지만 만약 트랜잭션 기능 없이 멱등적 프로듀서만 사용하고 있다면 이 에러는 무시해도 좋다.

 'out of sequnce number' 에러가 발생한 뒤에도 프로듀서가 정상 작동한다면, 이 에러는 보통 프로듀서와 브로커 사이에 메세지 유실이 있었음을 의미한다. 즉, 만약 브로커가 2번 메세지 뒤에 27번 메세지를 받았다면, 3번 메세지에서붙어 26번 메세지까지에 뭔가가 일어난 것이다. 만약 로그에 이러한 에러가 찍힌다면, 프로듀서와 브로커 설정을 재점검하고 프로듀서 설정이 고신뢰성을 위해 권장되는 값으로 잡혀 있는지, 아니면 언클린 리더 선출이 발생했는지의 여부를 확인해 볼 필요가 있다.

멱등적 프로듀서의 한계

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

 여러 개의 인스턴스를 띄우거나 하나의 인스턴스에서 여러 개의 프로듀서를 띄우는 애플리케이션들 역시 흔하다. 만약 이러한 프로듀서들 중 두 개가 동일한 메세지를 전송하려 시도할 경우, 멱등적 프로듀서는 중복을 잡아내지 못한다. 이러한 사례는 파일 디렉토리와 같은 원본에서 데이터를 읽어서 카프카로 쓰는 애플리케이션에서 꽤 흔하다. 만약 동일한 파일을 ㅇ릭어서 카프카에 레코드를 쓰는 두 개의 애플리케이션 인스턴스가 뜨게 되면 해당 파일의 레코드들은 2번 이상 쓰여지게 될 것이다.

멱등적 프로듀서는 프로듀서 자체의 재시도 메커니즘(프로듀서, 네트워크, 브로커 에러로 인해 발생하는)에 의한 중복만을 방지할 뿐 그 이상은 하지 않는다.

멱등적 프로듀서 사용법

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

• 프로듀서 ID를 받아오기 위해 프로듀서 시동 과정에서 API를 하나 더 호출한다.
• 전송되는 각각의 레코드 배치에는 프로듀서 ID와 배치 내 첫 메세지의 시퀀스 넘버가 포함된다(각 메세지의 시퀀스 넘버는 첫 메세지의 시퀀스 넘버에 변화량을 더하면 나온다). 이 새 필드들은 각 메세지 배치에 96 비트를 추가한다(프로듀서 ID는 long 타입이고 시퀀스 넘버는 integer 타입이다). 따라서 대부분의 경우 작업 부하에 어떠한 오버헤드도 되지 않는다.
• 브로커들은 모든 프로듀서 인스턴스에서 들어온 레코드 배치의 시퀀스 넘버를 검증해서 메세지 중복을 방지한다.
• 장애가 발생하더라도 각 파티션에 쓰여지는 메세지들의 순서는 보장된다. max.in.flight.requests.per.connection 설정값이 1보다 큰 값으로 잡혀도 마찬가지다(ㅅ는 기본값인 동시에 멱등적 프로듀서가 지원하는 가장 큰 값이다).

트랜잭션

 트랜잭션 기능은 카프카 스트림즈를 사용해서 개발된 애플리케이션에 정확성을 보장하기 위해 도입되었다. 스트림 처리 애플리케이션이 정확한 결과를 산출하도록 하기 위해, 각 입력 레코드는 정확히 한 번만 처리되어야 하며 그 처리 결과 역시 (장애 상황에서도) 정확히 한 번만 반영되어야 한다. 아파치 카프카의 트랜잭션 기능은 스트림 처리 애플리케이션이 정확한 결과를 산출할 수 있도록 한다. 이는 다시 개발자들이 정확성이 핵심 요구 조건인 활용 사례에서 스트림 처리 애플리케이션을 사용할 수 있도록 해준다.

 카프카의 트랜잭션 기능은 스트림 처리 애플리케이션을 위해 특별히 개발되었음을 염두에 둘 필요가 있다. 그런만큼 스트림 처리 애플리케이션의 기본 패턴인 '읽기-처리-쓰기' 패턴에서 사용하도록 개발되었다. 트랜잭션 기능은 이런 맥락에서 '정확히 한 번' 의미 구조를 보장할 수 있는 것이다(각 입력 레코드의 처리는 애플리케이션의 내부 상태가 업데이트되고 결과가 출력 토픽에 성공적으로 쓰여졌을 떄에야 완료된 것으로 간주된다).

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

 토픽에서 데이터를 읽어서, 처리하고, 결과를 다른 토픽에 쓴다. '정확히 한 번' 처리라 함은 이러한 읽기, 처리, 쓰기 작업이 원자적으로 이루어진다는 의미다. 읽어 온 원본 메세지의 오프셋이 커밋되고 결과가 성공적으로 쓰여지거나, 아니면 둘 다 안 일어나거나, 우리는 부분적인 결과(오프셋은 커밋되었는데 결과는 안 쓰여진다던가, 그 반대)가 결코 발생하지 않을 거라는 보장이 필요하다.

 이러한 작동을 지원하기 위해 카프카 트랜잭션은 원자적 다수 파티션 쓰기(atomic multipartition write) 기능을 도입했다. 이 아이디어는 오프셋을 커밋하는 것과 겨로가를 쓰는 것은 둘 다 파티션에 메세지를 쓰는 과정을 수반한다는 점에 착안한 것이다. 결과는 출력 토픽에, 오프셋은 _consumer_offsets 토픽에 쓰여진다는 점이 다를 뿐이다. 만약 우리가 트랜잭션을 시작해서 양쪽에 메세지를 쓰고, 둘 다 성공해서 커밋할 수 있다면(아니면 재시도하기 위해 중단할 수 있다면), 그 다음부터는 '정확히 한 번' 의미 구조가 알아서 해 준다.

 아래 그림은 읽어온 이벤트의 오프셋을 커밋함과 동시에 두 개의 파티션에 원자적 다수 파티션 쓰기를 수행하는 간단한 스트림 처리 애플리케이션을 보여준다.

트랜잭션적 프로듀서와 여러 파티션에 대한 원자적 쓰기

 트랜잭션을 사용해서 원자적 다수 파티션 쓰기를 수행하려면 트랜잭션적 프로듀서를 사용해야 한다. 트랜잭션적 프로듀서와 보통 프로듀서의 차이점이란 transactional.id 설정이 잡혀 있고 initTransactions()을 호출해서 초기화해주었다는 것뿐이다. 카프카 브로커에 의해 자동으로 생성되는 producer.id와는 달리 transactional.id 프로듀서 설정의 일부이며, 재시작을 하더라도 값이 유지된다. 사실 transactional.id의 주 용도가 재시작 후에도 동일한 프로듀서를 식별하는 것이다. 카프카 브로커는 transactional.id에서 producer.id로의 대응 관계를 유지하고 있다가 만약 이미 있는 transactional.id 프로듀서가 initTransactions()를 다시 호출하면 새로운 랜덤값이 아닌 이전에 쓰던 producer.id 값을 할당해 준다.

 

트랜잭션은 대부분 프로듀서 쪽 기능이다. 즉, 트랜잭션적 프로듀서를 생성하고, 트랜잭션을 시작하고, 다수의 파티션에 레코드를 쓰고, 이미 처리된 레코드들을 표시하기 위해 오프셋을 쓰고, 트랜잭션을 커밋하거나 중단하는 이 모든 작업이 프로듀서로부터 이루어진다. 하지만 이것으로 끝이 아니다. 트랜잭션 기능을 사용해서 쓰여진 레코드는 비록 결과적으로 중단된 트랜잭션에 속할지라도 다른 레코드들과 마찬가지로 파티션에 쓰여진다. 컨슈머에 올바른 격리 수준이 설정되어 있지 않을 경우, 우리가 기대하는 '정확히 한 번' 보장은 이루어지지 않을 것이다.

 우리는 isolation.level 설정값을 잡아줌으로써 트랜잭션 기능을 써서 쓰여진 메세지들을 읽어오는 방식을 제어할 수 있다. 이 값이 read_committed로 잡혀 있을 경우, 토픽들을 구독한 뒤 consumer.poll()을 호출하면 커밋된 트랜잭션에 속한 메세지나 처음부터 트랜잭션에 속하지 않는 메세지만 리턴된다(중단된 트랜잭션에 속한 메세지나 아직 진행중인 트랜잭션에 속하는 메세지는 리턴되지 않는 것이다). 하지만 isolation.level 설정을 기본값인 read_uncommitted로 두면 진행중이거나 중단된 트랜잭션에 속하는 것들 포함, 모든 레코드가 리턴된다. read_committed로 설정한다고 해서 특정 트랜잭션에 속한 모든 메세지가 리턴된다고 보장되는 되것도 아니다. 트랜잭션에 속하는 토픽의 일부만 구독했기 때문에 일부 메세지만 리턴받을 수도 있는 것이다. 또한, 트랜잭션이 언제 시작되고 끝날지, 어느 메세지가 어느 트랜잭션에 속하는지에 대해서 애플리케이션은 알 수 없다.

 메세지의 읽기 순서를 보장하기 위해 read_committed 모드에서는 아직 진행중인 트랜잭션이 처음으로 시작된 시점(Last Stable Offset, LOS) 이후에 쓰여진 메세지는 리턴되지 않는다. 이 메세지들은 트랜잭션이 프로듀서에 의해 커밋되거나 중단될 때까지, 혹은 transaction.timeout.ms 설정값(기본값: 15분)만큼 시간이 지나 브로커가 트랜잭션을 중단시킬 때까지 보류된다. 이렇게 트랜잭션이 오랫동안 닫히지 않고 있으면 컨슈머들이 지체되면서 종단 지연이 길어진다.

read_committed 모드로 작동 중인 컨슈머는 read_uncommitted 모드로 작동하는 컨슈머(기본값)보다 약간 더 뒤에 있는 메세지를 읽는다.

 스트림 처리 애플리케이션은 입력 토픽이 트랜잭션 없이 쓰여졌을 경우에도 '정확히 한 번' 출력을 보장한다. 원자적 다수 파티션 쓰기 기능은 만약 출력 레코드가 출력 토픽에 커밋되었을 경우, 입력 레코드의 오프셋 역시 해당 컨슈머에 대해 커밋되는 것을 보장한다. 결과적으로 입력 레코드는 다시 처리되지 않는다.

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

 트랜잭션 기능과 관련해서 자주 하는 실수가 두 가지 있다. 하나는 '정확히 한 번 보장'이 카프카에 대한 쓰기 이외의 작동에서도 보장된다고 착각하는 것이다. 또 하나는 컨슈머가 항상 전체 트랜잭션을 읽어 온다고 (트랜잭션 간의 경계에 대해 알고 있다고) 가정하는 것이다.

 다음은 카프카의 트랜잭션 기능이 '정확히 한 번' 보장에 도움이 되지 않는 몇 가지 경우다.

1. 스트림 처리에 있어서의 부수 효과(side effect)

 스트림 처리 애플리케이션의 처리 단계에 사용자에 이메일을 보내는 작업이 포함되어 있다고 해 보자. 이 애플리케이션에서 '정확히 한 번' 의미 구조를 활성화한다고 해서 이메일이 한 번만 발송되는 것은 아니다. 이 기능은 카프카에 쓰여지는 레코드에만 적용된다. 레코드 중복을 방지하기 위해 시퀀스 넘버를 사용하는 것이나 트랜잭션을 중단 혹은 취소하기 위해 마커를 사용하는 것은 카프카 안에서만 작동하는 것이지, 이메일 발송을 취소시킬 수 있는 것은 아니기 때문이다. 이는 스트림 처리 애플리케이션 안에서 외부 효과를 일으키는 어떠한 작업(REST API 호출, 파일 쓰기 등)에도 해당된다.

2. 카프카 토픽에서 읽어서 데이터베이스에 쓰는 경우

 이 경우 애플리케이션은 카프카가 아닌 외부 데이터베이스에 결과물을 쓴다. 여기서는 프로듀서가 사용되지 않는다. 즉, 레코드는 JDBC와 같은 데이터베이스 드라이버를 통해 데이터베이스에 쓰여지고, 오프셋은 컨슈머에 의해 카프카에 커밋된다. 하나의 트랜잭션에서 외부 데이터베이스에는 결과를 쓰고 카프카에는 오프셋을 커밋할 수 있도록 해주는 메커니즘 같은 건 없다. 대신, 4장에서 설명한 것과 같이 오프셋을 데이터베이스에서 저장하도록 할 수는 있다. 이렇게 하면 하나의 트랜잭션에서 데이터와 오프셋을 동시에 데이터베이스에 커밋할 수 있다(이 부분은 카프카가 아닌, 데이터베이스의 트랜잭션 보장에 달렸다).

3. 데이터베이스에서 읽어서, 카프카에 쓰고, 여기서 다시 다른 데이터베이스에 쓰는 경우

하나의 앱에서 데이터베이스 데이터를 읽고, 트랜잭션을 구분하고, 카프카에 레코드를 쓰고, 여기서 다시 다른 데이터베이스에 레코드를 쓰고, 그 와중에도 원본 데이터베이스의 원래 트랜잭션을 관리할 수 있는 앱을 개발할 수 있다면 솔깃할 것이다.

 불행히도 카프카 트랜잭션은 이러한 종류의 종단 보장(end-to-end guarantee)에 필요한 기능을 가지고 있지 않다. 하나의 트랜잭션 안에서 레코드와 오프셋을 함께 커밋하는 문제 외에도 또 다른 문제가 있기 떄문이다. 카프카 컨슈머의 read_committed 보장은 데이터베이스 트랜잭션을 보존하기엔 너무 약하다. 컨슈머가 아직 커밋되지 않은 레코드를 볼 수 없는 건 사실이지만, 일부 토픽에서 랙이 발생했을 수도 있는 만큼 이미 커밋된 트랜잭션의 레코드를 모두 봤을 거라는 보장 또한 없기 떄문이다. 트랜잭션의 경계를 알 수 있는 방법 역시 없기 때문에 언제 트랜잭션이 시작되었는지, 끝났는지, 레코드 중 어느 정도를 읽었는지도 알 수 없다.

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

이 경우는 좀 더 미묘하다. 하나의 카프카 클러스터에서 다른 클러스터로 데이터를 복사할 때 '정확히 한 번'을 보장할 수 있다. 어떻게 이것이 가능한지에 대한 상세한 내용은 미러메이커 2.0에 '정확히 한 번' 기능을 추가하는 KIP-656(https://cwiki.apache.org/confluence/display/KAFKA/KIP-656%3A+MirrorMaker2+Exactly-once+Semantics)에서 볼 수 있다. 이 제안은 원본 클러스터의 각 레코드가 대상 클러스터에 정확히 한 번 복사될 것을 보장하는 내용을 담고있다.

 하지만 이것이 트랜잭션의 원자성을 보장하지는 않는다. 만약 애플리케이션이 여러 개의 레코드와 오프셋을 트랜잭션적으로 쓰고, 미러메이커 2.0이 이 레코드들을 다른 카프카 클러스터에 복사한다면, 복사 과정에서 트랜잭션 속성이나 보장 같은 것은 유실된다. 마찬가지로 이 정보들은 카프카의 데이터를 관계형 데이터베이스에 복사할 떄도 유실된다. 카프카에서 데이터를 읽어오는 컨슈머 입장에서는 트랜잭션의 모든 데이터를 읽어왔는지 알 수도 없고 보장할 수도 없는 것이다. 예를 들어서, 토픽의 일부만 구독했을 경우 전체 트랜잭션의 일부만 복사할 수 있다.

5. 발행/구독 패턴

좀 더 미묘한 경우다. 읽기-처리-쓰기 패턴의 맥락에서 '정확히 한 번'이 어떠한 의미인지를 논의하였다. 하지만 발행/구독 패턴은 매우 일반적인 활용 사례다. 발행/구독 패턴에 트랜잭션을 사용할 경우 몇 가지 보장되는 것이 있기는 하다. 즉, read_committed 모드가 설정된 컨슈머들은 중단된 트랜잭션에 속한 레코드들을 보지 못할 것이다. 하지만 이러한 보장은 '정확히 한 번'에 미치지 못한다. 오프셋 커밋 로직에 따라 컨슈머들은 메세지를 한 번 이상 처리할 수 있다.

 이 경우 카프카가 보장하는 것은 JMS 트랜잭션에서 보장하는 것과 비슷하지만, 커밋되지 않은 트랜잭션들이 보이지 않도록 컨슈머들에 read_committed 설정이 되어 있어야 한다는 전제 조건이 붙는다. JMS 브로커들은 모든 컨슈머에게 커밋되지 않은 트랜잭션의 레코드를 주지 않는다.

 메세지를 쓰고 나서 커밋하기 전에 다른 애플리케이션이 응답하기를 기다리는 패턴은 반드시 피해야 한다. 다른 애플리케이션은 트랜잭션이 커밋될 때까지 메세지를 받지 못할 것이기 때문에 결과적으로 데드락이 발생한다.

트랜잭션 사용법

 트랜잭션은 브로커 기능이기도 하며, 카프카 프로토콜의 일부인 만큼 여러 클라이언트들이 트랜잭션을 지원한다.

 트랜잭션 기능을 사용하는 가장 일반적이고도 권장되는 방법은 카프카 스트림즈에서 Exactly-once 보장을 활성화하는 것이다. 이렇게 하면 트랜잭션 기능을 직접적으로 사용할 일은 전혀 없지만, 카프카 스트림즈가 대신 해당 기능을 사용해서 우리가 필요로 하는 보장을 제공해 준다. 트랜잭션 기능 자체가 이런 활용 사례를 염두해 두고 설계된 만큼, 카프카 스트림즈를 통한 간접 사용은 가장 쉬우면서도 예상했던 결과를 얻을 가능성이 높은 방법이다. 카프카 스트림즈 애플리케이션에서 '정확히 한 번' 보장 기능을 활성화하려면 그냥 processing.guarantee 설정을 exactly_once이나 exactly_once_beta로 잡아주면 된다. 그걸로 끝이다.

트랜잭션 성능

 트랜잭션은 프로듀서에 약간의 오버헤드를 발생시킨다. 프로듀서를 생성해서 사용하는 동안 트랜잭션 ID 등록 요청은 단 한번 발생한다. 트랜잭션의 일부로서 파티션들을 등록하는 추가적인 호출은 각 트랜잭션에 있어서 파티션 별로 최대 한 번씩만 이루어진다. 그리고 각 트랜잭션이 커밋 요청을 전송하면, 파티션마다 커밋 마커가 추가된. 트랜잭션 초기화와 커밋 요청은 동기적으로 작동하기 때문에 성공적으로 완료되거나, 실패하거나, 타임아웃되거나 할 때까지 어떤 데이터도 전송되지 않는다. 그렇기 때문에 오버헤드는 더 증가 한다.

 프로듀서에 있어서 트랜잭션 오버헤드는 트랜잭션에 포함된 메세지의 수와는 무관하다는 점을 명심하라. 그렇기 때문에 트랜잭션마다 많은 수의 메세지를 집어넣는 쪽이 상대적으로 오버헤드가 적을뿐 아니라 동기적으로 실행되는 단계의 수도 줄어든다. 결과적으로 전체 처리량은 증가한다.

 컨슈머 쪽에 대해서는 커밋 마커를 읽어오는 작업에 관련해서 약간의 오버헤드가 있다. 트랜잭션 기능이 컨슈머 성능에 미치는 핵심적인 영향은 read_committed 모드 컨슈머에서는 아직 완료되지 않은 트랜잭션의 레코드들이 리턴되지 않는다는 것이다. 트랜잭션 커밋 사이의 간격이 길어질수록 컨슈머는 메세지가 리턴될 때까지 더 오랫동안 기다려야 할 것이다. 결과적으로 종단 지연 역시 그만큼 길어진다.

 하지만, 컨슈머는 아직 완료되지 않은 트랜잭션에 속하는 메세지들을 버퍼링할 필요가 없다는 점 역시 유념하길 바란다. 브로커는 컨슈머가 보낸 읽기 요청을 받는다고 해서 이 메세지들을 리턴하지 않는다. 따라서 트랜잭션 데이터를 읽을 때 컨슈머 쪽에 추가적인 작업은 없다. 자연히 처리량이 줄어들지도 않는다.