CODEWITHSUNDEEP
javaspringspring-bootapache-kafka
Suraj Menon
Suraj Menon

Reputation: 1594

Connect to multiple Kafka Cluster's using spring Kafka

I have a spring boot application that consumes messages from a topic(say topic1) in a Kafka cluster. This is how my code looks like currently. 

@Configuration
public class KafkaTopicConfig {

    @Value(value = "${kafka.bootstrapAddress}")
    private String bootstrapAddress;

    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        return new KafkaAdmin(configs);
    }

    @Bean
    public NewTopic topic1() {
         return new NewTopic("baeldung", 1, (short) 1);
    }
}

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

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

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> 
      kafkaListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, String> factory =
          new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

@KafkaListener(topics = "topicName", groupId = "foo")
public void listen(String message) {
    System.out.println("Received Messasge in group foo: " + message);
}

Now I want to start consuming from a different topic in another Kafka cluster. One way is to create another bean for this. But is there a better way to do this?

Upvotes: 2

Views: 9605

Answers (1)

Ren&#233; Link
Ren&#233; Link

Reputation: 51323

You need another ConsumerFactory and KafkaListenerContainerFactory that connects to the other cluster's bootstrap servers.

Then you can use the containerFactory in the @KafkaListener annotation.

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

...

@KafkaListener(...., containerFactory="otherClusterFactory")

Upvotes: 3

Related Questions