KafkaConsumer

💡 이번 포스팅에서는 KafkaConsumer에 대해 정리해 보도록 하겠습니다.

 

이전에 KafkaProducer 객체를 통해 카프카 토픽에 메시지를 전송해 보았습니다.

이번에는 KafkaConsumer 객체를 통해 프로듀서가 전송한 메시지를 읽어오도록 하겠습니다.

 

public static void main(String[] args) {

    String topicName = "simple-topic";

    Properties props = new Properties();
    props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.111.133:9092");
    props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "simple-group");
    props.setProperty(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "5000"); // optional
    props.setProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "90000"); // optional
    props.setProperty(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "600000"); // optional

    try (KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props)) {
        kafkaConsumer.subscribe(List.of(topicName));

        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000));

            consumerRecords.forEach(record -> {
                logger.info("record key: {}, record value: {}, partition : {}",
                        record.key(), record.value(), record.partition());
            });
        }
    }
}

 

가장 기본적인 KafkaConsumer 객체 생성 코드입니다.

Property로 GroupId 까지는 필수 값으로 설정해야 하며, 이후 나머지 옵션들은 선택 값으로 입력합니다. (입력하지 않을 경우 기본 값이 적용됩니다.)

 

kafkaConsumer.subscribe(List.of(topicName));

생성한 컨슈머가 어떤 토픽의 메시지를 읽어올지 토픽명을 전달합니다.

가장 보편적인 방법으로 컨슈머가 읽어올 토픽을 리스트로 전달합니다.

여러 개의 토픽을 전달하게 되면 각 토픽의 리더 파티션을 찾아 컨슈머가 메시지를 읽어옵니다.

 

 

ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000));

컨슈머가 poll() 메서드를 호출함으로써 메시지를 읽어옵니다.

이때 Duration을 설정할 수 있는데 이 시간은 컨슈머가 메시지가 오기를 기다리는 최대 시간입니다.

이전 포스팅에서 정리했듯 컨슈머는 Fetcher 객체에 저장되어 있는 Linked Queue 형태의 메시지를 가져오게 되는데 Fetcher 객체에 메시지가 없을 경우 기다릴 최대 시간입니다.

 

말 그대로 최대 시간이기 때문에 아래와 같은 옵션들을 변경하면 최대 시간만큼 기다리지 않고도 설정한 옵션의 사이즈만큼 메시지가 쌓이면 바로 컨슈머로 전송하게 됩니다.

https://soono-991.tistory.com/81

 

Manual Commit

위에서는 따로 enable.auto.commit 옵션을 설정하지 않았기 때문에 AutoCommit이 적용되었습니다.

enable.auto.commit을 false로 설정한 후 직접 commitSync() , commitAsync() 메서드를 호출하여 수동으로 오프셋을 커밋할 수 있습니다.

props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

 

private static void pollCommitSync(KafkaConsumer<String, String> kafkaConsumer) {
    try (kafkaConsumer) {
        int loopCount = 0;
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000));

            logger.info(" ###### logCount : {} consumerRecords count: {}", loopCount++, consumerRecords.count());

            consumerRecords.forEach(record -> {
                logger.info("record key: {}, record value: {}, partition : {}, record offset: {}",
                        record.key(), record.value(), record.partition(), record.offset());
            });

            try {
                if (consumerRecords.count() > 0) {
                    kafkaConsumer.commitSync();                      // 수동 커밋
                    logger.info("commit sync has been called");
                }
            } catch (CommitFailedException e) {
                logger.error(e.getMessage());
            }
        }
    } catch (WakeupException e) {
        logger.error("wakeup exception has been called");
    } catch (Exception e) {
        logger.error(e.getMessage());
    } finally {
        logger.info("finally consumer is closing");
    }
}

 

메시지를 읽는 로직은 동일하지만 kafkaConsumer.commitSync() 메서드를 호출해야 오프셋 정보가 커밋됩니다.

하지만 commitSync()를 호출하면 Main Thread가 오프셋 정보를 커밋할 때까지 block 되기 때문에 이 부분이 싫다면 commitAsync() 메서드를 사용합니다.

 

kafkaConsumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        logger.error("offsets {} is not completed, error: {}", offsets, exception.getMessage());
    }
});

 

commitSync()와 달리 commitAsync()는 OffsetCommitCallback을 인자로 받습니다.

commitAsync()는 오프셋 커밋을 비동기로 실행하기 때문에 커밋 실행 결과를 callback으로 전달받게 됩니다.

이때 callback에는 두 개의 파라미터가 담겨 있습니다.

 

offsets에는 커밋한 오프셋과 관련된 메타 데이터들이 담겨있으며, 오프셋 커밋에 실패할 경우 해당 에러/예외가 Exception 객체에 담겨 전송됩니다.

callback에서는 Exception 파라미터가 null인지 여부에 따라 오프셋 커밋에 성공/실패했는지를 확인할 수 있습니다.

 

동기 방식의 오프셋 커밋은 커밋에 실패할 경우 다시 재시도합니다.

반면 비동기 방식의 오프셋 커밋은 커밋에 실패하여도 다시 재시도를 하지 않습니다.

 

오프셋 커밋의 경우 여러 이유로 인해 메시지 중복 또는 유실이 발생한다고 합니다.

이 부분을 정확히 한번 전송(exactly-once)을 사용하면 해결할 수 있지만 별도의 설정이 필요합니다.

관련해서 정리된 포스팅이 있어 참고하시면 좋을 것 같습니다.

 

[KAFKA] Consumer 메시지 중복 처리

 

Kafka 메시지 처리 중복 or 누락 문제 - 한현상 블로그

 

 

특정 파티션만 할당

컨슈머의 subscribe() 메서드에 토픽명을 전달하면 해당 토픽의 메시지를 컨슈머가 읽어온다고 했습니다.

만약 특정 파티션만 할당하여 메시지를 읽어오고 싶은 경우에는 assign() 메서드를 사용하여 파티션 정보를 전달하면 됩니다.

 

// 토픽 파티션 생성
TopicPartition topicPartition = new TopicPartition(topicName, 0);

// 파티션 정보 전달
kafkaConsumer.assign(List.of(topicPartition));

 

특정 파티션의 특정 오프셋부터 메시지 읽기

kafkaConsumer.seek() 메서드를 사용하면 특정 파티션의 특정 오프셋부터 메시지를 읽어올 수 있습니다.

// 토픽 파티션 생성
TopicPartition topicPartition = new TopicPartition(topicName, 0);

// 파티션 할당
kafkaConsumer.assign(List.of(topicPartition));

// 오프셋 설정
kafkaConsumer.seek(topicPartition, 10);

 

 

참고 자료

• https://camel-context.tistory.com/54
• https://tillog.netlify.app/posts/kafka-message-issue
• <실전 카프카 개발부터 운영까지> - 고승범 지음
• <인프런 카프카 완벽 가이드 - 코어편> - 권철민