Apache Kafka - Producer Acks, Batch, Page Cache, Flush

Producer Acks

Producer Parameter 중 하나

• acks 설정은 요청이 성공할 때의 정의하는데 사용되는 Producer에 설정하는 Parameter다.

acks=0

• acks=0 : ack가 필요하지 않음. 이 수준은 자주 사용되지 않는다. 메시지 손실이 다소 있더라도 빠르게 메시지를 보내야 하는 경우에 사용된다.

acks=1

• acks=1(default) : Leader가 메시지를 수신하면 ack를 보낸다. Leader가 Producer에게 ACK를 보낸 후 Follower가 복제하기 전에 Leader에 장애가 발생하면 메시지가 손실된다.
• At most once(최대 한 번) 전송을 보장한다.

acks=-1, acks=all

• acks=-1 : acks=all과 동일하다. 메시지가 Leader가 모든 Replica까지 Commit되면 ack를 보낸다. Leader를 잃어도 데이터가 살아남을 수 있도록 보장한다. 그러나 대기 시간이 더 길고 특정 실패 사례에서 반복되는 데이터 발생 가능성이 있다.
• At least once(최소 한 번) 전송을 보장한다.

Producer Retry

재전송을 위한 Parameters

• retry는 네트워크 또는 시스템의 일시적인 오류를 보완하기 위해 모든 환경에서 중요하다.
• retries 파라티머를 조정하는 대신에 delivery.timeout.ms 조정을 통해 재시도 동작을 제어한다.
• acks=0에서 retry는 무의미하다.

Producer Batch 처리

메시지를 모아서 한번에 전송

• Producer에서 Record를 만들고 send()를 호출해서 내부적으로 Serializer가 직렬화를 해주고 Partitioner에서 Key가 있으면 Partition 번호를 정해주거나 Key가 없으면 Round Robin을 통해서 정해주고 압축 옵션이 있으면 압축을 해준다. 그리고 그것을 Record Accumulator에 특정 Topic의 특정 Partitiond에 담는다. 그리고 그것을 ProduceRequest라고 하는 Object에 담아가지고 Broker 쪽으로 보낸다.

• 이 때 메시지를 한번에 모아서 보내는 옵션이 배치처리를 위해서 사용되는 옵션이 있는데 이 옵션을 사용한다면 Broker가 건바이건으로 하나씩 처리하지 않고 100개나 1000개를 한번에 처리해준다.
• Latency는 건바이건이 낫겟지만 처리량 측면에서는 한꺼번에 보내는 것이 더 좋다.

linger.ms (default :0, 즉시 보냄) : 메시지가 함께 Batch 처리될 때까지 대기 시간
batch.size (default: 16KB) : 보내기 전 Batch의 최대 크기

• Batch 처리의 일반적인 설정은 linger.ms=100 및 batch.size=1000000

Producer Delivery Timeout

send() 후 성공 또는 실패를 보고하는 시간의 상한

1. Producer가 send()를 하면 메시지를 send해서 메시지를 저장하는 Buffer를 할당받을 때 까지 대기하는 시간이 있다 -> max.block.ms
2. send()를 하면 delivery.timeout.ms의 범위 내에 들어가게 된다.
3. delivery.timeout.ms의 맨 처음에는 데이터를 배치처리해서 일괄처리해서 보낼 때 까지 대기하는 시간인 linger.ms가 있다.
4. 메시지가 쌓이면 데이터를 send()한다.
5. send() 후에 request를 받을 때 까지 기다리는 응답 시간인 request.timeout.ms가 있다.
6. 만약 Error가 발생했을 경우 또는 Retry가 필요할 경우 retry가 될텐데 그 때 retry 중간 중간 사이에 있는 대기 시간인 retry.backoff.ms가 있다.

Message Send 순서 보장

enable.idempotence

Producer가 메시지를 보내는데 동시에 여러 Request가 Broker 쪽으로 날아가게 되는 상황을 생각해보자.

1. Producer가 메시지를 보낸다.
2. 메시지를 모았다가 배치로 보내게 되는데 이 때 배치가 한꺼번에 동시에 여러개를 보내게 되면 Multiple in-flight request라고 한다.
   max.inflight.requests.per.connection=5(default) : Producer 하나가 하나의 Broker에 커넥션을 보낼 수 있는 한꺼번에 보낼 수 있는 최대한의 개수
3. max.inflight.requests.per.connnection의 수 만큼 Batch가 Broker쪽으로 날아가게 된다.
4. 성공적인 경우라면 Batch 0이 들어가고 그 다음 1,2,3,4 ... 이런식으로 순차적으로 들어가야 Partition에 순서대로 쌓인다.
5. 하지만 만약에 Batch 0이 실패했지만 Batch 1이 성공했다면 ??
   --> Batch 0에 대한 Retry가 발생하겠지만 Batch 1이 Partition에 먼저 쌓이게 된다.
   --> 즉 순서가 달라진다.

• 이런 문제를 방지하기 위해 메시지의 순서를 보장해주는 enable.idempotence 옵션을 true로 설정해주면 된다.

• enable.idempotence를 사용하면 Batch 0이 실패한다면 그 뒤에 따라오는 1,2,3,4 .. 같은 후속 Batch 들도 실패 처리가 된다.
  --> OutOfOrderSequenceException

Page Cache 와 Flush

• 메시지는 Partition에 기록된다.
• Partition은 Log Segment File로 구성된다. (defualt: 1GB마다 새로운 Segment 생성)
• Broker는 성능을 위해서 Log Segment는 OS Page Cache에 기록한다.
• 로그 파일에 저장된 메시지의 데이터 형식은 Broker가 Producer로부터 수신한 것, 그리고 Consumer에게 보내는 것과 정확히 동일하므로 Zero-Copy가 가능하다.
• Page Cache는 다음과 같은 경우 디스크로 Flush 된다.
  • Broker가 완전히 종료되었을 때
  • OS Background "Flusher Thread" 실행 시

ZeroCopy : 데이터가 User Space에 복사되지 않고 CPU 개입 없이 Page Cache와 Network Buffer 사이에서 직접 전송되는 것의 의미한다. 이것을 통해 Broker Heap 메모리를 절약하고 또한 엄청난 처리량을 제공한다.

Flush 되기 전에 Broker 장애가 발생하면?

이를 대비하기 위해서 Replication 하는 것

• OS가 데이터를 디스크로 Flush하기 전에 Broker의 시스템에 장애가 발생하면 해당 데이터가 손실된다.
• Partition이 Replication(복제)되어 있다면 Broker가 다시 온라인 상태가 되면 필요시 Leader Replica(복제본)에서 데이터가 복구된다.
• Replication이 없다면 데이터는 영구적으로 손실될 수 있다.

Kafka 자체 Flush 정책

• 마지막 Flush 이후의 메시지수(log.flush.interval.messages) 또는 시간(log.flush.interval.ms)으로 Flush(fsync)를 트리거하도록 설정할 수 있다.
• Kafka는 운영 체제의 Background Flush 기능(예: pdflush)을 더 효율적으로 허용하는 것을 선호하기 때문에 이러한 설정은 기본적으로 무한(기본적으로 fsync 비활성화)으로 설정한다.
• 이러한 설정을 기본값으로 유지하는 것을 권장한다.
• *.log 파일을 보면 디스크로 Flush된 데이터와 아직 Flush 되지 않은 Page Cache(Os Buffer)에 있는 데이터가 모두 표시된다.
• Flush된 항목과 Flush 되지 않은 항목을 표시하는 Linux 도구(ex: vmtouch)도 있다.

요약

• Producer Acks : 0,1,all(-1)
• Bacth 처리를 위한 옵션 : linger.ms, batch.size
• 메시지 순서를 보장하려면 Producer에서 enable.idempotence를 true로 설정하면 된다.
• 성능을 위해 Log Segment는 OS Page Cache에 기록된다.