Kafka 컨슈머 만들기: 동작 흐름과 기본 설정 이해하기

개요

메시지를 보내는 것 만큼, 받는 것도 중요하다. 카프카 프로듀서에 이어 이번에는 컨슈머에 대해 공부해봤다. 처음에는 poll 메서드가 뭘 하는지, subscribe 메서드는 왜 필요한지, 설정값들은 왜 이렇게 많은지 조금 헷갈렸는데 책을 읽으며 예제 코드를 하나씩 다라가며 개념을 정리해보니 어느 정도 감이 잡히기 시작했다.  이 글은 그 과정을 기록한 것으로, 컨슈머의 기본 동작부터 주요 설정값, 그리고 토픽을 구독하고 메시지를 읽어오는 방식까지 이해한 대로 정리해봤다. 

 

1.카프카 컨슈머란?

카프카에서 메시지를 읽는 주체를 컨슈머(Consumer)라고 한다. 메시지를 보낼 떄는 프로듀서(Producer)가 사용되며, 그 메시지를 읽고 처리하는 역할을 담당하는 것이 바로 컨슈머이다. 

카프카는 모든 메시지를 '토픽'이라는 단위에 저장하고, 컨슈머는 이 토픽에서 메시지를 꺼내가는 방식으로 작동한다.다. 

 

1.1 카프카 컨슈머 그룹

컨슈머는 일반적으로 '컨슈머 그룹' 이라는 단위로 묶여서 작동한다. 동일한 컨슈머 그룹에 속한 여러 컨슈머가 하나의 토픽을구독하는 경우, 각 컨슈머는 서로 다른 파티션에서 메시지를 읽게 된다. 그룹 내부에서는 같은 메시지를 여러 번 읽지 않도록 파티션을 나누어 처리하지만, 다른 그룹에 속한 컨슈머는 동일한 메시지를 읽을 수 있다.

 

예를 들어, 4개의 파티션을 가진 토픽1이 있다고 가정하고, 컨슈머1이 컨슈머 그룹1 소속으로 이 토픽을 구독하고 있다면, 컨슈머1은 4개의 파티션 모두에서 메시지를 수신하게 된다.

 

4개의 파티션과 1개의 컨슈머

 

이후 컨슈머2가 같은 그룹에 추가되면, 카프카는 자동으로 파티션을 재분배하여 각 컨슈머가 2개의 파티션에서 메시지를 읽도록 조정한다.

 

컨슈머 그룹에 컨슈머2를 추가했다.

 

컨슈머가 4개까지 늘어나면, 각 컨슈머는 하나의 파티션에서 메시지를 읽게 된다. 

각각 하나의 파티션에서 메시지를 읽고 있다.

 

하지만 컨슈머 수가 파티션 수보다 많아지면, 일부 컨슈머는 할당받을 파티션이 없어 유휴 상태로 남게 된다. 

 

유휴 컨슈머가 생겼다.

 

새로운 컨슈머 그룹을 생성하면, 해당 그룹은 카프카 입장에서 완전히 새로운 구독자로 간주되므로 동일한 토픽의 메시지를 다시 읽을 수 있다. 

새로운 컨슈머 그룹이 추가되었다. 두 그룹 모두 모든 메시지를 받는다.

 

대량의 메시지를 더 빠르게 소비하려면 파티션 수를 늘리고, 컨슈머를 확장하거나 컨슈머 그룹을 적절히 활용해야 한다.

 

 

2. 파티션 리밸런스란?

컨슈머 그룹에 속한 컨슈머들은 자신들이 구독하는 토픽의 파티션들을 나눠 가진다. 리밸런스란 이 파티션들을 컨슈머 간에 다시 재할당 하는 과정을 말한다. 컨슈머가 종료되거나, 새로운 컨슈머가 그룹에 추가되는 등의 이벤트가 발생하면 카프카는 자동으로 파티션을 재분배하기 위해 리밸런스를 수행한다.  리밸런스는 카프카 그룹에 고가용성(high availability)과 확장성(scalability)을 제공하는 중요한 기능이다.

 

2.1 조급한 리밸런스 

조급한 리밸런스 방식에서는 리밸런스가 시작되면 모든 컨슈머가 메시지 소비를 중단하고, 자신에게 할당된 모든 파티션의 소유권을 포기한다. 이후 컨슈머 그룹에 다시 참여(rejoin)하여 완전히 새로운 파티션 할당을 받게 된다.

이 방식은 단순하지만, 리밸런스가 발생하는 동안 전체 그룹이 잠시 멈추는 현상(stop the world)이 발생할 수 있다는 단점이 있다.

 

 

2.2 협력적 리밸런스 

협력적 리밸런스는 리밸런스 중에도 가능한 한 많은 컨슈머가 계쏙 메시지를 소비할 수 있도록 설계된 방식이다. 필요한 경우에만, 특정 파티션에만 재할당되고 재할당되지 않은 파티션은 기존 컨슈머가 계속 읽을 수 있다. 조급한 리밸런스의 전체 작업이 중단되는 단점 없이 리밸런스를 수행할 수 있다.

 

컨슈머는 하트비트(Heartbeat)를 통해 카프카 브로커 중 그룹 코디네이터 역할을 맡은 브로커에 자신이 살아있음을 주기적으로 알린다. 이 하트비트는 백그라운드 스레드에서 일정 간격으로 전송되며, 이 간격이 너무 길어지면 카프카는 해당 컨슈머가 죽었다고 판단하고 세션 타임아웃을 발생시킨다. 

 

카프카 2.4 버전까지는 조급한 리밸런스가 기본 값이었는데, 3.1 버전 부터는 협력적 리밸런스가 기본값으로 적용되었다.

 

 

3. 카프카 컨슈머 생성

카프카에서 메시지를 소비하려면 가장 먼저 KafkaConsumer 인스턴스를 생성해야 한다. 

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer);
props.pur("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer);

KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(props);

 

• bootstrap.servers: kafka 브로커의 주소 
• group.id: 컨슈머 그룹 ID, 같은 그룹 ID 를 사용하는 컨슈머끼리 파티션을 나누어 읽는다.
• key,value serializer: 메시지를 역직렬화할 클래스 (바이트 배열 -> 자바 객체)

컨슈머 설정에서 알아두면 좋은 옵션들

• auto.offset.reset : 초기 구독 시 일기 시작 위치를 정의한다. latest가 기본값으로 가장 최신부터 읽기 시작한다는 뜻이고, earliest는 파티션의 맨처음부터 모든 데이터를 읽는 방식이다. 
• enable.auto.commit: 컨슈머가 자동으로 오프셋을 커밋할지의 여부를 결정한다. 기본값은 true로, 오프셋을 언제 커밋할지를 직접 결정하고 싶으면 이 값을 false로 놓으면된다. 이 값이 true일 경우, auto.commit.inerval.ms를 사용해서 얼마나 자주 오프셋이 커밋될지를 제어할 수 있다. 
• max.poll.records: poll()을 호출할 때 마다 리턴되는 최대 레코드 수를 지정한다. 폴링을 반복할 때 마다 처리되는 레코드의 개수이다.
• session.timeout.ms : 이 시간 안에 하트비트를 보내지 않으면 컨슈머는 죽은것으로 간주된다. 기본값은 10초이다. 만약 하트비트를 보내지 않은 채로 session.timeout.ms가 지나가면 그룹 코디네이터는 해당 컨슈머를 죽은 것으로 간주하고, 죽은 컨슈머에게 할당되어 있던 파티션을 다른 컨슈머에게 할당해주기 위해 리밸런스를 실행시킨다. 
• hearbeat.inerval.ms : 컨슈머가 그룹 코디네이터에게 하트비트를 보내는 간격을 결정한다. 

 

3.1 토픽 구독과 메시지 폴링 루프 

컨슈머는 토픽을 구독하고 poll() 메서드를 통해 메시지를 주기적으로 읽어온다.

consumer.subscribe(List.of("my-topic"));

Duration timeout = Duration.ofMills(100);

while (true) {
	ConsumerRecords<String, String> records = consumer.poll(timeout);
    
    for (ConsumerRecord<String, String> record : records) {
    	System.out.printf("받은 메시지: key=%s, value=%s, offset=%d, partition=%d%n",
        	record.key(), record.value(), record.offset(), record.partition()
    }

}

 

• 이 루프는 무한루프이기 때문에 종료되지 않는다. 컨슈머 애플리케이션은 대부분 오랫동안 실행되며, 계쏙해서 새로운 메시지를 읽는다. 
• 컨슈머는 poll()을 주기적으로 호출하지 않으면 컨슈머는 죽은것으로 간주되어 리밸런스가 발생한다. poll()에 전달하는 매개변수는 레코드가 없을 때 기다릴 최대 시간 이다.
• poll()은 레코드들이 저장된 List 객체를 리턴한다. 각각의 레코드는 레코드가 저장되어 있던 토픽, 파티션, 파티션에서의 포스셋, 키값과 벨류값을 포함한다. List를 반복해 가며 각각의 레코드를 하나씩 처리한다. 

이 폴링루프는 생각보다 많은 일을 한다. 새 컨슈머에서 처음으로 poll()을 호출하면 컨슈머는 그룹 코디네이터를 찾아서 컨슈머 그룹에 참가하고, 파티션을 할당 받는다. 아까 학습했떤 리밸런스도 함께 여기서 처리된다. 

 

poll()이 max.poll.,interval.ms에 지정된 시간 이상으로 호출되지 않을 경우, 컨슈머는 죽은 것으로 판정되어 컨슈머 그룹에서 퇴출된다. 

따라서 폴링 루프 안에서 예측 불가능한 시간동안 블록되는 작업을 수행하는건 피해야한다. 

 

결론

카프카 컨슈머는 단순히 메시지를 읽기만 하는게 아니라, 컨슈머 그룹 안에서 역할을 분담하고, 리밸런스를 처리하는 등 내부적으로 다양한 메커니즘이 동작하고 있었다. poll 메서드는 단순히 메시지를 꺼내는 함수 정도로 생각했는데 실제로는 컨슈머 그룹에 참여하고 파티션을 할당받으며, 하트비트를 보내고 오프셋을 관리하는 일련의 동작들이 모두 그 안에서 이뤄지고 있었다. 

이번 학습을 통해 카프카에서의 데이터 읽기 흐름에 대해 조금 더 구체적으로 그림이 그려졌고, 다양한 설정 옵션들에 따라 컨슈머의 동작 방식이 달라질 수 있는지도 알게 되었다. 다음에는 컨슈머의 다양한 오프셋 커밋과 리밸런스 시에 발생할 수 있는 이슈들, 그리고 이를 제어할 수 있는 리밸런스 리스너에 대해 학습할 예정이다.

처음에는 막막하기만 했던 카프카인데, 이해한 개념을 정리해보면서 차근차근 읽어나가니 조금씩 카프카에 대해서 알것만 같다.(아마도..?) 앞으로도 이렇게 정리하면서 읽어서 최대한 내껄로 만들고 싶다.