'오픈소스뜯어보기' 태그의 글 목록

오픈소스뜯어보기

카프카 오픈소스 뜯어보기 Consumer poll 메서드 (kafkaConsumer) 2022.11.22

카프카 오픈소스 뜯어보기 Consumer poll 메서드 (kafkaConsumer)

seanical 2022. 11. 22. 13:12

2022. 11. 22. 13:12

카프카의 개념을 이해했다는 전제하에 작성한 글입니다.

브로커의 데이터를 가져오기 위하여 보통 spring-kafka의 @kafkaListener 어노테이션을 등록하여 사용하거나 아래와 같이 poll 메서드를 작성하는 경우가 많습니다.
컨수머에 대한 아키텍처 구성에 대해서 알아보고 어떤 흐름을 통해 데이터를 컨숨할수 있을지 알아보려고 합니다.

public class Consumer {
    public void kafkaClient() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "127.0.0.1:9092");
        props.put("group.id", "consumer");
        props.put("enable.auto.commit", "true");
        props.put("auto.offset.reset", "latest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList("consumer"));
 
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Topic: %s, Partition: %s, Offset: %d, Key: %s, Value: %n\n",
                            record.topic(), record.partition(), record.offset(), record.key(), record.value());
                }
            }
    }

Kafka는 용어가 많기 때문에 혼란이 되어서 키워드 별로 하는 일을 정확히 명시해야 이해하기 쉽습니다.

GroupCoordinator : 브로커 중 하나로 컨슈머 그룹의 메타데이터와 그룹을 관리
ConsumerCoordinator : HeartbeatThread 생성, 컨수머 그룹 파티션 재할당
ConsumerNetworkClient : 컨수머의 Request를 NetworkClient로 보내는 주체
NetworkClient : ConsumerNetworkClient에게 받아 브로커에게 통신을 보내는 주체
Fetcher : 브로커 노드에게 데이터 요청하는 주체
SubscriptionState : 컨수머의 구독상태 오프셋 상태 관리

Consumer는 데이터만을 가져오게 감쳐줘 있지만 사실 데이터를 가져오기 위해서는 위 정리한 키워드 GroupCoordinator(Broker 객체)를 제외하고 객체들이 필요합니다.

전체적인 큰 그림을 보기 위하여 Consumer가 ConsumerNetworkClient에게 Request를 보낼 것들이 뭐가 있을까 생각해봅시다.

컨수머는 GroupCoordinator의 정보에 의존하여 메타데이터를 받아옵니다.
브로커가 여러개있을 경우(=브로커의 서버가 여러대) 브로커 중 GroupCoordinator를 찾는 요청도 있을 겁니다.

기본적으로 우리의 첫 번째 목적인 브로커에 있는 데이터 요청 도 있을 겁니다.
또한 Consumer가 받아온 데이터를 읽었다는 처리 Flag(commit) 를 브로커에게 넘겨주는 요청도 있을 겁니다.

이외에도 사실 요청이 굉장히 많습니다.

위 얘기를 꺼낸 이유는 이 모든 요청을 보내기 위해
ConsumerNetworkClient의 unsent라는 Map<node,newClientRequest> 자료구조에 담아놓습니다.
노드별 원하는 Request를 설정하여 map에 쌓여있는 것들을 NetworkClient에게 보내고 Broker한테서 응답이 오면
NetworkClient에게 Response를 받고 원하는 데이터 형태를 셋팅하여 우리가 원하는 일들을 수행합니다.

들어가기 앞서 poll의 핵심 로직 코드만 설명하고있습니다. 참고해주세요

1. Fetcher에게 Consumer에 할당된 Topic partition을 가져오라고 명령합니다.

Fetcher#sendFetches

sendFetches는 크게 4가지의 역할을 합니다

노드 별 FetchRequest 준비
노드별 FetchRequest.Builder Setup
ConsumerNetworkClient에게 request 전송
Future Reqeust 성공 실패 callback 메서드 주입

1번에서는 내가 읽는 Topic partition의 offset들을 어디까지 읽었고 그 이후부터 Broker에 저장되어있는 메시지의 offset을 가져갈거야라고 준비를 하는 단계입니다.

2번에서는 ConsumerNetworkclient에게 요청을 보낼 때 위에서 말한 여러 가지의 종류의 Request가 존재하기 때문에 Request의 목적은 FetchRequest를 전달이라는 것을 선언하는 줄입니다. (Request의 추상화)

3번에서는 실제 ConsumerNetworkClient에게 request를 보냅니다.

4번에서는 3번에서의 요청이 성공했을 때 실패했을 때를 나눠 callback메서드를 주입해줍니다.
(interface를 통한 익명 클래스 사용)

4번에 대한 추가 설명

해당 Callback메서드가 호출되는 시점은 누군가에 의해 성공했다고 판단되었고 (여기서 누군가 = 나중에 나오겠지만 ConsumerNetworkClient의 자료구조 map에 담긴 pendingcompletion에서 꺼내어 판단합니다)

onSuccess의 매개변수에 ClientResponse가 담아져 있었을 겁니다.

따라서 1번 코드 부분에서는 브로커 메시지 정보가 담겨있는 응답에 대해 파티션에 대해 데이터를 정리하는 역할입니다.

2번 코드 부분에서는 completedFetches라는 곳에 실제 완료된 데이터를 담아둡니다.

결과적으로 Client(KafkaConsumer)가 poll을 호출했을 때 completedFetches의 데이터를 가져가 반환합니다.

2. ConsumerNetworkClient의 send 요청을 보낸다.

위 1번의 3번 코드 부분이 이어서 진행되는 부분입니다.

ConsumerNetworkClient#send

send는 크게 3가지의 역할을 합니다

미래의 응답 값이 처리된다면 불릴 Handler지정
ClientRequest 객체 생성
노드별 ClientRequest를 unsent에 담기

1번의 코드를 2번의 코드 ClientRequet의 매개변수로 넣어주네요

지금은 아니지만 1번의 나중 역할은 보내는 Request가 성공, 실패했을 때 ConsumerNetworkClient의 pendingCompletion에 넣어두는 역할을 합니다.

(해당 부분은 뒤에 나올 NetworkClient의 poll 메서드 중 handleCompletedSend의 부분에서 처리되어 pendingCompletion에 들어갑니다.)

ConsumerNetworkClient는 pendingCompletion를 갖고 있습니다.
pendingCompletion 역할은 ClientRequest가 요청 실패했던 성공했던 일단 브로커로부터 응답이 온 Request들을 모아두고 있습니다

아래의 onComplete, onFailure이 pendingCompletion에 넣어주는 부분입니다. (코드를 보시려면 더보기를 눌러주세요)

ConumerNetwrokClient#RequestFuturerㅇcompletionHandler

2번 코드를 다시 보면 많은 매개변수가 있지만 expectResponse: true라는 것을 한번 봅시다.

우리는 이 요청이 브로커의 응답을 받아야만 하는 작업이면 true로 설정하고 요청만 보내고 끝나는 작업은 false로 표기하여 보냅니다.

(아래 handleCompleteSends 부분에서 나옵니다)

3번 코드를 보면 위에서 처음 말한 내용이 기억나시나요?

ConsumerNetworkClient의 unsent라는 Map<node,newClientRequest> 자료구조에 담아놓습니다.

해당 부분 코드가 그 역할을 담당하고 있습니다.

unsent라는 곳에 노드별 원하는 request(그룹코디네이터찾기, 데이터가져오기,etc) 들을 담아서 가지고 있습니다.

unsent라는 곳에 담아서 데이터 통신을 하여 보내주는 역할은 KafkaConsumer의 pollForFetches라는 부분에서 호출됩니다.

갑자기 PollForFetches는 무슨 얘기지? 하시는 분들이 있으실 겁니다.

1. Fetcher에게 Consumer에 할당된 Topic partition을 가져오라고 명령합니다.
2. ConsumerNetworkClient의 send 요청을 보낸다.

이 부분이 호출되는 곳의 코드를 보면 KafkaConsumer의 pollForFethes라는 메서드가 있습니다.

정리해보면 위의 첫 번째, 두 번째 단계는 아래 1번 해당 코드 부분에서 호출되는 부분입니다.

sendFetches메서드에 노드별 ClientRequest 세팅을 해놓았고 위에서 unsent에 push 해놨습니다. (노드에게 ClientRequest 보내기전 준비를 마친상태)

2번 코드 부분에서 실제 ConsumerNetworkClient의 poll메서드를 호출시켜 unsent에 담겨있는 것을 실제 보내는 역할을 맡고 있습니다.

큰 목차 2. ConsumerNetworkClient의 send 요청을 보낸다.
메서드 이름이 send라 실제로 네트워크를 타고 보낸다고 해석할 수 있습니다. 하지만 여기선 노드별 ClientRequest를 준비하는 과정이라는 게 더 맞을 것 같습니다.

추가적으로 2번 코드 부분의 client는 ConsumerNetworkClient 객체입니다.

큰 목차 1번,2번을 정리해보면 아래와 같은 그림이 되겠네요

3. ConsumerNetworkClient가 NetworkClient에게 실제 요청을 보내고 응답 값을 callback으로 completedFetches에 받는다.

poll은 크게 4가지의 역할을 합니다

NetworkClient에게 ClientRequest 전송
NetworkClient에게 응답값 가져오기
ClientRequest를 보냈기 때문에 unsent 비우기
callbck 메서드 실행

그럼 이제 위 2번 코드 poll 메서드 안으로 들어가 봅시다.

1번 코드에서는 unsent에 담아놓았던 거를 실제로 NetworkClient에게 보내는 요청이 들어갑니다.

(코드를 보려면 더보기를 눌러주세요)

여기서 client는 NetworkClient입니다.

해당 메서드를 타고 들어가면 doSend를 호출합니다.

맨 아랫줄을 보면 실제로 selector를 통해서 통신하는 게 보이네요.

그 윗줄인 inFlightRequests를 봐봅시다.

NetworkClient는 브로커로 요청을 보냈는데 아직 응답이 오직 않은 것들을 inFlightRequests에 담아서 관리합니다.

현재 전송을 보냈기때문에 inFlightRequests에 담아줍니다.

2번 코드에서는 NetworkClient에게 요청을 보냈으니 NetworkClient의 poll을 호출하여 응답 값들을 받아오려고 합니다.

(코드를 보려면 더보기를 눌러주세요)

1번 코드는 selector를 통해 통신하니 실제 거기서도 응답 값을 가져오는 코드입니다.

2번 코드는브로커 요청 완료되어 있는 것을 확인해줄 수 있는 this.selector.completedSends()를 호출합니다.

해당 노드로 보낸 inFlightRequests의 가장 오래된 request를 확인하여 브로커의 응답이 필요하지 않은 것은 InflighRequests에서 빼는 것을 확인할 수 있습니다.

저희의 요청은 expectResponse: true이기 때문에 해당 if문으로 들어가지 않습니다.

3번 코드는 selector에게 수신받은 것에 대하여 응답 확인한 것에 대해 InflighRequests에서 빼는 것을 확인할 수 있습니다.

4번 코드는 2,3번을 통해 response가 끝난 것들에 대해 onComplete()를 호출합니다.

기억나실지 모르지만 RequestFutureResponseHandler 기억나실까요?

다시 한번 사진을 가져오겠습니다.

여기서 위의 oncomplete()결과로 pendingCompletion에 추가가 됩니다.

현재 저희는 NetworkClient안에서 놀고 있지만 ConsumerNetworkClient에게도 이 응답사실을 전파해야 합니다.

이 역할을 pendingCompletion이 이어주고 있다고 생각하면 편합니다.

3번 코드는 unsent에 담겨있던 ClientRequest는 다 전송했으니 비워줍니다.

4번 코드는 pendingCompletion에 넣어있는 것들을 꺼내서 callback메서드를 실행합니다.

ConsumerNetworkClient#firePendingCompletedRequests

아까는 RequestFutureCompletionHandler에 의해 pendingcompletion에 추가하였고
이번엔 pendingCompletion을 꺼내 pendingCompletion에 등록된 callback메서드를 fireCompletion으로 인해 호출되는것을 볼 수있습니다.

아래의 fireCompletion코드를 보면 future.complete(response)가 보이실 겁니다.

ConumerNetwrokClient#RequestFuturercompletionHandler

위에 작성한 글을 다시 가져와 보겠습니다.

해당 Callback메서드가 호출되는 시점은 누군가에 의해 성공했다고 판단되었고 onSuccess의 매개변수에 ClientResponse가 담아져 있었을 겁니다.

fireCompletion에 호출되는 동시에
등록된 response를 onSuccess(ClinetResponse resp)의 매개변수로 넘겨주는 거를 확인할 수 있습니다.

언제 이 callback을 주입했을까요? 맨 처음 부분의 Fetcher#sendFetches부분에서 넣어줬습니다.

이제 completedFetches에 응답 값 데이터를 넣어줄 수 있게 됐네요.

poll메서드 안에서 처음에 그리고 마지막 firePendingCompletedRequests() 작업이 두 번이나 호출하는것을 볼수 있습니다.
poll은 주기적으로 호출되는 메서드이기 때문에 전에 요청한 poll 부분이 이제 응답이 완료되어서 먼저 처리해주는 겁니다.
그리고 try문안에서 네트워크 전송 요청이 끝났기 때문에 다시 한번 firePendingCompletedRequests를 호출하는 거죠