3_Producer

Producer

 

드디어 Kafka라는 분산 messageQ에 데이터를 밀어넣어주는 producer이다.

앞서 broker를 일꾼이라고 비유한 바가 있는데, 이 일거리들을 던져주는게 producer의 역할이라고 할 수 있다.

 

Kafka에서는 데이터를 "메시지, 레코드, 이벤트" 등으로 부르고는 한다. 

사용하는 사람에 따라 편한 용어를 쓰는 것 같다.(필자는 해당 글에서 record라고 지칭하도록 하겠다.)

 

Record

Kafka에서 들어오는 데이터인 record는 다음과 같이 두 부분으로 나뉜다.

•  Header: topic, partition, timestamp 등의 metadata
• Body: key, value(Avro, Json 등의 다양한 포맷의 형태)

다음과 같이 웹에서 request와 비슷하게 header와 body로 나뉘게 된다. 

header에는 record관련 metadata가 들어있고, body는 실질적으로 사용하게 되는 데이터가 담기게 된다.

 

Kafka는 record를 byte array형태로 저장하게 되고 이 것은 serializer가 담당하게 된다.

 

Kafka가 record를 저장하는 architecture

이제 producer가 record를 보내게 되면 Kafka가 이것을 적절한 topic과 partition에 배정하는 것에 대해서 알아보자.

 

1. producer에서 record를 만든 이후에 send() 함수를 호출하게 된다.
2. send() 함수를 호출하게 되면 serializer를 통해서 해당 record를 byte array로 변경해준다. 
3. partitioner에 의해서 어느 파티션으로 해당 record를 배정할 지 정하게 된다.
   • byte array를 Compress(압축)을 해서 저장할지는 optional이다
4. producer에서 마지막으로 RecordAccumulator를 통해서 kafka-broker에게 record를 전송하게 된다.
5. broker는 성공인지 실패인지 metadata를 return하게 된다.
   • 실패시에는 재시도 하게 된다.

partitioner 알고리즘
partition = hash(key) % (number of partitions)
key가 null일 경우 Round Robin(2.4이전), Sticky(2.4이후)로 보내주게 된다.

** sticky: 하나의 batch(partition을 나눈 단위)가 닫힐때 까지 보내고 랜덤으로 다음 partition을 선택한다.

 

Ack

ack란 network 공부를 했다면 3way, 4way handshake때 들어봤을 것이다.

kafka에서 ack도 network에서의 ack와 거의 동일하게 메시지가 잘 보내졌는지를 확인하는 용도로 쓰인다.

• ack=0: ack가 필요하지 않다. 실무에서는 메시지의 손실을 최소화 해야하기 때문에 자주 사용하지 않는다. 주로 사용하는 곳은 손실이 있더라도 데이터를 빨리 전송해야하는 경우에 사용된다.
• ack=1(default): Leader(replicas중에서)이 메시지를 수신하면 ack를 보내주는 방식으로 연결을 확인하는 방법이다.  At most once(최대 한번) 전송을 보장하는 옵션이다.
• ack=-1: ack=all과 동일한 옵션이다. Leader가 모든 replica까지 commit되면 ack를 보낸다. At least once(최소한번)을 보장한다.

Ack 관련 옵션들

retries=MAX_INT
# 메시지 send하기 위해 재시도하는 횟수

retry.backoff.ms=100
# 재시도 사이에 추가되는 대기 시간

request.timeout.ms=30000
# producer가 응답을 기다리는 최대 시간

delivery.timeout.ms=120000
# send()후 성공 또는 실패를 보고하는 시간의 상한
# delivery.timeout.ms 조정으로 재시도 동작을 제어.
# acks=0에서는 무의미 하다.

 

Batch

Batch 처리는 **RPC(Remote Procedure Call) 수를 줄여서 broker가 처리하는 작업이 줄어들기 때문에 더 나은 처리량을 제공하게 된다.

 

RPC(Remote Procedure Call)이란?
원격 프로시저 호출(영어: remote procedure call, 리모트 프로시저 콜, RPC)은 별도의 원격 제어를 위한 코딩 없이 다른 주소 공간에서 함수나 프로시저를 실행할 수 있게하는 프로세스 간 통신 기술이다. 다시 말해, 원격 프로시저 호출을 이용하면 프로그래머는 함수가 실행 프로그램에 로컬 위치에 있든 원격 위치에 있든 동일한 코드를 이용할 수 있다.(wikipedia)

한마디로 원격지의 프로세스에 접근하여 procedure 또는 function을 호출하여 사용한다는 것이다. 분산 컴퓨팅 환경에서 프로세스간 상호 통신 및 컴퓨팅 자원의 효율적 사용을 위해 발전된 기술이다.

출처: 위키백과

 

Batch 관련 옵션들

linger.ms=0 # 즉시보냄
# 메시지가 함께 Batch처리될 때까지 대기 시간

batch.size=1000000
# 보내기 전 batch의 최대 크기

batch 처리의 일반적인 설정으로는 linger.ms=100, batch.size=1000000으로 한다.

 

Message send 순서보장

이렇게 여러번 재시도를 하게되면 재시도를 하는 도중 순서가 변경될 수 있다.

이를 방지하기 위한 옵션도 있다.

max.in.flight.requests.per.connection=5 # default
# 여러개의 배치가 동시에 날아갈 수 있는 갯수

enable.idempotence=true
# 하나의 batch 실패시 같이 partition으로 들어오는 후속 batch도 모두 OutOfOrderSequenceException과 함께 실패

 

Page Cache & Flush

partition들은 log segment file(간단히 segment)로 구성된다는 것을 앞선 Topic에서 배울 수 있었다.

segment들은 메모리가 아닌 디스크에 저장되기 때문에 성능에 관한 생각을 해볼 수 있을 것이다.

kafka는 이를 해결하기 위해 OS page cache에 기록된다.

Disk에 저장된 포맷과 OS page cache에 써진 것과 정확히 동일해 **zero copy가 가능하다.

zero copy
전송 데이터가 User Space에 복사되지 않고, CPU의 개입없이 Page Cache와 network buffer 사이에서 직접 전송되는 것을 의미.

OS가 데이터를 디스크로 Flush하기 전 broker의 시스템에 장애 발생시에는 손실이 일어난다.

여기서 손실에 대한 복구는 replication으로 하게 된다.(replication이 없다면 영구적 손실이 된다.)

 

Kafka는 운영체제의 backgroud flush 기능을 더 효율적으로 허용하는 것을 선호한다.

 

Producer 관련 CLI

# topic 메시지 생성
$ kafka-console-producer.sh --bootstrap-server my-kafka1:9092,my-kafka2:9092,my-kafka3:9092 \
--topic hello.kafka
>hello
>kafka
>0
>1
...

# key를 가지는 메시지 생성
$ kafka-console-producer.sh --bootstrap-server my-kafka1:9092,my-kafka2:9092,my-kafka3:9092 \
--topic hello.kafka \
--property "parse.key=true"
--property "key.separator=:"    # key의 구분자를 :으로 설정
>key1:no1
>key2:no2
>key3:no3