Kafka Consumer 클래스의 주요 구성 요소와 poll() 메서드 동작 메커니즘 이해

Kafka Consumer 클래스의 주요 구성 요소와 poll() 메서드 동작 메커니즘 이해

Kafka Consumer의 주요 구성 요소와 poll() 메서드

• 브로커나 Consumer 내부 Queue에 데이터가 없다면 바로 데이터를 반환. 그렇지 않을 경우에는 1000ms동안 데이터Fetch를 브로커에 계속 수행하고 결과 반환

ConsumerRecords<K, V> consumerRecords = KafkaConsumer.poll(Duration.ofMillis(1000));

 

Kafka Consumer 구성

 

poll() 메서드의 동작

• Linked Queue에 데이터가 있을 경우 Fetcher는 데이터를 가져오고 반환하며 poll() 수행 완료.
• ConsumerNetworkClient는 비동기로 계속 브로커의 메시지를 가져와서 Linked Queue에 저장.
• Linked Queue에 데이터가 없을 경우 ConsumerNetworkClient는 1000ms 만큼 계속해서 브로커에 메시지 요청 후 poll() 수행 완료.

while(true){
    ConsumerRecords<String, Integer> records = consumer.poll(Duration.ofMillis(1000));
    for(ConsumerRecord<String, Integer> record : records){
        String key = record.key();
        String value = record.value();
       
        # poll로 가져온 데이터로 상대적으로 시간이 걸리는 작업을 수행하는 메인 Thread
        ...
    }
}

 

Consumer Fetcher 관련 주요 설정 파라미터 이해

• fetcher.min.bytes = 16384, fetch.max.wait.ms = 500 (default)
• Fetcher는 Linked Queue 데이터를 가져오되, Linked Queue에 데이터가 없을 경우 ConsumerNetworkClient에서 데이터를 브로커로부터 가져올 것을 요청한다.

 

fetch.min.bytes

• Fetcher가 record들을 읽어들이는 최소 bytes. 브로커는 지정된 fetch.min.bytes 이상의 새로운 메시지가 쌓일때까지 전송하지 않는다.
• 기본은 1

 

fetch.max.wait.ms

• 브로커에 fetch.min.bytes 이상의 메시지가 쌓일 때까지 최대 대기 시간.
• 기본은 500ms