---

Kafka Producer

Topic에 메시지를 보낸다.

성능 / 로드밸런싱 / 가용성 / 업무 정합성등을 고려하여
어떤 브로커의 파티션으로 메시지를 보내야 할지 전략적 (Key)으로 결정

• Kafka에 데이터를 저장하는 주체

• Kafka에 데이터를 전송하는 애플리케이션, 서버
  • 1. 각각의 메시지를 토픽 파티션에 맵핑
  • 2. 파티션 리더에 Write 요청 전송

• 키 값을 정하면,
  해당 키를 가진 모든 메시지를 동일한 파티션에 전송 가능

• 키 값을 정하지 않으면,
  파티션은 Round Robin 방식으로 파티션에 균등하게 분배 (2.4 이전)
  2.4 이후에는 Sticky Partitioning으로 분배 전략

• 반드시, 리더 파티션에 있는 Broker와 통신

• Kafka Broker로 데이터를 전송할 때,
  파티셔너, 배치 생성을 거치게 된다.

---

Kafka Producer 구조

---

Producer Record

• Offset은 레코드가 생성될 때에는 포함하지 않는다.
• Topic과 메시지 값만 있어도 전송하는데에는 문제 없다.

---

Workflow

---

Serializer

객체 (Object)를 객체의 유형, 데이터의 포맷, 적용 시스템에 상관없이,
이동 / 저장 / 복원을 자유롭게 하기 위해서 바이트 배열 (바이트 스트림)형태로 저장하는 것.

• 객체는 Serialization과 Deserialization을 통해서 System to System 또는 서로 다른 저장영역에 이동 / 저장 / 복원을 자유롭게 수행한다.

Producer Record를 직렬화

• Key, Value를 Serializer를 기반으로 직렬화한다.
  • String, byteArray, ByteBuffer...

---

Partitioner

Producer는 어떤 Topic의 Partition에 Write할 지 결정

• murmur2 알고리즘 기반
• Round-Robin 방식

• kafka는 같은 키를 레코드 집합에 전달함으로써 주어진 수의 파티션에 대해 받은 순서대로 메시지를 같은 파티션에 기록하도록 할 것

  • 수신된 메시지 순서를 유지하려면 메시지에 적절한 키를 사용하는 것이 중요
  • 사용자 지정 파티션을 생산자에게 전달하여 메시지를 작성할 때 사용자 지정 파티션을 제어할 수도 있다.

---

Compression

Record Accumulator에 Write하기 전에 압축

• 처리량 향상, 낮은 지연, 디스크 활용 향상

---

Record Accumulator

Record는 Topic의 Partition에 누적

• 네트워크 전송은 매우 무거운 처리
  • Producer는 지정된 만큼의 메시지를 저장했다가 한 번에 Broker에게 전달
    • 높은 처리량 달성

• Record Accumulator가 담당
  • RA가 각 Topic의 Partiton에 대응하는 Batch Queue를 구성하고 메시지들을 Record Batch 형태로 묶어 Queue에 저장

---

Sender thread

각 Batch Queue에 저장된 레코드 배치들은 때가 되면 각각 Broker에 전달

• Sender는 Thread 형태로 구성

• Records는 두 가지 조건에 따라 producer에서 전송된다.
  • 정의된 배치 크기에 도달
  • 정의된 대기 시간에 도달

---

Options

• key.serializer
  • 레코드의 메시지 키를 직렬화하는 클래스 지정
  • 이것에 따라 역직렬화

• value.serializer
  • 레코드의 메시지 값을 직렬화하는 클래스 지정
  • 이것에 따라 역직렬화

• linger.ms
  • Accumulator에 있는 Batch를 전송하기 전까지 기다리는 최소 시간
  • Batch를 최대한 모아서 보내고 싶다면 값을 늘린다.

• max.request.size
  • Producer가 보낼 수 있는 최대 메시지 바이트 크기
  • Default : 1MB

• retry
  • Broker로부터 Error를 전달 받고 재전송을 시도하는 횟수
  • Default : 2147483647 -- 매우 큼

• max.in.flight.requests.per.connections
  • 한 번에 요청하는 최대 커넥션 수
  • 설정 값만큼 동시 요청
  • Default : 5

• max.request.size
  • Producer가 보낼 수 있는 최대 메시지 바이트 크기
  • Default : 1MB

• delivery.timeout.ms
  • acks를 받지 못한 메시지에 대해 설정된 시간만큼 retry를 진행
  • 시간이 지나도록 받지 못한다면 메시지는 Fail

• acks
  • 0, 1, -1(all) 중 하나로 선택
  • Default : 1
  • Producer가 전송한 데이터가 Broker들에 정상적으로 들어갔는지 전송 성공 여부를 확인
• partitioner.class
  • 파티셔너 클래스 지정
  • Default : org.apache.kafka.clients.producer.internal.DefaultPartitioner
    • 2.5.0 이후에는 UniformStickyPartitioner

• enable.idempotence
  • 멱등성 프로듀서 동작 여부
    • 중복 없이 전송
    • Producer의 Message 전송 Retry시 중복 제거
  • Default : false
    • 3.0.0 이후 true
• transaction.id
  • 레코드를 트랜잭션 단위로 묶을 지 여부
  • Default : null

---

ACK

• acks
  • 0, 1, -1(all) 중 하나로 선택

뒤로 갈수록 신뢰성이 높다.

• 복제의 정도를 확인하는 것
• acks 옵션에 따라 처리 속도에도 차이가 있다.
  • Replication Factor가 1인 경우 변화가 거의 없다.
    • 보통 2 이상 설정하므로 acks 별 동작 방식의 이해 필요

• Default : 1
• Producer가 전송한 데이터가 Broker들에 정상적으로 들어갔는지 전송 성공 여부를 확인

• ISR (In-Sync-Replica)와 연관이 되어 있다.

• Leader - Follower 파티션 간 복제 시, 시간이 소요
  • Leader 파티션에 데이터가 적재된 다음 Follower 파티션이 복제할 때
    시간이 소요되기 때문에 Offset 차이가 발생
    • Replication Lag

acks = 0

• 신뢰도가 가장 낮음
  • 즉, Producer가 Leader 파티션으로 데이터를 전송만 한 것

• 실제 저장까지 되었는지 여부는 확인하지 않는다.

• send()를 호출하면 끝

• 데이터 전송 속도는 가장 빠르다.

• 데이터 유실을 감당할 수 있는 경우에 사용하면 좋다.

  • EX) 물류 차량 위치

---

acks = 1

• 신뢰도가 중간

• Producer가 보낸 데이터가 Leader 파티션에 정상적으로 적재되었는지 확인

  • 적재되지 않았다면, 될 때 까지 시도 가능

• Follower 파티션까지 데이터 적재 보장을 하지 않는다.

  • 따라서 유실 가능

---

acks = -1

• 신뢰도가 가장 높음

• Producer가 보낸 데이터가 Leader 및 Follower 파티션에 정상적으로 적재되었는지 확인

  • `min.insync.replica(Topic) 옵션값에 따라 확인

• 처리되는 속도 매우 느림

  • 꼭 필요하면 사용

• 브로커 장애에도 데이터 저장 보장

---

출처

Kafka Producer Overview