인프런 커뮤니티 질문&답변

답변 감사드립니다! 제가 질문 드렸던 부분은 Confluent Kafka의 오픈소스 버전이라 따로 url을 링크드리긴 어려울 것 같습니다.

대신에 철민님께서 IDE에서 아래 클래스 및 메서드를 확인해봐주시면 좋을 것 같습니다.

1. ConsumerNetworkClient.poll() 메서드

public void poll(Timer timer, PollCondition pollCondition, boolean disableWakeup) {
    
    lock.lock();
    try {
        ...
        long pollDelayMs = trySend(timer.currentTimeMs());

   
        if (pendingCompletion.isEmpty() && (pollCondition == null || pollCondition.shouldBlock())) {
            ...
            client.poll(pollTimeout, timer.currentTimeMs());
        } else {
            client.poll(0, timer.currentTimeMs());
        }
        ...
        trySend(timer.currentTimeMs());

        ...
    } finally {
        lock.unlock();
    }

    ...
}

• 위 메서드를 보면 client.poll()을 하기 전/후로 trysend()라는 메서드를 호출하고 있습니다.

• 여기서 client는 ConsumerNetworkClient 내부에 있는 NetworkClient이고, 이녀석은 내부적으로 nioSelector를 가지고 있어서, nioSelector를 통해 실제로 Broker에 통신을 하는 것처럼 보입니다.

• trysend()에서 뭔가 메세지를 보내는 역할을 합니다.

2. ConsumerNetworkClient.send()

long trySend(long now) {
   ...
    for (Node node : unsent.nodes()) {
        Iterator<ClientRequest> iterator = unsent.requestIterator(node);
        if (iterator.hasNext())
            pollDelayMs = Math.min(pollDelayMs, client.pollDelayMs(node, now));

        while (iterator.hasNext()) {
            ClientRequest request = iterator.next();
            if (client.ready(node, now)) {
                client.send(request, now);
                iterator.remove();
            } else {
                // try next node when current node is not ready
                break;
            }
        }
    }
    return pollDelayMs;
}

• ConsumerNetworkClient는 unsent라는 리스트 형태의 자료구조를 가지고 있습니다. 이 자료구조는 ConsumerNetworkClient가 보내야 하지만 아직 보내지 못한 요청을 가지고 있습니다. 이 요청에는 Find Cooridnator / Fetch / Commit / Heartbeat 등의 API가 담겨져 있습니다.

• client.ready() 메서드를 이용해서 현재 client / node의 상황을 판단하고 보낼 수 있는 경우에는 unsent에 포함되어 있는 모든 요청을 보내는 것처럼 보입니다. 예를 들어 unsent에 [HeartBeat API - HeartBeat API - Fetch API - Fetch API - HeartBeat API]가 존재한다면 trysend() 메서드를 한번 실행하면 unsent = []가 될 것입니다.

제가 살펴봤을 때는 Kafka-Heartbeat-Thread와 Main Thread는 모두 ConsumerNetworkClient.send()를 호출하고 있습니다. 위의 로직을 살펴봤을 때, HeartBeat-Thread가 어떤 요청을 보내야는지 구별할 수 없기 때문에 Heartbeat Thread는 자신이 HeartBeat Thread라고 할지라도 unsent에 Fetch API가 있으면 그 요청을 바로 보내주는 것으로 예상됩니다!

혹시 더 필요하신 부분이 있으시다면 언제든지 알려주세요!

항상 좋은 강의와 답변 감사합니다 :)

와우, 이렇게까지 내부 소스코드를 분석하시다니 대단하시군요.

제 생각엔,

말씀하신대로 unsent는 ConsumerClientNetwork이 broker에 보내야 할 Request들을 가지고 있습니다. 가령 broker의 Consumer Group Coordinator와의 통신을 위한 Request라든지...

근데 이들 Request는 NetworkClient 비동기적으로 broker에 요청을 보내지만, 아직 보내지지 않은 Request도 있을 수 있습니다. 이때 ConsumerClientNetwork()의 poll()을 호출하면 trysend()로 기존에 unsent에 있는 Request를 먼저 broker로 다 보낸 뒤에 NetworkClient의 poll()이 다시 호출되어 broker에서 메시지를 받아오는 처리를 하게 됩니다.

unsent에 들어가 있는 Request가 fetch request, heartbeat request도 들어갈 수는 있겠지만, unsent는 브로커로 보내지 못한 request를 가지고 있을 뿐이고 fetch의 경우는 poll()을 호출했을 때만 수행이 됩니다.

그러니까, 외부 Consumer Client가 poll( )을 호출하면 ConsumerNetworkClient가 trysend()로 기존에 쌓여있는 Request를 Broker로 일단 먼저 보내고, NetworkClient가 poll( )을 호출해서 데이터를 브로커로 부터 가져오게 됩니다.

NetworkClient가 비동기 I/O Selector로 구성되었다 하더라도, 이건 Request와 RequestCompletion, 그리고 poll()을 비동기적으로 작업을 한다는 의미지 NetworkClient 혼자 poll() 수행해서 스스로 브로커로 부터 가져올 수는 없습니다. 외부 Consumer Client에서 fetch를 위한 poll()또는 Heartbeat Thread에서 poll()이 수행되어야 NetworkClient가 poll( )을 수행 할 수 있습니다.

그런데 commit sync 방식일 경우에는 poll ( ) 수행 후 commit 시 까지는 blocking이 걸리기 때문에 commit 작업을 수행하는 도중에 NetworkClient가 스스로 브로커로 부터 메시지를 poll( )하지는 않습니다. 물론 이 부분은 100% 확실하지는 않습니다만 소스 코드 분석과 Consumer의 내부 메커니즘으로 판단컨데 제 생각으로 (확실히) 그럴거라 생각합니다.

감사합니다.