RetryTopicConfiguration not working for multiple cluster

[hariraogotit]

Team,

1. I have two of this bean - RetryTopicConfiguration. One configured for London cluster and other for Sydney cluster.
2. When I run the application, they are creating retry topics in both the clusters i.e london and sydney retry topics are created both in London and in Sydney
3. Expectation is london retry topics to be created in London and sydney retry topics to be created in Sydney cluster.

Any lead is much appreciated guys please.

Please find my listener

@KafkaListener(
         topics = {
                 "AP-sydney"
         },
         id = "sydney",
         containerFactory = "sydneyCluster")
 public void consume(ConsumerRecord<String, String> record) {
------------
@KafkaListener(
         topics = {
                 "EU-london"
         },
         id = "london",
         containerFactory = "londonCluster")
 public void consumeLondon(ConsumerRecord<String, String> record) {

Configuration details are below

@Bean("sydneyCluster")
 ConcurrentKafkaListenerContainerFactory<Integer, String> sydneyListenerContainer() {
   ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
           new ConcurrentKafkaListenerContainerFactory<>();
   factory.setConsumerFactory(consumerSydneyFactory());

   return factory;
 }

 @Bean // Beans for the First Consumer Factory
 public ConsumerFactory<Integer, String> consumerSydneyFactory() {
   return new DefaultKafkaConsumerFactory<>(consumerSydneyConfigs());
 }

 @Bean
 public Map<String, Object> consumerSydneyConfigs() {
   Map<String, Object> props = new HashMap<>();
   props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, sydneyServers);
   props.put(ConsumerConfig.GROUP_ID_CONFIG, "hr");
   props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
   props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
   return props;
 }


 @Bean("londonCluster")
 ConcurrentKafkaListenerContainerFactory<Integer, String> londonListenerContainer() {
   ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
           new ConcurrentKafkaListenerContainerFactory<>();
   factory.setConsumerFactory(consumerLondonFactory());

   return factory;
 }


 @Bean // Beans for the First Consumer Factory
 public ConsumerFactory<Integer, String> consumerLondonFactory() {
   return new DefaultKafkaConsumerFactory<>(consumerLondonConfigs());
 }



 @Bean
 public Map<String, Object> consumerLondonConfigs() {
   Map<String, Object> props = new HashMap<>();
   props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, londonServers);
   props.put(ConsumerConfig.GROUP_ID_CONFIG, "hr");
   props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
   props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
   return props;
 }

 @Bean(name = "sydneyKafkaTemplate")
 public KafkaTemplate<String, String> sydneyKafkaTemplate() {
   Map<String, Object> props = new HashMap<>();
   props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, sydneyServers);
   props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
   props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

   ProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(props);
   return new KafkaTemplate<>(producerFactory);
 }

 @Bean(name = "londonKafkaTemplate")
 public KafkaTemplate<String, String> londonKafkaTemplate() {
   Map<String, Object> props = new HashMap<>();
   props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, londonServers);
   props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
   props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

   ProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(props);
   return new KafkaTemplate<>(producerFactory);
 }

 @Bean
 public RetryTopicConfiguration retrySydneyKafkaConfig(
         @Qualifier("sydneyKafkaTemplate") KafkaTemplate<String, String> sydneyKafkaTemplate) {
   return RetryTopicConfigurationBuilder
           .newInstance()
           .includeTopics(List.of("AP-sydney"))
           .exponentialBackoff(1000, 2, 20000)
           .maxAttempts(4)
           .retryTopicSuffix("-AP-Australia-retry")
           .dltSuffix("-AP-Australia-dlt")
           .useSingleTopicForSameIntervals()
           .listenerFactory("sydneyCluster")
           .create(sydneyKafkaTemplate);
 }

 @Bean
 public RetryTopicConfiguration retryLondonKafkaConfig(
         @Qualifier("londonKafkaTemplate") KafkaTemplate<String, String> londonKafkaTemplate) {
   return   RetryTopicConfigurationBuilder
           .newInstance()
           .includeTopics(List.of("EU-london"))
           .exponentialBackoff(1000, 2, 20000)
           .maxAttempts(4)
           .retryTopicSuffix("-EU-England-retry")
           .dltSuffix("-EU-England-dlt")
           .useSingleTopicForSameIntervals()
           .listenerFactory("londonCluster")
           .create(londonKafkaTemplate);
 }

[garyrussell]

The topic creation logic simply adds NewTopic beans to the application context; these beans are picked up by the KafkaAdmin bean (presumably the one configured by Boot since you don't have any).

I would expect both topics to be created in just one cluster, not both; logs should help you understand what is happening.

You really need two KafkaAdmin beans, one for each cluster, with some filtering logic to only declare a subset of the topics in each cluster.

Unfortunately, the KafkaAdmin is not really designed for subclassing like that. We could make it easier to do that, but it will require changes.

In the meantime, you should set spring.kafka.admin.auto-create to false (or simply disable auto creation by adding .doNotAutoCreateRetryTopics() to your config builders) and create the topics yourself, perhaps using an AdminClient, or manually using the command line tools.

[garyrussell]

#2720

[hariraogotit]

Thank you @garyrussell. In short it works now.

I created two KafkaAdmin clients and used .doNotAutoCreateRetryTopics()

Creating the KafkaAdmin clients per cluster was challenging but got there at the end. Basically I did the below.

@Bean
 public KafkaAdmin kafkaAdminSydney() {
   return new KafkaAdmin(getCommonProperties(sydneyServers));
 }

 @Bean
 public KafkaAdmin kafkaAdminLondon() {
   return new KafkaAdmin(getCommonProperties(londonServers));
 }

   @Autowired
   KafkaAdmin kafkaAdminSydney;
   @Autowired
   KafkaAdmin kafkaAdminLondon;

  @PostConstruct
  public void createTopics() {
      kafkaAdminSydney.createOrModifyTopics(topic1(), topic2(), topic3(), topic4());
      kafkaAdminLondon.createOrModifyTopics(topic5(), topic6(), topic7(), topic8());
  }

[garyrussell]

This #2721 should make your configuration a bit easier - it will be in the 2.9.10 and 3.0.9 releases.

[hariraogotit]

Thank you @garyrussell

[hariraogotit]

This #2721 should make your configuration a bit easier - it will be in the 2.9.10 and 3.0.9 releases.