[Kafka] 실전 카프카 5장

5.1 파티셔너

프로듀서가 메시지를 보낼 때 해당 토픽의 어느 파티션으로 보내야 할지 결정해야 하고, 이때 파티셔너를 사용합니다. 기본적으로 메시지의 키를 해싱하여 파티션을 결정하는 알고리즘을 사용합니다. 따라서 동일한 메시지는 같은 파티션으로 전송됩니다. 하지만 처리량을 높이기 위해 파티션 수를 조절하는 경우, 같은 메시지가 다른 파티션으로 전송될 가능성이 있으므로 되도록 파티션 수를 변경하지 않는 것을 권장합니다.

5.1.1 라운드 로빈 전략

프로듀서가 전송하는 메시지의 키값은 필수가 아니므로 null이 될 수도 있습니다. 만약 키값이 null이면 라운드 로빈 알고리즘을 사용해 토픽의 파티션으로 메시지를 랜덤 전송합니다.

파티셔너를 거친 후 메시지는 배치 처리를 위해 프로듀서 버퍼 메모리 영역에서 대기를 하는데, 그 과정에서 라운드 로빈의 효율이 떨어질 수 있습니다.

3개의 버퍼 메모리가 채워지지 않으면 배치 전송을 할 수 없어 대기합니다. 이 경우 버퍼에서의 최대 대기 시간을 조절할 수도 있지만 배치 전송을 하는 목적을 이루지 못할 수도 있습니다. 이런 단점을 보완하기 위해 스티키 파티셔닝 전략이 등장했습니다.

5.1.2 스티키 파티셔닝 전략

하나의 파티션에 메시지 수를 먼저 채워서 카프카로 빠르게 배치 전송하는 전략입니다. 스티키 파티셔닝 전략으로 약 30% 이상 지연시간이 감소했고 프로듀서의 CPU 사용률도 줄어드는 효과를 얻었다고 합니다.

출처: https://www.confluent.io/blog/apache-kafka-producer-improvements-sticky-partitioner/

5.2 프로듀서의 배치

프로듀서는 배치 전송을 위해 아래 옵션을 제공합니다.

• buffer.memory : 카프카로 메시지를 전송하기 위해 담아두는 프로듀서의 버퍼 메모리. 기본값은 32MB
• batch.size : 배치 전송을 위해 메시지를 묶는 단위. 기본값은 16KB
• linger.ms : 배치 전송을 위해 버퍼 메모리에서 대기하는 메시지들의 최대 대기시간. 기본값은 0ms

배치 전송은 불필요한 I/O를 줄일 수 있지만 전송에 지연이 발생할 수 있으므로 전략적으로 결정해야 합니다.

처리량을 높이려면 batch.size와 linger.ms 값을 크게 설정해야 하고, 지연 없는 전송이 목표라면 batch.size와 linger.ms 값을 작게 설정해야 합니다.

배치 전송을 사용할 때 주의할 점은 buffer.memory의 크기를 batch.size 값보다는 크게 설정해야 하는 것입니다.

5.3 중복 없는 전송

메시지 시스템들의 전송 방식에는 3가지가 있습니다.

• at-least-once : 적어도 한 번 전송
• at-most-once : 최대 한 번 전송
• exactly-once : 정확히 한 번 전송

적어도 한 번 전송

브로커로부터 메시지B에 대한 ACK를 받지 못했다면 두 가지 상황으로 나뉩니다.

1. 브로커는 메시지를 받았지만 ACK 메시지가 네트워크 오류로 프로듀서에게 전달되지 못한 상황
2. 브로커가 메시지를 못받은 상황

1번 상황이라면 브로커에는 메시지B가 중복으로 적재될 가능성이 있습니다. 그리고 카프카는 기본적으로 적어도 한 번 전송 방식을 기반으로 동작합니다.

최대 한 번 전송

프로듀서는 브로커가 메시지B를 받았다고 가정하고 다음 메시지를 보냅니다. 이 전송 방식은 메시지가 손실되더라도 높은 처리량을 필요로 하는 상황에서 사용됩니다.(ex. 대량 로그 수집, IoT 센서 데이터 수집 등)

중복 없는 전송

프로듀서를 구분할 수 있도록 PID가 지정되며 메시지를 보낼 때 시퀀스 번호를 메모리에 저장합니다. 브로커는 메시지를 받은 후 PID와 시퀀스 번호로 이미 저장된 메시지가 있다면 ACK만 응답하여 메시지를 중복 저장하지 않습니다.

중복 없는 전송을 위한 프로듀서 설정 옵션은 아래와 같습니다.

• enable.idempotence=true : 기본값은 false이며 중복 없는 전송을 허용할지 결정하는 옵션
• max.in.flight.requests.per.connection=1~5 : ACK를 받지 않은 상태에서 하나의 커넥션에서 보낼 수 있는 최대 요청 수. 기본값은 5
• acks=all : 프로듀서 acks와 관련된 옵션. 기본값은 1
• retries=5 : ACK를 받지 못한 경우 재시도를 해야 하므로 0보다 큰 값으로 설정

5.4 정확히 한 번 전송

앞서 멱등성 옵션을 사용하여 중복 전송을 방지할 수 있다고 했지만, 이것이 정확히 한 번 전송한다는 의미는 아닙니다. 카프카에서 정확히 한 번 전송은 트랜잭션과 같은 처리를 의미하며, 중복 없는 전송은 정확히 한 번 전송의 일부 기능이라고 볼 수 있습니다.

5.4.1 디자인

프로듀서가 카프카로 정확히 한 번 메시지를 전송할 때 원자적으로 처리되어 성공하거나 실패합니다. 이러한 전송을 위해 카프카에는 트랜잭션 코디네이터가 존재합니다. 카프카는 내부 토픽 _transaction_state에 트랜잭션 로그를 저장합니다.

5.4.3 단계별 동작

정확히 한 번 전송을 위해서는 트랜잭션 API를 이용하며, 가장 먼저 트랜잭션 코디네이터를 찾아야 합니다.

프로듀서는 브로커에게 FindCoordinatorRequest를 보내서 트랜잭션 코디네이터의 위치를 찾습니다. 트랜잭션 코디네이터의 주 역할은 프로듀서 PID와 transactional.id를 맵핑하여 트랜잭션을 관리하는 것입니다. 만약 트랜잭션 코디네이터를 찾지 못하면 신규 트랜잭션 코디네이터가 생성됩니다.

프로듀서는 initTransactions() 메서드를 이용해 트랜잭션 전송을 위한 InitPidRequest를 트랜잭션 코디네이터로 보냅니다. 이때 TID(transactional.id)가 설정된 경우, InitPidRequest와 함께 TID를 전송하고 트랜잭션 코디네이터는 TID와 PID를 맵핑합니다.

이후 프로듀서는 beginTransaction() 메서드를 이용해 트랜잭션의 시작을 알립니다. 프로듀서는 트랜잭션이 시작되었다고 기록하지만 트랜잭션 코디네이터 관점에서는 첫 메시지가 전송될 때까지 트랜잭션이 시작되지 않은 것으로 간주합니다.

다음으로는 트랜잭션 상태 추가 동작입니다. 트랜잭션 코디네이터는 전체 트랜잭션을 관리하며, 각 트랜잭션 상태의 내용을 기록하는 일도 중요합니다. 프로듀서는 토픽 파티션 정보를 트랜잭션 코디네이터에게 전달하고, 코디네이터는 해당 정보를 트랜잭션 로그에 기록하고 트랜잭션의 현재 상태를 Ongoing으로 표시합니다. 또한 트랜잭션 로그에 추가되는 첫 번째 파티션이라면 타이머를 시작하여 1분 동안 트랜잭션 상태에 변화가 없는 경우 트랜잭션을 실패로 처리합니다.

프로듀서는 대상 토픽으로 메시지를 전송합니다.

메시지 전송을 종료한 프로듀서는 commitTransaction() 또는 abortTransaction() 메서드를 호출하여 트랜잭션 완료됨을 트랜잭션 코디네이터에게 알립니다. 이후 코디네이터는 두 단계의 커밋 과정을 시작하며 첫 번째 단계로 트랜잭션 로그에 해당 트랜잭션에 대한 PrepareCommit 또는 PrepareAbort를 기록합니다.

두 번째 단계에서 트랜잭션 로그에 기록된 토픽의 파티션에 트랜잭션 커밋 표시를 기록합니다. 이때 기록하는 메시지가 컨트롤 메시지입니다.

현재 파티션0의 메시지 오프셋이 1이라면, 컨트롤 메시지로 인해 오프셋이 2로 증가합니다. 컨트롤 메시지는 컨슈머에게 해당 PID의 메시지가 제대로 전송됐는지 나타내는 용도로도 사용됩니다. 따라서 커밋되지 않은 메시지는 컨슈머에게 반환되지 않습니다.

트랜잭션 코디네이터는 트랜잭션 로그에 완료됨을 기록하고 프로듀서에게 알리고 트랜잭션 처리를 마무리합니다.