[학습 테스트로 배워보는 kafka] 4. 학습 테스트로 카프카 프로듀서(kafka producer) 알아보기

---

해당 시리즈에서 제공하는 모든 소스코드는 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 알아보기

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

---

 

이번시간에는 드디어 카프카 브로커로 메시지를 발행해보는 시간이다

 

KafkaProducer

 

java 에서 카프카 클러스터로 메시지를 발행할 때 사용하는 java client 는 KafkaProducer 이다

 

KafkaProducer 는 메시지 발행을 비동기로 수행하기 때문에 많은 양의 Record 전송을 지원한다.

 

KafkaProducer 인스턴스 생성하기

 

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

 

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

 

나는 앞으로 많은 테스트에서 Producer 인스턴스를 생성할 예정이라서 Helper 클래스로 producer 생성 로직을 분리시켰다

 

public class KafkaProducerTestHelper {
  public static KafkaProducer<String, String> getSimpleProducer() {
    Map<String, Object> props = Map.of(
      // bootstrap server 설정
      "bootstrap.servers", "localhost:9092",
      // kafka 로 전송할 message 의 s/d 클래스 설정
      "key.serializer", "org.apache.kafka.common.serialization.StringSerializer",
      "value.serializer", "org.apache.kafka.common.serialization.StringSerializer"
    );
    return new KafkaProducer<>(props);
  }
}

 

그리고 각각의 테스트 케이스에서는 Producer 의 인스턴스를 다음과 같이 만들었다

 

 

이러한 방법을 통해 앞으로 진행할 테스트에 대해서 프로듀서 인스턴스를 생성할 예정이다

 

test 1 - KafkaProducer 를 이용한 메시지 발행 테스트

이번 테스트에서는 kafka producer 를 이용해서 메시지를 발행해보려한다.

 

테스트 코드 분석

• 주석 1. 발행할 메시지를 정의
  • 일반적으로 카프카에 저장될 이벤트 로그는 Record 라는 개념의 데이터가 저장된다.
  • 첫번째 파라미터로 topic 을 넘기고 두번째 파라미터로 message payload 를 넘긴다
• 주석 2. 실제로 producer 를 통해서 record 를 카프카 브로커로 전송한다
  • produce 결과로 Future 를 반환받는다.
• 주석 3. Future.get() 연산을 통해 비동기 produce response 를 blocking 한다
• 주석 4. produce 결과로 반환된 정보에 대한 단언문

정리

• ✅ Producer 는 Record 라는 단위로 카프카 브로커로 메시지를 발행
• ✅ produce 결과를 asynchronous 하게 처리 가능
• ✅ 결과를 blocking 하여 확인하지 않으면 일부 메시지에 대해 누락이 발생할 수 있음

test 2. KafkaProducer 존재하지 않는 topic 에 대해서 메시지 발행 테스트

 

producer 가 메시지를 발행하였는데 만약 존재하지 않는 topic 에 발행했다면 어떻게 될까?

 

왜 우리는 앞선 테스트에서 토픽 생성을 하지 않아도 테스트가 성공했을까?

 

그 정답은 바로 Kafka 의 기본 설정에 있다.

 

다른 많은 라이브러리들과 마찬가지로 kafka 역시 실행 옵션 (configuration) 들을 필요로 하는데, 기본적으로 topic 생성 기능이 활성화 되어있다.

 

그래서 토픽을 사전에 생성하지 않더라도 메시지를 발행할 때 존재하지 않는 토픽에 발행을 하더라도 문제가 없었던 것이다.

 

존재하지 않는 topic 에 발행을 하는 테스트를 위해 @EmbeddedKafka 설정을 바꿔보자

 

어노테이션의 파라미터로 brokerProperties 에 "auto.create.topics.enable=false" 를 넘겨서 토픽 자동 생성 기능을 끄고 아래의 테스트를 실행해보자

 

 

그럼 아래와 같이 UNKNOWN_TOPIC_OR_PARTITION 에러가 발생한다.

 

나는 Junit Extension 을 통해 2초 timeout 이 발생하면 테스트를 통과하도록 익스텐션 클래스를 만들어서 추가해줬는데, 관련 코드를 더 확인해보고싶으면 kafka 학습테스트 github 에서 확인할 수 있다

정리

• ✅ Producer 는 존재하지 않는 topic 에 메시지를 발행할 수 없음
• ✅ Kafka 의 기본 설정이 존재하지 않는 topic 을 자동 생성함
• ✅ broker props 를 통해 해당 설정을 끌 수 있음

test 3. 메시지 발행 결과 처리 콜백 동작 테스트

 

test 1 에서 producer 가 produce 를 수행하면 메시지를 발행하고 Future 객체를 반환받는다고 설명했다.

 

만약 결과를 확인하고싶다면 Future.get() 연산을 통해 요청에 대해 blocking 하면 결과를 해당 스레드 내에서 확인할 수 있다.

 

하지만 대량의 메시지에 대해서 계속 block 하여 결과를 확인하는 것이 어떤 상황에선 올바르지 않은 선택일 수 있는데, 이를 위해 kafka 는 callback 이라는 것을 제공한다

 

이번 테스트에서는 kafka producer 가 제공하는 callback 이라는 기능에 대해서 알아보자.

 

KafkaProducer 는 메시지 발행 요청이 완료되었을 때에 대해 결과 처리를 위한 callback 을 등록할 수 있다

 

callback 은 org.apache.kafka.clients.producer.Callback 인터페이스를 implements 해야한다

 

 

콜백인터페이스를 구현하고 KafkaProducer.send() 의 두번째 인자로 해당 콜백을 넘기면 내부적으로 kafka broker 로 record 저장 요청을 마쳤을 때 해당 콜백을 실행시킨다.

 

테스트를 위한 콜백을 아래와 같이 구현해줬다.

 

@Slf4j
public class SimpleProduceCallback implements Callback {

  public static SimpleProduceCallback newOne() {
    return new SimpleProduceCallback();
  }

  @Override
  public void onCompletion(RecordMetadata metadata, Exception exception) {
    if (nonNull(metadata)) {
      log.info("✅ 메시지 발행 성공, topic: {}, partition: {}, offset: {}, timestamp: {}",
          metadata.topic(),
          metadata.partition(),
          metadata.offset(),
          metadata.timestamp());

    } else {
      log.error("❌ 메시지 발행 실패", exception);
    }
  }
}

onCompletion() 메서드의 첫번째 인자로 들어오는 metatdata 는 성공적으로 메시지를 발행했을 때 정상적인 값이 세팅되어 전달된다

 

만약 전송에 실패하였다면 -1 이 설정된 metatdata 객체를 반환받을 것이다.

 

아래 테스트를 실행해보자

 

 

마지막 라인에서 sut.close() 를 추가해준 이유는 테스트 케이스가 종료되는 시점이 callback 이 호출되는 시점보다 느리기때문에 테스트 검증을 위해 blocking 하였다.

 

 

결과를 확인하면 성공적으로 로그가 나오는 것을 확인할 수 있다

 

정리

• ✅ Producer 는 레코드 전송 요청 결과를 2가지 방법으로 확인할 수 있다
• ✅ Future.get 연산을 통해 결과를 blocking 하여 확인할 수 있다.
• ✅ Callback 을 등록하여 non-blocking 방식으로 확인할 수 있다