[Spring] 분산 시스템에서 효과적인 데이터 전달 방법 (3) - Kafka

[이전 포스팅] RDB를 사용하는 애플리케이션에서 효과적인 데이터 전달 방법

2023.11.04 - [BackEnd/Spring] - [Spring] 분산 시스템에서 효과적인 데이터 전달 방법 (1) - Transaction Outbox Pattern & Polling Publisher Pattern

 

[이전 포스팅] RabbitMQ를 사용하는 애플리케이션에서 효과적인 데이터 전달 방법

2023.11.10 - [BackEnd/Spring] - [Spring] 분산 시스템에서 효과적인 데이터 전달 방법 (2) - RabbitMQ

 

2. Kafka를 사용한 전달 방법

 

2-1. Producer Confirm

 

아래는 Producer Confirm을 구현한 클래스이다.

 

- Spring Boot 2 버전

@Slf4j
@Component
@RequiredArgsConstructor
public class Producer {

    public void sendEvent(CreateTaskEvent event) {
    
        ListenableFuture<sendResult<String, CreateTaskEvent> future = 
        kafkaTemplate.send(TOPIC_TASK, event);
        
        future.addCallback(
                result -> log.info("offset : {}" , result.getRecordMetadata().offset),
                throwable -> log.error("fail to publish", throwable)
                
        );
    }
}

 

위는 스프링부트 2버전에서 구현한 Producer Confirm이다.

 

스프링부트 2버전에서는 KafkaTemplate의 send()메서드가 ListenableFuture를 리턴해서 해당 코드가 되지만,

 

스프링부트 3에선 ListenableFuture가 deprecate 되었다.

 

CompletableFuture에는 callback메서드가 없기 때문에, whenComplete()메서드를 사용해야 한다.

 

- Spring boot 3 버전

@Slf4j
@Component
@RequiredArgsConstructor
public class Producer {

    public void sendEvent(CreateTaskEvent event) {
    
        CompletableFuture<SendResult<String, CreateTaskEvent>> future 
        = kafkaTemplate.send(TOPIC_TASK, event);
        
        future.whenComplete(((result, throwable) -> {
            if (throwable == null) {
                log.info("offset : {}", result.getRecordMetadata().offset());
            } else {
                log.error("fail to publish", throwable);
            }
        }
        ));
    }
}

 

이렇게 kafka 환경에서 성공, 실패에 대한 callback을 구현할 수 있다.

 

2-2. Consumer ACK

Consumer ACK은 AcknowledgingMessageListner 인터페이스를 이용한다.

@FunctionalInterface
public interface AcknowledgingMessageListener<K, V> extends MessageListener<K, V> {

	/**
	 * Invoked with data from kafka. Containers should never call this since it they
	 * will detect that we are an acknowledging listener.
	 * @param data the data to be processed.
	 */
	@Override
	default void onMessage(ConsumerRecord<K, V> data) {
		throw new UnsupportedOperationException("Container should never call this");
	}

	/**
	 * Invoked with data from kafka.
	 * @param data the data to be processed.
	 * @param acknowledgment the acknowledgment.
	 */
	@Override
	void onMessage(ConsumerRecord<K, V> data, @Nullable Acknowledgment acknowledgment);

}

 

해당 인터페이스에 있는 onMessage 메서드에서 두 번째 인자로 Acknowledgement를 전달할 수 있는데, 이를 이용해서 Consumer ACK를 구현한다.

 

 

아래는 Consumer ACK을 구현한 코드이다.

 

@Component
@Slf4j
public class MyAcknowledgementMessageListner implements AcknowledgingMessageListener<String, String> {

    @Override
    @KafkaListener(
            //...
            containerFactory = "KafkaListenerContainerFactory"
    )
    public void onMessage(ConsumerRecord<String, String> data, Acknowledgment acknowledgment) {
        try {
            acknowledgment.acknowledge();
        } catch (Exception e) {
            log.error("Error to receive messages. ", e);
        }
    }
}

 

AcknowledgingMessageListener를 상속받아 onMessage를 Override한다.

 

인자로 받은 Acknowledgement의 acknowledge()메서드를 실행하면 성공한 Consumer ACK이 전송된다.

 

 

이렇게 수동적으로 ACK을 보내려면 미리 해두어야 할 설정이 있다.

 

@EnableKafka
@Configuration
public class KafkaConfig {

	//...

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        //Setting for Consumer ACK
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        return new DefaultKafkaConsumerFactory<>(props);
    }
    
    //...
}

 

CosnumerFactory 설정에서, AUTO COMMIT을 false로 설정해서 직접 ACK, NACK를 날릴 수 있도록 설정해둔다.

 

 

 

 

 

 

---

 

이로써 RDB, RabbitMQ, Kafka 환경에서 아래 조건을 충족했다.

• At Least Once : 최소 한 번 전달 구현
• Producer Confirm, Consumer ACK

앞으로 MSA구조로 프로젝트를 계획하게 된다면, 배운 점을 이용하여 안정성있는 데이터 전달을 구현할 수 있겠다.

 

 

 

 

 

참고
https://github.com/spring-projects/spring-boot/wiki/Spring-Boot-3.0-Migration-Guide
https://spring.io/projects/spring-kafka
https://www.youtube.com/watch?v=uk5fRLUsBfk&t=1797s&ab_channel=NHNCloud