Spring Boot에서 여러 Kafka 클러스터 사용하기

개요

프로젝트에서 일반적으로 하나의 Kafka 클러스터만을 사용하는 경우가 대부분이지만, 경우에 따라 서로 다른 환경의 Kafka 클러스터에 동시에 연결해야 할 필요가 생길 수 있다.

나는 이번에 작업하면서 기존의 연결된 Kafka 클러스터 말고도 또 하나의 새로운 클러스터를 연결해야 했는데, 작업을 진행하면서 얻었던 지식을 공유하고자 작성하게 되었다.

 

 

컨슈머 설정

Spring Boot는 기본적으로 ConcurrentKafkaListenerContainerFactory를 자동으로 구성하여 Kafka 리스너 컨테이너를 생성한다. 그러나 여러 Kafka 클러스터에 연결하려면 각 클러스터에 대해 별도의 ConcurrentKafkaListenerContainerFactory 빈을 생성하고, 해당 클러스터에 맞는 ConsumerFactory 빈을 설정해야 한다.

아래는 두 개의 kafka 클러스터(Kafka-A와 Kafka-B)를 설정하는 예시 코드이다.

kafka-A:
  bootstrap-server: b-2.kafka-a.amazonaws.com:9092,b-1.kafka-a.amazonaws.com:9092...

kafka-B:
  bootstrap-server: b-1.kafka-b.amazonaws.com:9092,b-2.kafka-b.amazonaws.com:9092...

@Configuration
public class KafkaConsumerConfig {

    @Value("${kafka-A.bootstrap-server}")
    private String kafkaABootstrapServers;

    @Value("${kafka-B.bootstrap-server}")
    private String kafkaBBootstrapServers;

    @Bean
    public ConsumerFactory<String, String> kafkaAConsumerFactory() {
        Map<String, Object> properties = new HashMap<>();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaABootstrapServers);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "application.name");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        
        return new DefaultKafkaConsumerFactory<>(properties);
    }

    @Bean
    public ConsumerFactory<String, String> kafkaBConsumerFactory() {
        Map<String, Object> properties = new HashMap<>();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBBootstrapServers);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "application.name");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        
        return new DefaultKafkaConsumerFactory<>(properties);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaAListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(kafkaAConsumerFactory());
        
        return factory;
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaBListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(kafkaBConsumerFactory());
        
        return factory;
    }
}

이제 @KafkaListener 애노테이션을 사용해 containerFactory 속성에 특정 클러스터의 빈을 명시함으로써, 해당 클러스터에 연결할 수 있다.

@Component
public class KafkaConsumerService {

    @KafkaListener(topics = "topicA", containerFactory = "kafkaAListenerContainerFactory")
    public void consumeFromKafkaA(String message) {
        System.out.println("Received from Kafka-A: " + message);
    }

    @KafkaListener(topics = "topicB", containerFactory = "kafkaBListenerContainerFactory")
    public void consumeFromKafkaB(String message) {
        System.out.println("Received from Kafka-B: " + message);
    }
}

빈만 추가해 주면 되어서 생각보다 되게 간단하다. 다만 기존에는 @KafkaListener에 containerFactory 속성을 명시하지 않고 사용했었는데, 이 경우 기본적으로 어떤 빈이 사용되길래 작동되는지 궁금했다.

 

containerFactory 속성 미지정 시 동작 방식

Spring Boot는 기본적으로 kafkaListenerContainerFactory라는 이름의 ConcurrentKafkaListenerContainerFactory 빈을 사용하여 Kafka 리스너 컨테이너를 생성한다. 따라서 @KafkaListener 애노테이션에서 containerFactory를 명시하지 않으면 아래와 같이 동작한다.

• 기본 빈 사용: kafkaListenerContainerFactory라는 이름의 빈이 정의되어 있다면, @KafkaListener에서 containerFactory를 명시하지 않아도 이 빈이 사용된다.
• 명시적 빈 사용: 만약 newKafkaListenerContainerFactory라는 이름의 빈을 별도로 정의하고, @KafkaListener 애노테이션에서 containerFactory = "newKafkaListenerContainerFactory"로 명시했다면 해당 빈이 사용된다.
• 빈 미정의 시 예외 발생: kafkaAListenerContainerFactory와 kafkaBListenerContainerFactory 같은 이름의 빈을 등록한 상태에서 @KafkaListener에서 containerFactory를 명시하지 않으면, 기본적으로 사용될 kafkaListenerContainerFactory 빈을 찾을 수 없다는 NoSuchBeanDefinitionException이 발생한다.

공식문서에서 기본 빈 네임이 kafkaListenerContainerFactory임을 확인할 수 있다.

https://docs.spring.io/spring-kafka/reference/kafka/receiving-messages/listener-annotation.html

 

 

프로듀서 설정

프로듀서 설정도 컨슈머 설정과 마찬가지로 각 클러스터에 대해 별도의 ProducerFactory 빈을 생성하고, 해당 클러스터에 맞는 KafkaTemplate을 설정해야 한다.

@Configuration
public class KafkaProducerConfig {

    @Value("${kafka-A.bootstrap-server}")
    private String kafkaABootstrapServers;

    @Value("${kafka-B.bootstrap-server}")
    private String kafkaBBootstrapServers;

    @Bean
    public ProducerFactory<String, String> kafkaAProducerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaABootstrapServers);
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public ProducerFactory<String, String> kafkaBProducerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBBootstrapServers);
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaATemplate() {
        return new KafkaTemplate<>(kafkaAProducerFactory());
    }

    @Bean
    public KafkaTemplate<String, String> kafkaBTemplate() {
        return new KafkaTemplate<>(kafkaBProducerFactory());
    }
}

사용을 할 때는 다음과 같이 사용하면 된다(간단하게 설명하기 위해 필드 주입을 시켰다).

@Component
public class KafkaProducerService {

    @Autowired
    private KafkaTemplate<String, String> kafkaATemplate;

    @Autowired
    private KafkaTemplate<String, String> kafkaBTemplate;

    public void sendMessageToKafkaA(String topic, String message) {
        kafkaATemplate.send(topic, message);
        System.out.println("Sent to Kafka-A: " + message);
    }

    public void sendMessageToKafkaB(String topic, String message) {
        kafkaBTemplate.send(topic, message);
        System.out.println("Sent to Kafka-B: " + message);
    }
}

이 설정을 통해 kafka-A와 kafka-B 클러스터에 각각 메시지를 전송할 수 있으며, 상황에 따라 적절한 KafkaTemplate을 선택하여 사용할 수 있다.

만약 KafkaProducerService에서 설정되지 않은 kafkaCTemplate을 주입받으려 한다면, KafkaProducerConfig에 해당 빈이 정의되어 있지 않기 때문에 NoSuchBeanDefinitionException이 발생하게 된다. 만약 특정 빈이 정의되어 있지 않은 경우, 기본으로 사용할 KafkaTemplate을 지정하고 싶다면 아래와 같이 @Primary 애노테이션을 사용하여 지정할 수 있다.

@Bean
@Primary
public KafkaTemplate<String, String> defaultKafkaTemplate() {
    return new KafkaTemplate<>(kafkaAProducerFactory());
}

이 경우 kafkaTemplate이 기본적으로 사용되며, 다른 KafkaTemplate이 명시적으로 주입되지 않은 경우 기본 빈으로 선택된다. 즉 다음과 같이 일치하지 않는 빈 네임의 KafkaTemplate을 주입받더라도 @Primary로 지정된 defaultKafkaTemplate이 주입된다.

@Service
public class KafkaProducerService {

    /**
     * 이 필드의 이름이 실제로 존재하는 빈 이름과 일치하지 않더라도,
     * @Primary로 지정된 KafkaTemplate이 주입된다.
     */
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
        System.out.println("Sent to Kafka-A: " + message);
    }
}

다만 이러한 경우에는 잘못된 KafkaTemplate이 주입될 가능성이 있으므로, 실수를 방지하기 위해 마커 인터페이스를 구현한 클래스를 사용하는 등 휴먼 에러를 줄일 수 있는 방법을 고려해 보는 것도 좋을 것 같다.