Consumer

💡 이번 포스팅에서는 Consumer -1에 이어서 Consumer에 대해 정리해 보도록 하겠습니다.

출처: https://d2.naver.com/helloworld/0974525

컨슈머는 위와 같이 ConsumerCoordinator, ConsumerNetworkClient, Fetcher, SubscriptionState 등의 주요 내부 객체와 별도의 HeartBeat Thread를 생성합니다.

이전 포스팅에서 정리하지 않은 Fetcher, ConsumerNetworkClient, SubscriptionState에 대해서 정리해 보도록 하겠습니다.

Fetcher, ConsumerClientNetwork

Fetcher, ConsumerClientNetwork 객체는 브로커의 토픽 파티션에서 메시지를 Fetch 및 Poll을 수행하는 객체입니다.

출처: https://dan.iftodi.com/2019/02/apache-kafka-consumer/

위 그림들과 같이 컨슈머는 poll() 을 통해 메시지를 가져오는데 이때 메시지를 Fetcher 객체에서 가져옵니다.

Fetcher 객체에서 메시지를 Linked Queue 형태로 저장하고 있으며 메시지가 있을 경우 그대로 반환하지만, 만약 Fetcher 객체에 메시지가 없다면 ConsumerNetworkClient 객체를 통해 브로커로부터 메시지를 가져오도록(fetch) 요청합니다.

브로커가 ConsumerNetworkClient 객체의 fetch 요청에 의해 메시지를 전송하게 되면 ConsumerNetworkClient는 이 메시지를 Fetcher 객체에 저장하게 됩니다.

그리고 Fetcher 객체는 이제 데이터가 있으므로 컨슈머에게 메시지를 반환하게 됩니다.

Fetcher, ConsumerNetworkClient와 관련된 설정은 아래와 같습니다.

파라미터	기본값	설명
fetch.min.bytes	1	브로커로부터 한번에 가져올 수 있는 최소 데이터 사이즈입니다. 만약 옵션에서 지정한 사이즈보다 모인 데이터가 적을 시에는 바로 요청에 대해 응답하지 않고 데이터가 누적될 때까지 기다리거나 응답을 기다리는 최대 시간까지 기다렸다가 응답하게 됩니다.
fetch.max.bytes	50MB	브로커로부터 한번에 가져올 수 있는 최대 데이터 사이즈입니다. (파티션 총합)
fetch.max.wait.ms	500ms	fetch.min.bytes 이상의 메시지가 쌓일때까지 최대 대기 시간입니다. fetch.min.bytes 이상의 메시지가 쌓이지 않아도 fetch.max.wait.ms가 지나면 메시지를 전송합니다.
max.partition.fetch.bytes	1MB	Fetcher가 파티션별로 한번에 최대로 가져올 수 있는 bytes 입니다.
max.poll.records	500	Fetcher가 한번에 가져올 수 있는 레코드 수 입니다.