[Kafka 원리]Kafka Producer 처리 방식 이해하기(feat. 튜닝, batch, linger, acks, compression, message send주기, message 순서 보장)

목차

• Producer란?
• Producer Message 전송 원리 이해하기
• Producer Message 전송 완료 listener 원리 이해하기
• Batch 처리(feat. batch.size, linger.ms)
• Messge Compression(메세지 압축)
• Message send Retry 원리 이해하기
• Message 순서 보장 알아보기(feat. idempotence, max.in.flight.requests.per.connection)

Producer란?

• Producer는 kafka broker에 메세지를 보내는 client이다.
• record라는 형식으로 데이터를 만들어 보낼 수 있다.
• record는 avro, json형식으로도 보낼 수 있다.
• record는 byte형식으로 serialize 되어 보내지고 broker에 저장된다.
• record를 작성할 때 bootstrap.server, key serializer, value serializer를 필수로 작성하여야한다.

 

Producer가 Broker에 message전송 할 때 구동 원리 이해하기

• Producer(Client)가 메세지(ProducerRecord)를 만들어 Send메소드를 호출 하면 Serializer가 내용을 serializing한다.
• Serializer는 Partitioner에게 해당 내용을 보낸다. Partitioner는 Key값을 확인하여 있다면, 해당 Topic에 Patition별로 보낼 준비를 한다. 없다면 Round Robin형식으로 Topic으로 알아서 할당할 준비를 한다.
• Partitioner는 Key값을 기준으로 어느 Partition에 메세지를 보낼지 정하여 바로 보내거나 또는 batch 기준이 있다면, batch처리하여 모아서 보낸다.
• 압축옵션이 있다면 Compression을 진행한다. (압축은 broker에서 진행 하는 것이 아닌 client에서 진행된다.)
• 위에 로직이 완료 되었다면 Topic으로 Producer가 만들어준 Record를 전송한다. Batch처리가 있다면, 기준이되는 Batch.size와 Linger.ms 속성에 따라 처리한다. (배치 처리 주요 내용 Batch.size가 도달 하거나 Linger.ms에 설정된 시간에 도달하면, Broker에 메세지를 보내주게 된다.)

 

Producer가 메세지 전송가 잘 확인 되었는지 확인 하는 방법(feat. acks)

• 위와 같은 원리로 Producer는 메세지를 전송한다.
• 더 중요한 것이 있는데 메세지가 잘 전달 되었는지 확인 하는 것이 필요하다. 그 때 사용하는 것이 ACK이다.
• 메세지를 보낼때 ack옵션을 설정하여 보낼 수 있고, 설정 값에 따라서 반환하는 방법이 다르다.
• ack 옵션은 default값으로 1로 설정 되어있다.
• 아래 예시는 topic이 partition 1 replica 2로 설정 되어있다고 가정

ACK=0

• Ack=0으로 설정하면 메세지를 보내기만 한다.
• replica를 아무리 적용한다고 해도 Producer에서는 확인 할 수 없다.
• 빠르게 데이터를 전송해야할 때만 사용하는 옵션이다
• At most once(최대 한번)적용하는 옵션이다.

ACK=1

• 리더가 받았는지 확인하는 옵션이다
• replica가 적용되었는지 확인 할수 없다
• leader가 장애가 발생한다면 메세지가 손실이 될 수 있다.
• At most once(최대 한번)이나 Exactly once(정확히 한번) 적용하는 옵션이다.

ACK=-1 or ACK=all

• 리더에 메세지가 전송되고 replica가 다 적용된 후에 돌려 받는 형식의 옵션이다.
• 리더가 장애가 발생하였어도 복제가 끝난 다음에 ack를 돌려주기 때문에 메세지가 보장이 된다.
• at least once(최소 한번)은 메세지를 보장해주는 옵션이다.
• 팔로워가 패치가 다 되었는지 기다려야하기 때문에 앞의 옵션보다 시간이 많이 걸린다.
• Producer가 메세지 Throuput이 적어지고 대기량이 많아진다면, batch.size와 linger.ms로 튜닝을 진행하거나 partition을 늘려서(축소안됨) 병렬 처리 할 수 있도록 해야한다.

Producer Batch 처리

Batch처리를 하는 이유 - 위에 그림에서 메세지를 보낼 때 batch라는 용어가 보여졌을 텐데 Batch처리를 하는 이유에 대해 알아 봐야 할 것 같다.

• Kafka는 메세지를 보낼 때 ACK 로직을 타게 돼는데 메세지당 한번의 ACK로직이 발생한다. 
• Producer가 보내는 메세지의 량이 많아지게 돼면 메세지 처리 속도가  늦어져 Lag이 걸리는 latency가 발생한다.
• 이를 해결하기 위한 Producer의 방식은 Batch 처리를 이용한다. batch를 이용하면 메세지를 묶음 으로 보내기 때문에 replica처리 로직이 줄어들어 Latency를 방지 할 수 있다. 즉, 메세지 send 처리가 대기 줄일 수 있다.
• batch 설정 옵션은 batch.size와 linger.ms이

 

Batch 처리 주요 옵션

옵션	설명
batch.size(default 16kb)	size를 정의 하여 메세지의 용량이 size에 도달 할 때 까지 기다렸다가 보낸다.
linger.ms(default 0)	batch.size가 도달하지 않으면 메세지를 보내지 않기 때문에 마냥 기다릴 수는 없어 해당 시간을 설정하여 size가 도달하지 않더라도 시간이 초과하면 메세지를 보내게 된다.

 

Messge Compression(메세지 압축)

• 메세지를 batch로 처리 했을 때 사용하게 된다.
• batch처리 할때 꼭 사용하는 것이 아니라 처리량(메세지 보내는 횟수)이 낮은데 메세지의 용량이 클 때 주로 사용한다.
• 활성화 하는 법은 compression.type이 default값이 none인데 이 부분을 수정해 주면 활성화 된다. compression.type의 값은 Gzip, snappy, Lz4, Zstd가 있다.

 

Producer가 Retry 할 때 어떻게 작동 할까?

위에 그림에서 메세지를 보냈는데 실패하는 경우도 알아 봐야 한다. 아래 그림에서 설정한 값의 내용을 알 수 있다.

옵션	설명	Default
retries	메세지를 send하기 위해 재시도 되는 횟수	MAX_INT
retry.backoff.ms	재시도 사이에 추가되는 대기 시간	100
request.timout.ms	Producer가 응답을 기다리는 최대 시간	30.0000(30초)
delevery.timeout.ms	send 후 성공 또는 실패를 보고하는 시간의 총 시간	120.0000(2분)

 

메세지 send 주기

속성에 대한 메세지 관련 주기는 아래와 같은 그림을 가진다. delivery.timeout.ms는 메세지를 완료 보고 받는데 걸리는 총 시간이기 때문에 linger.ms, retry.backoff.ms, request.timeout.ms가 합쳐진 시간 보다 커야 한다.

https://www.conduktor.io/kafka/kafka-producer-retries

 

Message 순서 보장

batch처리를 할 때 순서를 보장하는 것은 어떻게 동작 할까? 처리량을 늘리기 위해서 batch를 여러개 동시에 보낼 수 있다. 관련 속성은 max.in.flight.requests.per.connection=5(default)이다. 한번의 connection당 요청으로 나라갈 수 있는 최대 갯수이다. 하지만 batch 5개를 보내는 도중에 실패하게 되면 어떻게 될까? commit되는 메세지의 순서가 뒤바뀌게 된다. 그래서 이를 방지 하기 위해 enable.idempotence=true를 주어 뒤바뀌어도 전체 batch message send가 전체 retry되도록 하여야한다.

 

Message 순서 보장을 위한 옵션

옵션	설명
max.in.flight.requests.per.connection	한번의 connection당(메세지를 브로커에 보내는) 요청으로 메세지가 날라갈 수 있는 최대 갯수이다
enable.idempotence	멱등성 메세지의 순서 보장을 할 것인지 말것인지 설정 만약 true로 설정했는데 메세지 순서 보장이 되지 않는다면 OutOfOrderSequenceException오류가 발생하여 retry가 된다.

Producer 주요 옵션

아래는 Producer를 설정 할 수 있는 옵션의 종류입니다. 아래 내용을 보고 Broker에 어떻게 메세지를 전송할지 설정 할 수 있습니다.

옵션	설명
boostrap.servers	클라이언트가 카프카 클러스터에 처음 연결하기위한 서버와 포트
client.dns.lookup	ip와 연결하지 못할 경우 다른 ip 연결 시도 설정. (default = use_all_dns_ips)
acks	메세지를 받았는지 확인하는 옵션 0 = 빠른 전송, 1 = 리더가 받았는지만 확인, all = 리더 팔로우가 받았는지 확인
buffer.memory	카프카 서버로 보내는 메세지 버퍼(딜레이, 배치), 메모리는 바이트로 설정 됨
compression.type	압축 해서 보낼지 선택 옵션 - none. gzip. snappy, lz4,zstd 중 선택
enable.idempotence	producer가 메세지 전송을 단한번만 하게 하는 옵션
max.in.flight.requests.per.connection	커넥션에서 프로듀서가 최대한 ack없이 전송 할 수 있는 요청 수
retries	데이터 다시 보내는 횟수
batch.sizes	프로듀서가 파티션으로 보내는 배치사이즈. ack를 설정하면 복제되는 것을 검수하기 때문에 많이 보낼 수록 카프카 서버가 부하과 옴 배치로 보내는 것은 성능에 도움을 줌
linger.ms	배치 메세지를 보내기전 추가적인 기다리는 시간 조절 배치 사이즈에 도달하지 않았는데 이 곳의 설정 시간이 초과하면 메세지 보내게 됨

 

 

 

 

 

 

 

 

참고문헌

https://medium.com/@sdjemails/kafka-producer-overview-4c44b1b9ece1

https://www.popit.kr/kafka-운영자가-말하는-producer-acks/

https://4betterme.tistory.com/168

https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html

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

https://dzone.com/articles/take-a-deep-dive-into-kafka-producer-api

https://devidea.tistory.com/90

https://www.cloudkarafka.com/blog/apache-kafka-idempotent-producer-avoiding-message-duplication.html

https://www.conduktor.io/kafka/kafka-producer-retries

https://www.conduktor.io/kafka/kafka-message-compression

https://www.conduktor.io/kafka/kafka-producer-acks-deep-dive