const{Fragment:i,jsx:e,jsxs:n}=arguments[0];function _createMdxContent(r){const a={a:"a",blockquote:"blockquote",br:"br",code:"code",figure:"figure",h2:"h2",h3:"h3",h4:"h4",li:"li",ol:"ol",p:"p",pre:"pre",span:"span",ul:"ul",...r.components};return n(i,{children:[n(a.blockquote,{children:["\n",n(a.p,{children:["Naver D2 의 내용을 기반으로 Producer 에 대해 정리한 내용.",e(a.br,{}),"\n","2020년 작성된 내용이므로, 변화에 따른 내용도 포함되어 있다."]}),"\n"]}),"\n",e(a.p,{children:"Kafka 는 Distributed Streaming Platform 으로, Produce - Consumer 가 연결되어 데이터가 전달된다."}),"\n",e(a.figure,{"data-rehype-pretty-code-figure":"",children:e(a.pre,{tabIndex:"0","data-language":"mermaid","data-theme":"github-dark github-light",children:n(a.code,{"data-language":"mermaid","data-theme":"github-dark github-light",style:{display:"grid"},children:[e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"graph LR"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Producer[Producer] --\x3e Kafka[Kafka Cluster]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Kafka --\x3e Consumer[Consumer]"})})]})})}),"\n",e(a.h2,{id:"기본-구성-요소",children:e(a.a,{className:"anchor",href:"#기본-구성-요소",children:"기본 구성 요소"})}),"\n",e(a.p,{children:"Kafka Producer 는 크게 3가지 요소로 구성된다."}),"\n",e(a.figure,{"data-rehype-pretty-code-figure":"",children:e(a.pre,{tabIndex:"0","data-language":"mermaid","data-theme":"github-dark github-light",children:n(a.code,{"data-language":"mermaid","data-theme":"github-dark github-light",style:{display:"grid"},children:[e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"graph TD"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" KafkaProducer[KafkaProducer
사용자가 send() 호출]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" RecordAccumulator[RecordAccumulator
배치 단위로 저장]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Sender[Sender Thread
브로커로 전송 및 응답 처리]"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" KafkaProducer --\x3e RecordAccumulator"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" RecordAccumulator --\x3e Sender"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Sender --\x3e Broker[Kafka Broker]"})})]})})}),"\n",n(a.ul,{children:["\n",e(a.li,{children:"KafkaProducer : 사용자가 직접 사용하는 요소. 클래스의 send 를 호출해서 Record 를 전송"}),"\n",e(a.li,{children:"RecordAccumulator : Record 를 바로 전송하는게 아닌 해당 클래스에서 배치 단위로 저장"}),"\n",e(a.li,{children:"Sender : 별도 Sender Thread 가 쌓인 Record 를 Broker 로 전송, 응답 받고 전송 시 설정한 콜백 있으면 실행"}),"\n"]}),"\n",e(a.h3,{id:"kafkaproducersend",children:e(a.a,{className:"anchor",href:"#kafkaproducersend",children:"KafkaProducer.send"})}),"\n",e(a.figure,{"data-rehype-pretty-code-figure":"",children:e(a.pre,{tabIndex:"0","data-language":"mermaid","data-theme":"github-dark github-light",children:n(a.code,{"data-language":"mermaid","data-theme":"github-dark github-light",style:{display:"grid"},children:[e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"graph LR"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Send[send 호출] --\x3e Serializer[Serializer
Byte Array 변환]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Serializer --\x3e Partitioner[Partitioner
파티션 결정]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Partitioner --\x3e Compression[Compression
압축]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Compression --\x3e RecordAccumulator[RecordAccumulator
저장]"})})]})})}),"\n",n(a.p,{children:["send 가 호출되면, Serializer -> Partitioner -> Compression 작업이 이뤄지고",e(a.br,{}),"\n","최종적으로 RecordAccumulator 에 Record 가 저장된다."]}),"\n",e(a.h4,{id:"serialization",children:e(a.a,{className:"anchor",href:"#serialization",children:"Serialization"})}),"\n",e(a.p,{children:"Record 의 Key, Value 는 지정된 Serializer 에 의해 Byte Array 로 변환"}),"\n",e(a.figure,{"data-rehype-pretty-code-figure":"",children:e(a.pre,{tabIndex:"0","data-language":"java","data-theme":"github-dark github-light",children:n(a.code,{"data-language":"java","data-theme":"github-dark github-light",style:{display:"grid"},children:[n(a.span,{"data-line":"",children:[e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"Properties props "}),e(a.span,{style:{"--shiki-dark":"#F97583","--shiki-light":"#D73A49"},children:"="}),e(a.span,{style:{"--shiki-dark":"#F97583","--shiki-light":"#D73A49"},children:" new"}),e(a.span,{style:{"--shiki-dark":"#B392F0","--shiki-light":"#6F42C1"},children:" Properties"}),e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"();"})]}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",n(a.span,{"data-line":"",children:[e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"props."}),e(a.span,{style:{"--shiki-dark":"#B392F0","--shiki-light":"#6F42C1"},children:"put"}),e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"("}),e(a.span,{style:{"--shiki-dark":"#9ECBFF","--shiki-light":"#032F62"},children:'"key.serializer"'}),e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:", "}),e(a.span,{style:{"--shiki-dark":"#9ECBFF","--shiki-light":"#032F62"},children:'"org.apache.kafka.common.serialization.StringSerializer"'}),e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:");"})]}),"\n",n(a.span,{"data-line":"",children:[e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"props."}),e(a.span,{style:{"--shiki-dark":"#B392F0","--shiki-light":"#6F42C1"},children:"put"}),e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"("}),e(a.span,{style:{"--shiki-dark":"#9ECBFF","--shiki-light":"#032F62"},children:'"value.serializer"'}),e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:", "}),e(a.span,{style:{"--shiki-dark":"#9ECBFF","--shiki-light":"#032F62"},children:'"org.apache.kafka.common.serialization.StringSerializer"'}),e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:");"})]})]})})}),"\n",e(a.figure,{"data-rehype-pretty-code-figure":"",children:e(a.pre,{tabIndex:"0","data-language":"java","data-theme":"github-dark github-light",children:e(a.code,{"data-language":"java","data-theme":"github-dark github-light",style:{display:"grid"},children:n(a.span,{"data-line":"",children:[e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"Producer<"}),e(a.span,{style:{"--shiki-dark":"#F97583","--shiki-light":"#D73A49"},children:"String"}),e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:", "}),e(a.span,{style:{"--shiki-dark":"#F97583","--shiki-light":"#D73A49"},children:"String"}),e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"> producer "}),e(a.span,{style:{"--shiki-dark":"#F97583","--shiki-light":"#D73A49"},children:"="}),e(a.span,{style:{"--shiki-dark":"#F97583","--shiki-light":"#D73A49"},children:" new"}),e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" KafkaProducer<>(props, "}),e(a.span,{style:{"--shiki-dark":"#F97583","--shiki-light":"#D73A49"},children:"new"}),e(a.span,{style:{"--shiki-dark":"#B392F0","--shiki-light":"#6F42C1"},children:" StringSerializer"}),e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"(), "}),e(a.span,{style:{"--shiki-dark":"#F97583","--shiki-light":"#D73A49"},children:"new"}),e(a.span,{style:{"--shiki-dark":"#B392F0","--shiki-light":"#6F42C1"},children:" StringSerializer"}),e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"());"})]})})})}),"\n",e(a.p,{children:"도 가능"}),"\n",e(a.p,{children:"ByteArray, ByteBuffer, Double, Integer, Long 등 Serializer 도 제공한다."}),"\n",n(a.blockquote,{children:["\n",e(a.p,{children:"Jackson 을 사용하면, JSON 직렬화도 가능"}),"\n"]}),"\n",e(a.h4,{id:"partitioning",children:e(a.a,{className:"anchor",href:"#partitioning",children:"Partitioning"})}),"\n",n(a.p,{children:["Kafka Topic 은 여러 개의 Partition 으로 나뉘어져 있다.",e(a.br,{}),"\n","사용자 Record 는 지정된 Partitioner 에 의해 어떤 파티션으로 보내질지 정해진다."]}),"\n",n(a.p,{children:["Partitioner 를 지정하지 않으면, ",e(a.code,{children:"org.apache.kafka.clients.producer.internals.DefaultPartitioner"})," 가 사용된다."]}),"\n",n(a.ul,{children:["\n",n(a.li,{children:["\n",e(a.p,{children:"key 가 있는 경우 : Hash 기반으로, 특정 파티션 고정"}),"\n"]}),"\n",n(a.li,{children:["\n",e(a.p,{children:"key 가 없는 경우 : DefaultPartitioner 로 전략에 맞게 사용"}),"\n"]}),"\n",n(a.li,{children:["\n",n(a.p,{children:["예전에는 Round Robin 형식이였는데 문제점에 의해 개선을 했다고 한다.",e(a.br,{}),"\n","-> 배치 다 차기전에 ",e(a.code,{children:"linger.ms"})," 가 끝나고, 텅 비거나 작은 박스를 보내게 되는 네트워크 오버헤드 발생",e(a.br,{}),"\n","(메시지가 각 파티션으로 흩어지다 보니, Batch 가 채워지는 속도가 느렸음)"]}),"\n"]}),"\n"]}),"\n",n(a.p,{children:["이런 문제를 개선하기 위해 ",e(a.a,{href:"https://www.confluent.io/blog/apache-kafka-producer-improvements-sticky-partitioner/?fbclid=IwAR1eDEFO-Q0j7vqu5hXJxoYjRTy6yryHbZrJ0cSUKXJn1NUzkX3NHRualWk",children:"Sticky Partitioner"})," 의 방식으로 변경했다. - ",e(a.a,{href:"https://dol9.tistory.com/277",children:"참고 링크"})]}),"\n",n(a.ul,{children:["\n",n(a.li,{children:["Record 를 고르게 보내지 않고, 특정 Partition 이 다 찰때까지 Record Batch 를 꽉 채워서 보내게 동작",e(a.br,{}),"\n","-> 배치가 더 빨리 채워지고, 처리량은 늘어나고 지연 시간은 줄어든다"]}),"\n"]}),"\n",e(a.figure,{"data-rehype-pretty-code-figure":"",children:e(a.pre,{tabIndex:"0","data-language":"mermaid","data-theme":"github-dark github-light",children:n(a.code,{"data-language":"mermaid","data-theme":"github-dark github-light",style:{display:"grid"},children:[e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"graph TB"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:' subgraph RoundRobin["Round-Robin (기존)"]'})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" RR1[Record 1] --\x3e P0_RR[Partition 0]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" RR2[Record 2] --\x3e P1_RR[Partition 1]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" RR3[Record 3] --\x3e P2_RR[Partition 2]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" RR4[Record 4] --\x3e P0_RR"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" RR5[Record 5] --\x3e P1_RR"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Note1[각 파티션에 분산
→ 배치가 천천히 채워짐]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" end"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:' subgraph Sticky["Sticky Partitioner"]'})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" S1[Record 1] --\x3e P0_S[Partition 0]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" S2[Record 2] --\x3e P0_S"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" S3[Record 3] --\x3e P0_S"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" S4[Record 4] --\x3e P0_S"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" S5[Record 5] --\x3e P1_S[Partition 1]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Note2[한 파티션에 집중
→ 배치 빠르게 채워짐]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" end"})})]})})}),"\n",n(a.p,{children:["하지만, 그 다음에도 추가적인 문제가 존재했다.",e(a.br,{}),"\n","Sticky Partitioner 로 인해 파티션 레코드 불균형이 심하게 일어날 수 있다.",e(a.br,{}),"\n","브로커의 성능이 모두 동일하지 않을때 발생한다. - 파티션 Skews(쏠림)"]}),"\n",n(a.ul,{children:["\n",e(a.li,{children:"Sticky 방식은 특정 파티션의 배치가 전송 완료(ACK) 되어야 다음 파티션을 선택"}),"\n",e(a.li,{children:"특정 브로커의 네트워크가 느리거나, CPU 부하가 높으면 해당 브로커로 가는 배치 응답(ACK)이 늦어진다."}),"\n"]}),"\n",e(a.p,{children:"-> 응답이 늦어지는 동안, 해당 파티션에만 레코드를 쌓음 + 느린 브로커에 더 많은 데이터가 쌓이는 역설적 불균형 발생"}),"\n",n(a.p,{children:["이를 개선하기 위해 ",e(a.a,{href:"https://cwiki.apache.org/confluence/display/KAFKA/KIP-794%3A+Strictly+Uniform+Sticky+Partitioner",children:"Uniform Sticky Partitioning (Adaptive Partitioning)"})," 을 도입했다."]}),"\n",n(a.ol,{children:["\n",e(a.li,{children:"각 파티션 별 '전송 중인 데이터량'과 '응답 속도' 를 계산"}),"\n",e(a.li,{children:"응답이 느린 브로커 파티션은 선택될 확률을 낮추고, 응답이 빠른 브로커 파티션에 더 자주 붙인다. - Sticky 하게"}),"\n"]}),"\n",n(a.p,{children:["-> 배치 효율을 유지하되, 특정 브로커에 데이터를 몰리는걸 방지",e(a.br,{}),"\n","=> 전체 클러스터에 부하 균형"]}),"\n",e(a.h4,{id:"compression",children:e(a.a,{className:"anchor",href:"#compression",children:"Compression"})}),"\n",e(a.p,{children:"전송하려는 Record 를 압축해 네트워크 전송 비용도 줄이고, 저장 비용도 줄일 수 있다. (RecordAccumulator 에 저장될 때 바로 압축되어 저장)"}),"\n",n(a.ul,{children:["\n",e(a.li,{children:"compression.type 을 설정하여 코덱을 지정할 수 있다(기본값은 none)"}),"\n",e(a.li,{children:"gzip / snappy / lz4"}),"\n"]}),"\n",e(a.h3,{id:"recordaccumulatorappend",children:e(a.a,{className:"anchor",href:"#recordaccumulatorappend",children:"RecordAccumulator.append"})}),"\n",e(a.figure,{"data-rehype-pretty-code-figure":"",children:e(a.pre,{tabIndex:"0","data-language":"mermaid","data-theme":"github-dark github-light",children:n(a.code,{"data-language":"mermaid","data-theme":"github-dark github-light",style:{display:"grid"},children:[e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"graph TD"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" RecordAccumulator[RecordAccumulator]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:' Map["batches: Map<TopicPartition, Deque<RecordBatch>>"]'})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" RecordAccumulator --\x3e Map"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:' Map --\x3e TP0["TopicPartition 0"]'})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:' Map --\x3e TP1["TopicPartition 1"]'})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:' Map --\x3e TP2["TopicPartition 2"]'})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:' TP0 --\x3e Deque0["Deque<RecordBatch>"]'})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:' TP1 --\x3e Deque1["Deque<RecordBatch>"]'})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:' TP2 --\x3e Deque2["Deque<RecordBatch>"]'})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Deque0 --\x3e Batch01[RecordBatch 1]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Deque0 --\x3e Batch02[RecordBatch 2]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Deque0 --\x3e Batch03[RecordBatch 3]"})})]})})}),"\n",n(a.p,{children:["사용자가 전송하는 Record 는 전송 전에 먼저 RecordAccumulator 에 저장된다.",e(a.br,{}),"\n","batches 라는 Map 을 가지고 있는다. - Key 는 ",e(a.code,{children:"TopicPartition"}),", Value 는 ",e(a.code,{children:"Deque"})]}),"\n",n(a.ul,{children:["\n",n(a.li,{children:["저장하기 전, Serialized Size 검사 - ",e(a.code,{children:"max.request.size"})," or ",e(a.code,{children:"buffer.memory"})," 설정값보다 크면 RecordTooLargeException 발생"]}),"\n"]}),"\n",e(a.figure,{"data-rehype-pretty-code-figure":"",children:e(a.pre,{tabIndex:"0","data-language":"mermaid","data-theme":"github-dark github-light",children:n(a.code,{"data-language":"mermaid","data-theme":"github-dark github-light",style:{display:"grid"},children:[e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"graph LR"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:' subgraph Deque["Deque<RecordBatch>"]'})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" First[First
오래된 배치
Sender가 drain]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Middle[...]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Last[Last
최근 배치
append 대상]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" end"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Append[Record 추가] --\x3e|getLast| Last"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Drain[Sender 전송] --\x3e|getFirst| First"})})]})})}),"\n",n(a.p,{children:["추가될 Record 에 해당하는 TopicPartition 의 Deque 를 찾는다.",e(a.br,{}),"\n","-> Deque 의 Last 에서 RecordBatch 를 하나 꺼내서 Record 저장할 공간 있는지 확인",e(a.br,{}),"\n","-> 여유 공간이 있으면, 해당 RecordBatch 에 Record 를 추가, 없으면 새로운 RecordBatch 생성해서 Last 쪽으로 저장"]}),"\n",n(a.ul,{children:["\n",n(a.li,{children:["\n",e(a.p,{children:"drain 은 deque 맨 앞에서 꺼내서 처리 - 오래된 요소부터 보내는게 당연"}),"\n"]}),"\n",n(a.li,{children:["\n",e(a.p,{children:"append 시에는 가장 맨 뒤에서 꺼내서 처리 - 최근 생성된 RecordBatch 를 꺼내서 확인"}),"\n"]}),"\n",n(a.li,{children:["\n",n(a.p,{children:["RecordBatch 생성 시, BufferPool 에서 RecordBatch 가 사용할 ByteBufer 를 받아온다. - ",e(a.code,{children:"buffer.memory"})," 설정에 의해 결정",e(a.br,{}),"\n","-> Buffer Size 만큼 여유가 없으면 Allocation Blocking - ",e(a.code,{children:"max.block.ms"})," 설정값만큼 기다림, 시간 초과할 시 TimeoutException"]}),"\n"]}),"\n"]}),"\n",n(a.blockquote,{children:["\n",n(a.p,{children:["BufferPool 을 사용하는 이유?",e(a.br,{}),"\n","Kafka 의 처리량을 극대화 하기 위해 메시지 보낼 때마다",e(a.br,{}),"\n",e(a.code,{children:"ByteBuffer"})," 객체를 ",e(a.code,{children:"new"})," 로 생성하고 버리면, GC 로 인해 서버 부하가 생긴다.",e(a.br,{}),"\n","=> 미리 만들어두고, Buffer 를 재사용하는 방식으로 처리"]}),"\n"]}),"\n",n(a.ul,{children:["\n",n(a.li,{children:["RecordBatch 생성 시 사용하는 Buffer Size 는 ",e(a.code,{children:"batch.size"})," 설정값과 Record Size 중 큰 값으로 결정",e(a.br,{}),"\n","(",e(a.code,{children:"batch.size"})," 보다 작으면, 여러개의 Record 저장, Record 가 더 크면, 하나의 Record 만 저장)"]}),"\n"]}),"\n",e(a.h3,{id:"sender-thread",children:e(a.a,{className:"anchor",href:"#sender-thread",children:"Sender Thread"})}),"\n",n(a.p,{children:["Sender Thread 는 RecordAccumulator 에 저장된 Record 를 꺼내서 Broker 로 전송하고 응답 처리한다",e(a.br,{}),"\n","(정확히는, 무한 루프에서 주기적으로 체크 구조 - 보낼 준비 된 배치가 있는지 확인, 있으면 가져오기)"]}),"\n",e(a.figure,{"data-rehype-pretty-code-figure":"",children:e(a.pre,{tabIndex:"0","data-language":"mermaid","data-theme":"github-dark github-light",children:n(a.code,{"data-language":"mermaid","data-theme":"github-dark github-light",style:{display:"grid"},children:[e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"graph TD"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" RecordAccumulator[RecordAccumulator] --\x3e|drain| Drain[배치 수집]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Drain --\x3e Group[Node 단위로 그룹화]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Group --\x3e ProduceRequest[ProduceRequest 생성]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" ProduceRequest --\x3e InflightRequests[InflightRequests 대기열]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" InflightRequests --\x3e Send[Broker로 전송]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Send --\x3e Response[응답 처리]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Response --\x3e Callback[콜백 실행]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Callback --\x3e BufferPool[ByteBuffer 반환]"})})]})})}),"\n",e(a.p,{children:"RecordAccumulator 에서 Record 를 꺼냄 (drain 통해서 각 Broker 별 전송할 RecordBatch List 얻음)"}),"\n",n(a.ol,{children:["\n",n(a.li,{children:["\n",e(a.p,{children:"각 Broker Node 에 속하는 TopicPartition 목록 가져옴"}),"\n"]}),"\n",n(a.li,{children:["\n",e(a.p,{children:"TopicPartition 을 보면서 Deque First 쪽 RecordBatch 를 drain 하며 RecordBatch List 에 추가"}),"\n"]}),"\n",n(a.li,{children:["\n",n(a.p,{children:["Node 단위로 RecordBatch List 가 ",e(a.code,{children:"max.request.size"})," 설정값을 넘지 않을 때까지 수집"]}),"\n"]}),"\n",n(a.li,{children:["\n",e(a.p,{children:"모아둔 배치 리스트를 하나의 ProduceRequest 로 만듬(실제 브로커 날아가는 데이터 패킷 단위)"}),"\n",e(a.figure,{"data-rehype-pretty-code-figure":"",children:e(a.pre,{tabIndex:"0","data-language":"mermaid","data-theme":"github-dark github-light",children:n(a.code,{"data-language":"mermaid","data-theme":"github-dark github-light",style:{display:"grid"},children:[e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"graph TD"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:' subgraph Batches["RecordBatch List"]'})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" B1[Partition 0 Batch]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" B2[Partition 1 Batch]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" B3[Partition 3 Batch]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" B4[Partition 5 Batch]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" end"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:' subgraph Node["Node 단위로 그룹화"]'})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Broker1[Broker 1
Partition 0, 1]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Broker2[Broker 2
Partition 3, 5]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" end"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" B1 --\x3e Broker1"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" B2 --\x3e Broker1"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" B3 --\x3e Broker2"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" B4 --\x3e Broker2"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Broker1 --\x3e Request1[ProduceRequest 1]"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Broker2 --\x3e Request2[ProduceRequest 2]"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Note[네트워크 효율:
같은 브로커는 한 번에 전송]"})})]})})}),"\n",n(a.blockquote,{children:["\n",n(a.p,{children:["이 부분 꽤나 어렵다.",e(a.br,{}),"\n","데이터는 파티션별로 관리되지만, 전송은 브로커 단위로 묶어서 보낸다.",e(a.br,{}),"\n","=> 네트워크 효율성을 위해서"]}),"\n"]}),"\n",n(a.p,{children:["하나의 브로커가 여러 파티션을 관리할 수 있다. (EX: 파티션 0, 파티션 1을 같은 브로커에서 관리)",e(a.br,{}),"\n","-> 두 배치를 하나의 패킷으로 묶어서 보내면, 통신이 한번만 발생한다.",e(a.br,{}),"\n","(TCP 커넥션 및 오버헤드 감소 + I/O 효율)"]}),"\n"]}),"\n",n(a.li,{children:["\n",e(a.p,{children:"순서 보장 및 흐름 제어를 위해 InflightRequests 라는 대기열을 먼저 거침"}),"\n",n(a.ul,{children:["\n",n(a.li,{children:["Deque 의 Size 는 ",e(a.code,{children:"max.in.flight.requests.per.connection"})," 설정값으로 결정, 하나의 Broker 로 동시 전송할 수 있는 요청 수"]}),"\n",n(a.li,{children:["요청이 실패한 경우, retries 설정값이 1 이상인 경우 재시도 -> 이때, 위 설정값이 1보다 크면 순서가 바뀔수 있다⭐ - 2번째는 성공, 1번째는 실패",e(a.br,{}),"\n","(1로 설정하면, 순서를 보장할 수 있다. 하지만, 전송 성능은 떨어질 수 있다.)"]}),"\n"]}),"\n",n(a.blockquote,{children:["\n",e(a.p,{children:"이 역시도 최신 Kafka Version 에서는 개선이 되었다!!"}),"\n"]}),"\n",n(a.p,{children:[e(a.code,{children:"enable.idempotence=true"})," 이면, 시퀸스 번호를 검사하는 방식으로 한다.",e(a.br,{}),"\n","기존에 ",e(a.code,{children:"Seq 0"})," 까지 저장되어 있다고 가정."]}),"\n",n(a.p,{children:["네트워크 지연 때문에 요청 2(",e(a.code,{children:"Seq 2"}),") 보다 요청 1(",e(a.code,{children:"Seq 1"}),") 가 온다면?",e(a.br,{}),"\n","-> 순서가 안 맞기에 OutOfOrderSequenceException 을 반환한다.",e(a.br,{}),"\n","-> 프로듀서는 이에따라, 1번부터 다시 보낸다.",e(a.br,{}),"\n","=> 여러개를 동시에 보내도, 데이터 순서가 꼬이지 않는다."]}),"\n",e(a.figure,{"data-rehype-pretty-code-figure":"",children:e(a.pre,{tabIndex:"0","data-language":"mermaid","data-theme":"github-dark github-light",children:n(a.code,{"data-language":"mermaid","data-theme":"github-dark github-light",style:{display:"grid"},children:[e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:"sequenceDiagram"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" participant Producer"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" participant Broker"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Note over Broker: 현재 Seq: 0"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Producer->>Broker: Request 1 (Seq 1)"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Producer->>Broker: Request 2 (Seq 2)"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Note over Broker: 네트워크 지연으로
Req 2가 먼저 도착"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Broker--\x3e>Producer: OutOfOrderSequenceException
(Seq 2 rejected, expected 1)"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Note over Producer: 순서 보정 필요"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Producer->>Broker: Request 1 (Seq 1) 재전송"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Broker--\x3e>Producer: ACK (Seq 1)"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Producer->>Broker: Request 2 (Seq 2) 재전송"})}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Broker--\x3e>Producer: ACK (Seq 2)"})}),"\n",e(a.span,{"data-line":"",children:" "}),"\n",e(a.span,{"data-line":"",children:e(a.span,{style:{"--shiki-dark":"#E1E4E8","--shiki-light":"#24292E"},children:" Note over Broker: 순서 보장 완료"})})]})})}),"\n"]}),"\n",n(a.li,{children:["\n",e(a.p,{children:"Sender Thread 는 Java NIO 의 Selector (IO Multiplexing) 방식 사용해 비동기적으로 여러 채널 전송 처리"}),"\n"]}),"\n",n(a.li,{children:["\n",e(a.p,{children:"Broker 는 Connection 에서 요청이 들어온 순서대로 처리해서 응답 - 순서 보장"}),"\n"]}),"\n",n(a.li,{children:["\n",e(a.p,{children:"요청에 포함된, RecordBatch 콜백 실행 후, 응답을 Future 를 통해 사용자에게 전달"}),"\n"]}),"\n",n(a.li,{children:["\n",e(a.p,{children:"RecordBatch 에서 사용한 ByteBuffer 를 BufferPool 로 반환"}),"\n"]}),"\n"]}),"\n",e(a.p,{children:"를 통해 전송 처리 모두 마무리!"})]})}return{default:function(i={}){const{wrapper:n}=i.components||{};return n?e(n,{...i,children:e(_createMdxContent,{...i})}):_createMdxContent(i)}};

아티클

KafkaProducer Client Internals