[학습 테스트로 배워보는 kafka] 5. 학습 테스트로 카프카 컨슈머(kafka consumer) 알아보기

---

해당 시리즈에서 제공하는 모든 소스코드는 github repository 에서 제공됩니다. 자세한 코드와 테스트 케이스는 github repository 에서 확인해주세요.

GitHub - my-research/kafka: apache kafka docs & practical usages(with spring) that i composed

 

이번 [학습 테스트로 배워보는 kafka] 시리즈는 아래 순서대로 챕터가 구성되고, 시리즈 외로 kafka 관련하여 더욱 많은 학습 정보는 kafka 심화 세션 에서 확인할 수 있습니다.

시리즈 목차

1. 시리즈를 시작하며
2. kafka 빠르게 훑어보고 아는체하기
3. kafka 컨셉과 용어 정리
4. 학습테스트 준비하기
5. 학습 테스트로 kafka producer 알아보기
6. 학습 테스트로 kafka consumer 알아보기 <-- 현재 글
7. 학습 테스트로 partitioning 알아보기
8. 학습 테스트로 consumer group 과 rebalancing 알아보기

학습의 단계별 순서로 목차가 구성되어있으므로 선행되어야 하는 챕터가 존재합니다

---

 

앞서 우리는 메시지를 카프카 브로커로 발행하는 producer 의 java client 에 대해서 알아보았다.

 

이번 시간에는 카프카 브로커에 저장된 로그 메시지들을 consume 하는 consumer java client 인 KafkaConsumer 에 대해서 학습테스트를 통해 알아볼 예정이다

KafkaConsumer

KafkaConsumer 클래스는 지난 Producer 와 마찬가지로 카프카 브로커의 특정 토픽에 해당하는 이벤트 로그 Record 를 consume 하는 역할을 수행한다

 

Kafka 의 consumer 는 Connection Pooling 과 네트워크 프로토콜을 관리한다.

 

카프카는 다른 일반적인 메시지 큐와 다르게 메시지를 소비하더라도 해당 메시지를 삭제하지 않는다.

 

요즘 많이 사용하는 다른 메시지 큐인 sqs 를 보면, 메시지 ack 를 거치고 다면 DeletionPolicy 를 설정할 수 있는데, 카프카는 삭제가 없으니 관련 삭제 정책이 없다.

 

그래서 언제든지, 누구든지 topic 에 대한 event log 를 replay 할 수 있다.

 

이 컨셉이 바로 다른 메시지 미들웨어들과 카프카가 구분되는 중요한 이유라고 볼 수 있다.

 

KafkaConsumer 인스턴스 생성하기

 

생성자의 parameter 로 전달된 key-value 형태의 properties 를 통해서 client configuration 을 등록한다

 

 

해당 configuration 의 자세한 사항과 특정 설정들은 confluent.io/kafka config 에서 확인할 수 있다

 

앞선 프로듀서 테스트와 마찬가지로 여러 테스트에서 사용할 수 있도록 Test Hepler 클래스를 통해 쉽게 컨슈머 인스턴스를 생성할 수 있도록 분리하였다

 

public class KafkaConsumerTestHelper {

  public static KafkaConsumer<String, String> simpleConsumer() {
    Map<String, Object> props = Map.of(
            "bootstrap.servers", "localhost:9092",
            "group.id", "my-consumer",
            "enable.auto.commit", "true",
            "auto.offset.reset", "earliest",
            // kafka 로 소비할 message 의 s/d 클래스 설정
            "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer",
            "value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"
    );

    return new KafkaConsumer<>(props);
  }
}

 

여기서 눈여겨볼 설정은 enable.auto.commit 과 auto.offset.reset 인데 이는 추후 컨슈머의 commit 과 offset 파트에서 다시 이야기를 하도록 하겠다. (stay tuned!!)

 

테스트 1. KafkaConsumer 를 통한 메시지 소비 테스트

 

이번 테스트에서는 발행된 메시지를 KafkaConsumer 를 통해 소비해보도록 하겠다

 

테스트 분석

• 주석 1.
  • 토픽에 메시지를 발행한다
  • consumer 테스트를 쉽게 하기 위해 테스트용 토픽에 메시지를 produce 하는 Helper 클래스가 사용되었다. 자세한 코드는 github 에서 확인할 수 있다
• 주석 2.
  • 카프카 컨슈머 인스턴스를 특정 토픽에 subscribe 시킨다.
  • 카프카 컨슈머 인스턴스는 한번에 여러개의 토픽에 subscribe 할 수 있다
• 주석 3.
  • subscribe 가 되었다고 해서 레코드를 계속해서 consume 하는것이 아니다.
  • 실제로 이벤트 로그를 fetch 하는 일은 poll() 메서드에서 수행된다
  • 결과로 ConsumerRecords 를 반환한다. 여러개의 토픽에 대해 consume 이 가능하기 때문에 record 자체도 여러개다.
  • timeout 에 대한 Duration 을 2초로 지정한다
• 주석 4.
  • 특정 토픽에 메시지가 잘 소비되었는지 검증하는 단언문이다. 역시 헬퍼 클래스이며 자세한 코드는 github 에서 확인할 수 있다

 

test 2. 컨슈머의 multi topic consume 테스트

 

앞선 테스트에서 subscribe() 메서드를 통해서 여러개의 토픽에 대해 소비를 할 수 있다고 하였다.

 

실제로 테스트해보자

 

 

helper 클래스를 통해 2개의 서로 다른 토픽에 메시지를 수행했다.

 

KafkaConsumer 인스턴스에 서로 다른 앞선 2가지 토픽을 subscribe 하였고 마찬가지로 poll() 을 수행했다.

 

그리고 Records 단언을 통해 2개의 메시지를 소비하는지 확인하였고, 성공하였다

 

kafka 의 offset 과 commit

 

앞선 여러 시간에 걸쳐서 계속해서 강조해온 중요한 내용이 있다.

 

카프카는 애플리케이션에서 발생하는 이벤트에 대한 로그 시스템이다.

 

로그는 시스템 내에서 일어난 일련의 사건들이다.

 

우리는 일반적으로 발생한 로그들 사이에 어떤 새로운 로그를 추가하려하지 않는다. 단지 로그는 계속해서 쌓일 뿐이다.

 

또한 우리는 특정 시점에 발생한 로그들을 언제든 다시 조회할 수 있다.

 

카프카는 로그의 append only 와 seeking by offset 특성을 따르는 log based architecture 를 지향한다.

 

앞에서도 설명했듯이 카프카는 메시지를 소비하면, 다른 MQ 와 달리 메시지를 삭제(pop) 하지 않는다.

 

그럼 consumer 는 도대체 어떻게 자신이 가져가야할 message 에 대해서 알 수 있을까?

 

정답은 바로 offset 과 commit 에 있다.

 

offset 과 commit

 

kafka 에서 producer 가 토픽에 저장한 메시지들은 linear 하게 쌓인다.

 

그럼 메시지들은 각각의 위치 정보(offset)을 갖게 되는데, 카프카 컨슈머는 이 위치 정보(offset)를 통해서 어디까지 메시지를 소비하였는지알 수 있다.

 

이러한 위치 정보(offset)는 kafka 내부의 __consumer_offsets 라는 특별한 토픽에 저장한다.

 

그리고 컨슈머가 메시지를 소비하고 나면 본인이 처리한 위치 정보(offset)를 업데이트 (commit) 한다.

 

commit 이라는 과정을 거치고 나면 다음 poll 타임에 새로 update 된 offset 부터 일정량만큼 메시지를 소비하게 되고, 앞선 과정을 반복하게 된다.

왜 이렇게 offset 과 commit 이라는 개념을 만들었을까?

 

offset 과 commit 을 이용한다면 두가지의 큰 장점이 있다.

1. 메시지 replay
2. 안정성

 

kafka 가 다른 MQ 와 다른 이유는 message 를 pop 을 하지 않기 때문이라고 했다.

 

message 를 pop 하지 않고 메시지 중복을 막기 위해서는 offset 과 같은 위치정보가 필요하다.

 

그래야 특정 컨슈머가 이미 소비한 메시지에 대해서 중복해서 소비하지 않도록 한다.

 

이렇게 offset 을 저장하고 관리하면서 장애 상황에 대해 더 안정적으로 운영할 수 있게 된다.

 

장애가 발생하고 장애가 복구되었다면 마지막으로 commit 한 offset 부터 다시 읽으면 되기 때문이다

commit 의 종류 2가지

 

이러한 commit 은 자동/수동 commit 이 존재한다.

 

auto commit

 

auto commit 은 메시지를 가져올 때마다 commit 을 하는 방법이다.

 

일정 주기마다 혹은 메시지를 소비한 후에 자동으로 offset 을 commit 하는 방법이다.

 

KafkaConsumer 에서는 poll() 을 호출할 때 가장 마지막 offset 을 commit 한다.

 

이 방법의 경우, 오프셋을 관리하는 가장 편리한 방법으로 알려져있다.

 

하지만 메시지의 중복 처리 문제가 발생할 수 있는데, 만약 메시지를 소비했지만 poll 을 호출하기 전에 장애가 발생한다면 offset update 가 되지 않았으므로 메시지가 중복해서 소비된다.

 

manual commit

 

manual commit 은 메시지 처리가 완료될 때까지 commit 하지 않는다.

 

일반적인 acknowledgement 와 동일하게 동작한다.

 

컨슈머가 메시지를 처리한 뒤, 명시적으로 처리된 offset 을 commit 하는 방식이다

 

이 방법의 경우, 메시지 소비 자체는 느릴 수 있으나 메시지 처리의 정확성과 무손실을 보장하는 방법이다.

 

이제 테스트를 통해 알아보자

 

test 3. auto commit test

 

자동 커밋을 테스트해보자

 

테스트 설명

• 주석 1
  • 한번 메시지를 poll 할 때 3개의 메시지를 소비하도록 설정한다
• 주석 2
  • auto commit mode 를 활성화한다
• 주석 3
  • 메시지를 발행한다
  • 3개 단위로 알파벳, 이모지 순서로 발행한다.
• 주석 4
  • 첫번째 poll
  • 3개 단위이므로 알파벳만 소비된다
• 주석 5
  • 두번째 poll
  • 3개 단위이므로 이모지가 소비된다

☑️ expect: 자동 커밋 모드이므로 두 번째 poll() 호출시 이모지 메시지가 소비된다

테스트 결과

테스트는 성공한다

학습 테스트 정리

• ✅ Consumer 는 Record 라는 단위로 카프카 브로커로 메시지를 소비
• ✅ 실제 메시지 소비는 poll() 을 통해 수행됨
• ✅ subscribe() 를 통해 여러 토픽에 대해 컨슈머 등록을 할 수 있음
• ✅ consumer 는 commit mode 를 통해 자동/수동으로 commit 할 수 있음