MinJunKweon
commented
2 years ago 파티셔너(Partitioner)
- 카프카 토픽의 성능 향상을 위해 병렬처리가 가능하도록 파티션으로 나뉨
- 최소 하나 이상의 파티션으로 구성됨
- 프로듀서가 카프카로 전송한 메시지는 토픽 내 각 파티션의 로그 세그먼트에 저장됨
- 토픽으로 메시지를 보낼 때, 해당 토픽의 어느 파티션으로 메시지를 보내야할 지를 결정할 때 사용되는 것이 파티셔너(Partitioner)
- 메시지가 저장될 파티션을 결정하는 알고리즘은 메시지(레코드)의 키를 해시(hash) 처리하는 방식
- 메시지의 키가 같으면 메시지들은 모두 같은 파티션으로 전송됨
- 예상치 못한 많은 양의 메시지가 카프카로 인입되는 경우, 파티션을 늘릴 수 있는 기능을 제공함
- 파티션 수가 변경되면 키와 매핑된 해시 테이블도 변경됨
- 파티션의 수를 늘리면 이전에 사용하던 키가 다른 파티션으로 전송될 수 있음
라운드 로빈(Round-Robin) 전략
- 프로듀서 메시지에서 키값은 필수 값이 아니다.
- 키 값이
null이 되면 기본값인 라운드 로빈(Round-Robin) 알고리즘을 사용해 파티션들로 랜덤 전송된다. - 배치처리를 위해서 메시지 전송 전에 파티셔너를 거친 후 프로듀서의 버퍼 메모리 영역에 잠시 대기한 후 카프카로 전송됨
- 파티션별 최소 레코드 수 기준을 충족하지 않으면 전송이 지연된다. (하지만, 특정 시간을 초과하면 즉시 전송하도록 설정할 수 있다. 그래도 비효율적이지만)
스티키 파티셔닝(Sticky Partitioning) 전략
- 2.4+ 버전부터 스티키 파티셔닝 전략을 사용함
- 하나의 파티션에 레코드 수를 먼저 채워서 카프카로 빠르게 전송하는 전략
- 파티션 0에 최소 레코드 수 3개가 모두 채워져서 전송이 바로 된다. 지연이 없다.
프로듀서 배치(Batch)
| 옵션명 | 설명 | 기본값 |
|---|---|---|
| buffer.memory | 카프카로 전송하기 위해 담아두는 버퍼 메모리 모든 파티션의 레코드의 크기를 다 더한 값을 계산함 ※ 서버로 전달될 수 있는 것보다 빠르게 보낸다면, 프로듀서는 max.block.ms초(기본값 1분)까지 기다렸다가 가용 공간이 없으면 TimeoutException을 발생시킨다 |
32MB |
| batch.size | 배치 전송을 위한 메시지들을 묶는 단위를 설정하는 배치 크기 각 파티션의 배치 전송용 버퍼 메모리만 계산함 이 값만큼 메시지가 모이면 데이터를 한번에 전송함 |
16KB |
| linger.ms | 버퍼 메모리에서 대기하는 메시지들의 최대 대기시간 버퍼에 데이터가 많이 쌓이지 않았더라도 이 값만큼 시간이 지나면 무조건 브로커로 데이터를 전송함 |
0 (배치전송 사용X) |
| compression.type | 프로듀서에 의해 생성되는 모든 데이터에 대한 압축 종류 gzip, snappy, lz4, zstd를 쓸 수 있다 |
none |
배치처리 장점
- 여러 개의 메시지를 적은 요청/응답으로 전송하기 때문에 효율적임
compression.type을 설정하면 좀 더 적은 메모리로 효율적인 버퍼 메모리를 사용할 수 있다.
배치처리 단점
- 배치처리로 인해 지연이 발생할 수 있으므로 상황에 따라 설정값을 취사선택 해야함
- 처리량(Throughput)과 지연시간(Latency)는 Trade-off 관계다.
중복 없는 전송
- 중복 없는 메시지 전송은 매우 중요한 기능
- 중복 결제나 중복 배송 등 실생활에서 중복 메시지로 인한 문제가 발생할 여지가 많기 때문
적어도 한 번 전송(at-least-once)
- 프로듀서에서 메시지를 전송하면 브로커가 수신 후 프로듀서로 ACK를 전송함
- ACK가 전송 중에 누락되는 경우, 프로듀서는 같은 메시지를 다시 한번 전송함
- 프로듀서가 ACK를 수신하기 전까지는 전송이 되지 않았다고 판단하기 때문에 매번 재전송함
- 적어도 한 번 전송이 보장되지만, 일부 중복된 메시지가 존재할 수 있음
최대 한 번 전송(at-most-once)
- ACK가 누락되더라도 프로듀서는 메시지를 잘 전달했다고 가정함
- 재전송을 하지 않기 때문에 브로커에서 수신하는 메시지가 유실될 수 있음
- 높은 처리량이 필요한 대량의 로그 수집 같은 곳에서 사용하곤 하는 방식
중복 없는 전송
- 메시지에 프로듀서 ID - 메시지 번호 쌍을 헤더에 포함해서 함께 전송함
- ACK 메시지에도 프로듀서 ID - 메시지 번호 쌍을 포함해서 보낸다
- ACK가 유실되면 재전송을 하는 것까지는 적어도 한 번 전송과 똑같으나, 브로커는 헤더를 통해 같은 메시지라는 것을 파악할 수 있으므로 중복없는 전송이 가능하다
- 프로듀서 ID는 내부에서만 사용되므로 노출되지 않음. 시퀀스 번호는 0번부터 시작해 순차적으로 증가함
- 중복없는 전송은 오버헤드가 존재하지만, 기존 대비 최대 약 20% 정도만 성능이 감소했다고 함
- 프로듀서 ID와 시퀀스 번호는 브로커 파일에
snapshot확장자 파일에 저장됨
중복없는 전송과 관련된 프로듀서 옵션
| 옵션명 | 설명 | 기본값 |
|---|---|---|
| enable.idempotence | 프로듀서가 중복 없는 전송을 허용할지 결정하는 옵션 true로 설정되어야 아래 설정 값들을 사용할 수 있다. |
true |
| max.in.flight.requests.per.connection | ACK를 받지 않은 상태에서 하나의 커넥션에서 보낼 수 있는 최대 요청 수 | 5 |
| acks | all, -1 : ISR 전체에 대해 ack를 대기하는 옵션0 : 메시지 전송후 ACK를 대기하지 않는 옵션1 : 리더 브로커에만 ACK를 기다리는 옵션 |
all |
| retries | ACK를 받지 못한 경우 재시도하는 횟수 | 0 |
정확히 한 번 전송(exactly-once)
- 중복 없는 전송 ≠ 정확히 한 번 전송
- 중복 없는 전송 ⊂ 정확히 한번 전송
- 카프카에서 정확히 한 번 처리를 담당하는 프로세스 : 트랜잭션 API
- 트랜잭션은 Atomicity, Consistency, Isolation, Durability 이렇게 4가지 성질을 가짐
정확히 한 번 전송 디자인
- 프로듀서의 전송을 위해 서버 측에 Transaction Coordinator가 존재함
- 프로듀서에 의해 전송된 메시지의 커밋 또는 중단 등을 표시함
- 트랜잭션 코디네이터는 트랜잭션 로그를 카프카 내부 토픽인
__transaction_state에 저장함 __transaction_state토픽의 Partition 수와 Replication factor는 브로커 설정을 통해 관리자가 설정할 수 있음transaction.state.log.num.partitions(기본값 : 50)transaction.state.log.replication.factor(기본값 : 3)
__transaction_state토픽을 트랜잭션 로그라고 불림- 트랜잭션 로그 메시지 예시
peter-transaction-01::TransactionMetadata(transationalId=peter-transaction-01, producerId=48000, producerEpoch=0, txnTimeoutMs=60000, state=Empty, pendingState=None, topicPartitions=Set(), txnStartTimestamp=-1, txnLastUpdateTimestamp=1604742193026)
peter-transaction-01::TransactionMetadata(transationalId=peter-transaction-01, producerId=48000, producerEpoch=0, txnTimeoutMs=60000, state=Ongoing, pendingState=None, topicPartitions=Set(peter-test05-0), txnStartTimestamp=-1, txnLastUpdateTimestamp=1604742193106)
peter-transaction-01::TransactionMetadata(transationalId=peter-transaction-01, producerId=48000, producerEpoch=0, txnTimeoutMs=60000, state=PrepareCommit, pendingState=None, topicPartitions=Set(peter-test05-0), txnStartTimestamp=1604742193106, txnLastUpdateTimestamp=1604742193195)
peter-transaction-01::TransactionMetadata(transationalId=peter-transaction-01, producerId=48000, producerEpoch=0, txnTimeoutMs=60000, state=Empty, pendingState=CompleteCommit, topicPartitions=Set(), txnStartTimestamp=1604742193106, txnLastUpdateTimestamp=1604742193204)- 실제 토픽 파티션에 트랜잭션이 완료되거나 실패했음을 알리는 메시지는 컨트롤 메시지라고 불림
- 커밋이 정상인지 실패한 것인지 식별되는 정보로 불리는 특별한 타임의 메시지
- 컨트롤 메시지에는 메시지의 내용은 포함되지 않고, 애플리케이션들에게 노출되지 않음
정확히 한 번 전송을 위한 필수 설정 옵션
| 옵션명 | 설명 | 기본값 |
|---|---|---|
| enable.idempotence | 프로듀서가 중복 없는 전송을 허용할지 결정하는 옵션 true로 설정되어야 아래 설정 값들을 사용할 수 있다. |
true |
| max.in.flight.requests.per.connection | ACK를 받지 않은 상태에서 하나의 커넥션에서 보낼 수 있는 최대 요청 수 | 5 |
| acks | all, -1 : ISR 전체에 대해 ack를 대기하는 옵션0 : 메시지 전송후 ACK를 대기하지 않는 옵션1 : 리더 브로커에만 ACK를 기다리는 옵션 |
all |
| retries | ACK를 받지 못한 경우 재시도하는 횟수 | 0 |
| transactional.id | 프로듀서 프로세스마다 설정하는 고유한 아이디 트랜잭션을 구분하기 위한 프로듀서 아이디 (ex. peter-transaction-01) |
null |
- Java
kafka-client코드 상에서Producer인터페이스에 정의된 트랜잭션 메소드를 쓸 수 있음 initTransactions()메소드로 트랜잭션을 초기화beginTransaction()메소드로 트랜잭션 시작을 명시abortTransaction()메소드로 트랜잭션을 중단하도록 명시commitTransaction()메소드로 트랜잭션을 커밋
정확히 한 번 전송의 단계별 동작
- 프로듀서 —> 브로커
FindCoordinatorRequest전송하여 트랜잭션 코디네이터의 위치를 찾음- 만약 존재하지 않는다면 신규 트랜잭션 코디네이터가 생성됨
- 트랜잭션 코디네이터가 PID(Producer ID)와 TID(
transactional.id)를 매핑하고 트랜잭션 전체를 관리함 __trasnaction_state토픽의 파티션 번호는 TID를 기반으로 해시하여 결정됨- 이 파티션 리더가 있는 브로커가 트랜잭션 코디네이터의 브로커로 최종 선정됨
initTransactions()메소드가 호출될 때,InitPidRequest를 트랜잭션 코디네이터로 보냄. 이 때, TID도 함께 전송함- 트랜잭션 코디네이터는 TID와 PID를 매핑하고
__transaction_state트랜잭션 로그에 기록 - PID epoch를 하나 올리고 이전의 동일한 PID와 이전 에포크에 대한 쓰기 요청을 무시함
- 에포크를 쓰는 이유는 신뢰성 있는 메시지 전송을 위함 (오래된 메시지를 무시하기 위함)
- 트랜잭션 코디네이터는 TID와 PID를 매핑하고
beginTransaction()메소드를 사용해 새로운 트랜잭션의 시작을 알림- 트랜잭션 코디네이터 관점에서는 트랜잭션 시작 후 첫 번째 레코드 전송이 될 때 트랜잭션이 시작된 것으로 간주
- 프로듀서는 토픽 파티션 정보를 트랜잭션 코디네이터에 전달함
- 트랜잭션 코디네이터는 각 트랜잭션 상태의 내용을 기록함 (
현재상태 : Ongoing으로 표시) - 기본값으로 1분 동안 트랜잭션 상태에 대한 업데이트가 없다면 실패로 간주함
- 프로듀서가 브로커로 메시지 전송을 함
- 메시지 전송 대상 브로커는 트랜잭션 코디네이터와 다른 브로커일 수 있음
- 메시지에는 PID, 에포크, 시퀀스 번호가 함께 포함되어 전송됨
- 트랜잭션 종료를 위해
commitTransaction()혹은abortTransaction()중 반드시 1개를 호출해야함- 트랜잭션 코디네이터에 트랜잭션 종료를 알게되고 해당 트랜잭션에
PrepareCommit혹은PrepareAbort를 기록함
- 트랜잭션 코디네이터에 트랜잭션 종료를 알게되고 해당 트랜잭션에
- 트랜잭션 코디네이터는
PrepareCommit이나PrepareAbort에 대해서 파티션에 트랜잭션을 커밋이 되었음을 알림 (컨트롤 메시지)- 애플리케이션은 이 메시지를 표시하지 않지만 실제로 메시지는 저장되기 때문에 오프셋이 증가함
- 컨슈머는 이 메시지를 통해 메시지가 제대로 전송되었는지를 판단할 수 있음
- 트랜잭션 커밋이 끝나지 않은 메시지는 컨슈머에게 반환하지 않음
- 오프셋 순서 보장을 위해 트랜잭션 성공 또는 실패를 나타내는 LSO(Last Stable Offset)라는 오프셋을 사용함
- 트랜잭션이 완료되면 트랜잭션 코디네이터는 완료됨(Committed)라고 트랜잭션 로그에 기록함
- 트랜잭션이 완료되면 프로듀서에 완료메시지를 전달하고 모든 처리가 마무리됨
- 트랜잭션을 사용하는
read_committed설정을 쓰는 컨슈머들은 커밋된 메시지들만 읽을 수 있음
LOG-INFO
정리
파티셔너
프로듀서의 배치
중복 없는 전송
at-least-once와most-once, 그리고exactly-once총 세 가지 방식을 사용한다.at-least-once
most-once
exactly-once
카프카의 트랜잭션