카프카(kafka)에서 프로듀서(Producer)의 기본 역할은 메시지들을 Kafka Topic으로 전송하는 것입니다.
메시지 전송 방식에는 적어도 한번 전송(at-least-once), 최대 한 번 전송(at-most-once), 정확히 한번 전송(exactly-once)이 있으며 전송방식들의 차이들을 확인해 보도록 하겠습니다.
Producer가 전송하려는 메시지들은 Producer의 send() 메서드를 통해 시리얼라이저, 파티셔너를 거쳐 kafka로 전송됩니다.
마지막에 실습이 있는데 실습 환경을 미구성하셨다면 이전 포스팅을 참조해 주시면 감사합니다.
https://colevelup.tistory.com/17
[Kafka] kafka cluster 실습 환경 구축
kafka 관련 포스팅을 하기 이전 kafka cluster(카프카 클러스터) 를 구축 하여에 실습환경을 구성해보도록 하겠습니다. 컨테이너(container) 환경에서 실습환경은 진행됩니다. 가상화 기술과 Docker, 대량
colevelup.tistory.com
파티셔너(partitioner)
topic은 파티션으로 나뉘고 최소 하나 또는 둘 이상의 partition으로 구성됩니다.
producer가 카프카로 전송한 메시지는 topic 내 로그 세그먼트에 저장됩니다.
producer는 topic 내에 존재하는 여러 가지 partition 중에
어디로 메시지를 보내야 할지 결정해야 하며 이때 사용하는 것이 partitioner입니다.
어느 partition으로 보낼지 정하는 기본적인 알고리즘은 메시지의 키를 hash처리해서 partition을 구하는 방식을 사용합니다.
메시지의 키값이 같다면 모두 같은 partition으로 전송됩니다.
처리해야 될 메시지의 양이 많아져서 partition값이 증가하는 경우 메시지의 키와 매핑된 해시 테이블도 변경되므로 메시지가 의도와 다른 방식으로 전송이 이루어질 수 있으므로 되도록이면 파티션 수를 변경하지 않는 것을 권장합니다.
producer의 메시지 중 record()의 키값은 필수값이 아닙니다.
키값을 지정 안 한다면 키값은 null 값이 되며 round-robin으로 topic의 partition으로 record들을 전송합니다.
아래 그림은 partition 번호가 1번부터 되어있지만 이전 포스트에서 봤듯 partition번호는 0부터 시작합니다.
파티셔너를 거친 record들은 배치 처리를 위해 producer의 버퍼 메모리 영역에 잠시 대기한뒤 kafka로 전송됩니다.
하지만 이런 과정이 효율을 떨어트릴 수 있는데 record 가 5개가 지나가도 아직 전송을 못하며 위와 같이 6개째가 돼야 첫 번째 파티션에서 배치 전송을 위한 최소 레코드수 2가 충족되며 전송됩니다.
apache kafka ≥ 2.4
Sticky Partitioner
스티키 파티셔너(Sticky Partitioner)는 하나의 partition에 record를 먼저 채워서 kafka로 빠르게 배치 전송하는 전략입니다.
위의 그림을 보면 최소 전송 사이즈가 3 임에도 불구하고 round-robin때와 다르게 record가 3개만 채워져도 partition1에 있는 메시지들은 kafka로 전송이 됩니다.
confluence에서 제공하는 자료를 보면 latency가 눈에 띄게 적은 것을 확인할 수 있습니다.
producer의 cpu 사용율에서도 줄어든다고 합니다.
프로듀서(producer)의 배치(batch) 처리 옵션
producer는 배치 전송을 위해 옵션을 제공합니다
| 옵션값 | 의미 |
| buffer.memory | producer의 버퍼 메모리 옵션 기본값은 32MB |
| batch.size | 배치 전송을 위한 record를 묶는 단위 기본값은 16kb로 되어있습니다. |
| linger.ms | 배치 전송을 위해 버퍼에 대기하는 최대 시간 설정 옵션 ms이며 기본값은 0 0인경우 배치를 위해 기다리지 않고 바로 보냄. |
배치처리에 있어서 포인트는 처리량을 우선할지 지연시간을 우선할지를 결정해야 합니다.
처리량을 늘릴수록 카프카 버퍼에서의 대기시간이 생기기 때문에 지연시간이 생기며 반대로 지연시간을 줄이면 처리량이 줄어들기 때문에 어느 부분을 우선할지를 정해야 합니다.
최적의 값을 찾아가는 것이 중요하며 처리량을 올리려면 batch.size와 linger.ms의 값을 올리고 지연시간을 줄이려면 batch.size와 linger.ms값을 줄여야 합니다.
당연히 record가 들어갈 buffer.memory 옵션은 batch.size보다 커야 하며 실패 시 재시도를 수행하는 것을 고려하여 partition 크기가 5인경우 batch.size 기본값 16kb → buffer.memory 16x5 = 80kb보다도 더 큰 값으로 설정해야 합니다.
중복 없는 전송
메시지 전송 방식에는 적어도 한번 전송(at-least-once), 최대 한 번 전송(at-most-once), 정확히 한번 전송(exactly-once)이 있으며 전송방식들의 차이들을 확인해 보도록 하겠습니다.
아래그림에서 topic 은 편의를 위해서 만들었으며 ack를 전달하는 주체는 kafka broker입니다.
적어도 한번 전송(at-least-once)
1. Producer가 messageA를 브로커의 특정 topic으로 전송합니다
2. 전송 후 ack를 broker로부터 받는 경우 다음 메시지로 넘어갑니다
3. MessageB와 같이 ack를 받지 못하는 경우 MessageB를 재전송합니다.
하지만 적어도 한번 전송(at-least-once)의 경우 broker가 ack를 보내지 못하는 경우가 ack만 보내지 못하는 경우인지 메시지를 저장하지 못해서 ack를 보내지 못하는 경우인지 알 수가 없습니다
메시지를 재전송하기 때문에 메시지 손실 가능성은 매우 적지만, broker가 메시지를 저장은 했지만 ack를 보내지 못한 경우 메시지 중복가능성이 존재합니다.
최대 한 번 전송(at-most-once)
최대 한번 전송(at-most-once)의 경우 적어도 한번 전송(at-least-once)과는 다르게
ack를 받지 못하더라고 다음 메시지를 전송합니다.
최대 한번 전송(at-most-once)의 경우 메시지의 중복 가능성을 피하기 위해서 재전송을 하지 않습니다.
일부 메시지의 손실을 감안하고 중복전송을 하지 않습니다.
일부 메시지가 손실되더라도 높은 처리량을 요하는 IoT와 같은 환경에서 사용합니다.
kafka 중복 없는 전송
1. producer는 messageA를 전송할 때 PID(Producer ID)와 Message 번호 0을 헤더에 포함해서 전송합니다.
2. broker는 messageA를 저장하고 PID와 Message 번호 0을 메모리에 기록하고 ack를 반환합니다.
3. Producer는 PID는 동일하게 Message 번호를 1로 증가시켜서 전송합니다.
4. ack를 못 받는 경우 MessageB를 재전송합니다.
5. Broker는 메모리에 있는 PID와 Message 번호를 비교해서 이미 저장되어 있는 경우 저장하지 않고 ack만 다시 반환합니다.
포인트는 PID와 message번호이며 broker는 unique 한 PID로 producer들을 구분하며 각각의 producer로부터 오는 Message번호가 broker가 갖고 있는 Message 번호보다 정확히 하나 큰 경우가 아니면 저장하지 않음으로 적어도 한번 전송(at-least-once)의 메시지 중복의 문제를 해결했습니다.
PID는 producer에 의해서 자동 생성 됩니다.
PID와 Message 번호는 Broker의 메모리뿐만 아니라 Replication Log에도 저장됩니다.
Broker가 다운되어 새로운 리더가 선출되어도 중복 없는 메시지 전송은 가능합니다.
오버헤드(Overhead)
사실 이러한 과정은 수많은 메시지가 전송되는 과정에서 오버헤드를 동반할 수밖에 없습니다.
kafka에서는 PID와 Message 번호와 같이 숫자 필드만 추가하는 방식으로 구현했기에 오버헤드가 높은 편이 아닙니다.
confluent에서는 아래와 같이 오버헤드값에 대해 측정한 값을 말해줍니다.
Exactly-once Semantics is Possible: Here's How Apache Kafka Does it
Exactly-once Semantics is Possible: Here's How Apache Kafka Does it
Exactly-once is a hard problem to solve, but we've done it. Available now in Apache Kafka 0.11, exactly-once semantics.
www.confluent.io
After much thought, we settled on a design that involves minimal overhead per transaction (~1 write per partition and a few records appended to a central transaction log). This shows in the measured performance of this feature. For 1 KB messages and transactions lasting 100 ms, the producer throughput declines only by 3%, compared to the throughput of a producer configured for at least once, in-order delivery (acks=all, max.in.flight.requests.per.connection=1), and by 20% compared to the throughput of a producer configured for most once delivery with no ordering guarantees (acks=1, max.in.flight.requests.per.connection=5), which is the current default.
각 트랜잭션당 최소한의 오버헤드만 포함하는 경우(~1 파티션당 기록하고 중앙 트랜잭션 로그에 몇 개의 record를 추가합니다). 중복 없이 전송을 사용한 경우 측정된 성능은 1 KB의 메시지와 100 ms의 트랜잭션 기간이 있는 경우, 적어도 한 번, 순서대로 전송을 위해 구성된 프로듀서와 비교하여 프로듀서의 처리량은 오직 3%만 감소하였고, 최대 한 번 전송하며 순서대로 전송하지 않도록 구성된 프로듀서와 비교하여 20% 감소하였습니다(acks=1, max.in.flight.requests.per.connection=5). 이는 현재 기본값입니다.”
중복 없이 전송을 위한 설정값
Producer 옵션 값 설명
| Producer 옵션 | 값 | 설명 |
| enable.idempotence | true | 프로듀서가 중복 없는 전송을 허용할지 결정하는 옵션 default 는 false true로 변경 시 다음에 나오는 옵션들도 반드시 변경해야 ConftgExceptton이 미발생 |
| max.tn.flight.requests. per.connection | 1~5 | ACK를 받지 않은 상태에서 하나의 커넥션에서 보낼 수 있는 최대 요청 수 기본값은 5이며, 5 이하로 설정해야 합니다. |
| acks | all | 프로듀서acks와관련된옵션으로서, 기본값은1이며all로설정해 야 합니다. |
| retries | 5 | ack를 받지 못한 경우 재시도를 해야 하므로 0보다 큰 값으로 설정 해야 합니다. |
Kafka 중복 없이 전송 실습
topic : “wout-dupli01”
replicatrion-factor : 3
topic 생성
kafka-console-producer.sh --bootstrap-server kafka1:9092 --topic wout-dupli01 --proudcer.config
producer.config 파일을 생성해야 합니다
자유롭게 원하시는 경로에 생성해주시면 됩니다.
저는 /usr/share 하위 경로에 생성했습니다.
vi 명령어를 통해서 producer.config 파일에
ack 명령어를 빼고 enable.idempotence, max.in.flight.requests.per.connection,retries 옵션만 적어 두었습니다.
producer.config 옵션으로 경로에 있는 config 파일을 참조해서 producer를 실행시킬 시 에러가 발생합니다.
Producer.config 파일에 acks=all 옵션을 추가해줍니다.
정상적으로 메시지를 보낼 수 있으며 kafka exactly once 010 메시지를 보냈습니다.
broker에 log 저장 경로에 snapshot이 생성됩니다.
🤔 kafka cluster에 붙어있는 모든 kafka 브로커에 snapshot이 없다면?
1. leader-epoch-checkpoint 정보를 확인하고
2. partition leader 역할을 하는 broker를 강제 종료합니다.
3. docker-compose에서 해당 컨테이너만 내렸다가 다시 올려주시면 됩니다.
4. partiton leader broker에 위 log 경로로 가면 snapshot이 생성되어 있습니다.
kafka dump 명령어를 통해서 snapshot 파일을 확인해 보면
PID 정보 45000과 0부터 순차적으로 증가하는 Message 번호(Sequence (first, last))정보와 offset 정보가 들어있는 것을 확인할 수 있습니다.
kafka dump 명령어는 log 파일의 온전성 검사 및 로그파일 내용을 console에 출력할 때 사용됩니다.
https://jaceklaskowski.gitbooks.io/apache-kafka/content/kafka-tools-kafka-dump-log.html
kafka-dump-log.sh · The Internals of Apache Kafka
jaceklaskowski.gitbooks.io
참조
https://jaceklaskowski.gitbooks.io/apache-kafka/content/kafka-tools-kafka-dump-log.html
Producer Default Partitioner & Sticky Partitioner | Learn Apache Kafka
Exactly-once Semantics is Possible: Here's How Apache Kafka Does it
'MessageQueue > Apache Kafka' 카테고리의 다른 글
| [Kafka] Kafka와 zero-copy (0) | 2023.01.30 |
|---|---|
| [Kafka] Leader Epoch(리더 에포크)을 활용한 카프카 브로커(kafka broker) 복구 (1) | 2023.01.28 |
| [Kafka] Replication(리플리케이션)과 Leader(리더)와 Follower(팔로워)를 알아보자 (0) | 2023.01.25 |
| [Kafka] Topic, Partition, Segment, Segment 관리, Offset (0) | 2023.01.21 |
| [Kafka] kafka cluster 실습 환경 구축 (0) | 2023.01.21 |
댓글