Batch
Serialize -> Partitioning -> Record Accumulator-> Sender Thread -> Broker
send() 하나의 ProducerRecord를 입력 -> 바로 전송 안됨
내부 메모리에서 토픽파티션에 따라 Record Batch 단위로 묶인 뒤 전송됨
Record Accumulator
메시지 배치가 전송이 될 토픽 과 파티션에 따라 저장 되는 Producer 메모리영역
Producer의 Main Thread는 send()를 호출하고 Accumulator에 데이터 저장
Sender Thread는 별개로 데이터를 브로커로 전달
- buffer.memory
Record accumulator의 전체 메모리 사이즈 - batch.size
단일배치의사이즈 - linger.ms
Sender Thread로 보내기전 배치로 만들어서 보내기 위한 최대 대기시간
(Sender가 Batch에 메세지가 차기를 기다리는 최대 대기시간 0~20ms) - max.inflight.requests.per.connection
Sender Thread가 Partition마다 한번에 가져가는 Batch의 개수 0~5
Sync 방식은 ack를 기다리기 때문에 Batch에 메세지를 한개만 담을 수 있음
Acks
Producer에서 Kafka가 메시지를 전달 받았는지 확인하는데 사용되는 설정값
Sync, ASync 가리지 않고 acks를 기반으로 retry(재전송) 실행
단, ASync에서는 재전송에 따라 메세지 전송 순서가 바뀔 수 있음
Acks = 0
Leader Broker의 Ack메시지를 받지않고 다음 메시지를 바로 전송
메시지가 브로커에 기록되지 않더라도 재 전송 하지 않음
속도는 가장 빠르지만 메세지 소실의 우려가 있음
Acks = 1
Leader Broker에서 메세지를 받았는지 확인하는 ack만 받음
오류 메세지를 받거나 시간내에 오지않으면 재전송
다른 Broker들에게 잘 복사가 되었는지는 확인 안함
복제 도중에 Leader가 다운 될 경우 메세지 소실 우려
Acks = all or -1
Leader가 메세지를 받고 복제까지 완료된 후 보내는 ack를 받음
오류 메세지를 받거나 시간내에 오지않으면 재전송
메세지 손실이 일어나지 않도록 하지만 속도는 가장 느림
min.insync.replicas의 설정값보다 Broker개수가 적으면 오류
Retry
- max.block.ms
Send() 호출시 Accumulator에 입력하지 못하고 block되는 최대시간.
초과시 Timeout Exception (Accmulator가 Full 일때)- request.timeout.ms
ack 전송에 걸리는 최대시간 전송 재시도 대기시간 제외
초과시 retry를 하거나 Timout Exception 발생- retry.backoff.ms
ack 전송 재시도를 위한 대기시간- deliver.timeout.ms
Producer 배치 전송에 허용된 최대시간 초과시 Timeout Exception
delivery.timeout.ms >= linger.ms + request.timeout.ms- retries
retries는 재전송횟수를설정
보통 retries보다 delivery.timeout.ms를 조정 하는것을 권장
- example
retries=10, request.timeout.ms=10000ms, retry.backoff.ms=30ms인 경우
request.timeout.ms 기다린 후 재전송하기 전 30ms 이후 재전송 시도
이와같은 방식으로 재전송을 10회 시도하고 더 이상 retry 시도하지않음
만약 10회 이내에 delivery.timeout.ms에 도달하면 retry 시도하지 않음
Delivery
최대한번전송(at most once)
acks = 0, 중복 되지않고 한번만 메세지 전송
적어도한번전송(at least once)
akcs = 1 or all, 메세지 소실은 없지만 중복 가능
- example
Broker에 정상적으로 저장 됐지만 Network 오류로 ack를 보내지 못한 경우 Producer는 메세지를 다시 보냄 -> 같은 메세지 중복 저장
중복없이전송(idempotence)
acks = all & Producer ID와 Sequence를 Header에 저장하여 전송
(Sequence는 메시지의 고유 번호 0부터 시작하여 순차적으로 증가)
Sequence가 중복 될 경우 이를 메시지 로그에 기록하지 않고 ack만 전송
-> 메세지의 중복 제거 보장
가지고 있는 메시지의 Sequence보다 1만큼 큰 경우에만 브로커에 저장
-> 메세지의 순서 보장
- 필수 설정값
• enable.idempotence = true
• acks = all
• retries > 0
• 1 <= max.in.flight.requests.per.connection <= 5
위반시 Config 오류 발생
한계점이 있음 -> Transaction 기반 처리 필요
