[Kafka] Spring Kafka JSON Batch

Spring Kafka에서 제공하는 JSON Batch 모드는 Kafka에서 메시지를 읽어올 때, 여러 메시지를 한 번에 읽어와서 처리하는 방식이다.

이를 통해 처리량을 향상시킬 수 있다.

JSON Batch 모드를 사용할 경우, Spring Kafka는 지정한 배치 크기만큼 메시지를 읽어와서 리스트로 만들어 준다. 처리가 완료되면, 마지막으로 읽은 메시지의 오프셋을 커밋한다.

@KafkaListener(
    id = "컨슈머 이름",
    topics = "컨슘할 토픽명 리스트",
    groupId = "컨슈머 그룹명",
    containerFactory = "사용할 containerFactory"
)
fun consume(messages: List<String>) {
    // 처리 로직
}

JSON Batch 모드를 설정한 containerFactory 생성하는 메서드

@Bean
fun jsonKafkaBatchListenerContainerFactory(kafkaProperties: KafkaProperties): ConcurrentKafkaListenerContainerFactory<String, Object> {
    // ...
    val factory = ConcurrentKafkaListenerContainerFactory<String, Any>()
    factory.consumerFactory = consumerFactory
    // JSON_BATCH 설정
    factory.isBatchListener = true 
    // Batch 처리가 완료된 후에 Offset을 Commit
    factory.containerProperties.ackMode = ContainerProperties.AckMode.BATCH
    return factory
}

config

• max.poll.records Consumer가 Polling할 때 최대로 가져올 메시지 개수
• 이 값을 높일수록 Consumer가 한 번에 처리할 수 있는 메시지 개수가 늘어나며, Batch 처리의 성능 향상에 도움이 된다. 같은 시간 동안 더 많은 메시지를 처리할 수 있어서 Consumer가 Polling하는 빈도를 줄이고, 더 많은 메시지를 처리하도록 하는 효과가 있다.
• fetch.max.wait.ms Polling할 때 Broker로부터 메시지를 가져오기까지 기다리는 최대 시간 (기본값: 500ms)
• fetch.min.bytes Broker에서 응답할 메시지의 최소 크기
• fetch.max.bytes Broker에서 응답할 메시지의 최대 크기
• max.poll.interval.ms 두 번의 연속적인 폴(poll) 호출 사이에 걸리는 최대 시간
• 이 시간이 초과되면, Consumer는 그룹에서 제거되고 파티션은 다른 Consumer에게 재분배된다.이 값을 너무 낮게 설정하면, Consumer가 처리를 완료하기 전에 컨슈머 그룹에서 제거될 수 있다. (기본값: 300000ms(5분))
• max.partition.fetch.bytes 한 번에 한 Partition에서 가져올 수 있는 최대 메시지 크기