Non blocking retry spring kafka not picking the correct Producer kafkatemplate

[hariraogotit]

Team,

After getting more info from you - #2637

I tried out setting two kafka instances one named Sydney and other London

I see that when I send a message to Sydney, it goes to Sydney
Similarly when I send a message to London, it goes to London

BUT when I send a message to London and when the Spring non blocking retry kicks in the Sydney producer is used and not the London's

Kindly can you help me to figure out whats wrong in this code?
cluster.zip

This has 3 files

1. KafkaConfiguration containing all the configurations for kafka producer, consumer and retry
2. MainListener - simple listener
3. application.yml - containing the kafka broker servers

logs and input

Topic name from listener Sydney -> greetings
Message from the listener Sydney --> HI this message goes to Sydney
Group name Sydney  --> hr

Topic name from listener Sydney -> greetings
Message from the listener Sydney --> hari sydney
Group name Sydney  --> hr
2023-05-25T16:08:49.683+01:00  INFO 46768 --- [ad | producer-1] org.apache.kafka.clients.Metadata        : [Producer clientId=producer-1] Resetting the last seen epoch of partition greetings-sydney-retry-1000-0 to 0 since the associated topicId changed from null to zZlRfRitT3iig5kVQtKjgw
2023-05-25T16:08:49.706+01:00  INFO 46768 --- [etry-1000-0-C-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=consumer-hr-2, groupId=hr] Seeking to offset 12 for partition greetings-sydney-retry-1000-0
Topic name from listener Sydney -> greetings-sydney-retry-1000
Message from the listener Sydney --> hari sydney
Group name Sydney  --> hr
2023-05-25T16:08:50.731+01:00  INFO 46768 --- [ad | producer-1] org.apache.kafka.clients.Metadata        : [Producer clientId=producer-1] Resetting the last seen epoch of partition greetings-sydney-retry-2000-0 to 0 since the associated topicId changed from null to k2pILfLlTUSaHYTuCjOLpQ
2023-05-25T16:08:50.750+01:00  INFO 46768 --- [etry-2000-0-C-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=consumer-hr-1, groupId=hr] Seeking to offset 8 for partition greetings-sydney-retry-2000-0
Topic name from listener Sydney -> greetings-sydney-retry-2000
Message from the listener Sydney --> hari sydney
Group name Sydney  --> hr
2023-05-25T16:08:52.233+01:00 ERROR 46768 --- [etry-2000-0-C-1] k.r.DeadLetterPublishingRecovererFactory : Record: topic = greetings-sydney-retry-2000, partition = 0, offset = 8, main topic = greetings threw an error at topic greetings-sydney-retry-2000 and won't be retried. Sending to DLT with name greetings-sydney-dlt.

Topic name from listener London -> greetings
Message from the listener London --> This goes to London
Group name London  --> hr

Topic name from listener London -> greetings
Message from the listener London --> hari london
Group name London  --> hr
2023-05-25T16:12:08.933+01:00  INFO 46768 --- [etry-1000-0-C-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=consumer-hr-2, groupId=hr] Seeking to offset 13 for partition greetings-sydney-retry-1000-0
Topic name from listener Sydney -> greetings-sydney-retry-1000
Message from the listener Sydney --> hari london
Group name Sydney  --> hr

See in the above when I send a poison pill - hari London it goes to London but when the non blocking retry kicks in, it sends to Sydney and not London.

[hariraogotit]

Team

I think this is a BUG i.e Spring Kafka Non Blocking retries does not support multiple KafkaTemplates.

I tried even the following variation and it is not recognising the 2nd RetryTopicConfiguration.

Kindly let me know your thoughts.

@Bean
 ConcurrentKafkaListenerContainerFactory<Integer, String> sydneyRetryContainer() {
   return (ConcurrentKafkaListenerContainerFactory<Integer, String>) sydneyCluster();
 }


 @Bean
 ConcurrentKafkaListenerContainerFactory<Integer, String> londonRetryContainer() {
   return (ConcurrentKafkaListenerContainerFactory<Integer, String>) londonCluster();
 }

@Bean
 public RetryTopicConfiguration shaypeRetrySydneyKafkaConfig(
         ConcurrentKafkaListenerContainerFactory<Integer, String> sydneyRetryContainer) {
   RetryTopicConfiguration b =  RetryTopicConfigurationBuilder
           .newInstance()
           .notRetryOn(Collections.emptyList())
           .exponentialBackoff(1000, 2, 20000)
           .maxAttempts(3)
           .retryTopicSuffix("-sydney-retry")
           .dltSuffix("-sydney-dlt")
           .useSingleTopicForSameIntervals()
           .listenerFactory(sydneyRetryContainer)
           .create(sydneyKafkaTemplate());
   return b;
 }

 @Bean
 public RetryTopicConfiguration shaypeRetryLondonKafkaConfig(
         ConcurrentKafkaListenerContainerFactory<Integer, String> londonRetryContainer) {
   RetryTopicConfiguration b =  RetryTopicConfigurationBuilder
           .newInstance()
           .notRetryOn(Collections.emptyList())
           .exponentialBackoff(500, 2, 20000)
           .maxAttempts(4)
           .retryTopicSuffix("-london-retry")
           .dltSuffix("-london-dlt")
           .useSingleTopicForSameIntervals()
           .listenerFactory(londonRetryContainer)
           .create(londonKafkaTemplate());

   return b;
 }

 @Bean(name = "sydneyKafkaTemplate")
 public KafkaTemplate<String, byte[]> 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, byte[]> producerFactory = new DefaultKafkaProducerFactory<>(props);
   return new KafkaTemplate<>(producerFactory);
 }

 @Bean(name = "londonKafkaTemplate")
 public KafkaTemplate<String, byte[]> 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, byte[]> producerFactory = new DefaultKafkaProducerFactory<>(props);
   return new KafkaTemplate<>(producerFactory);
 }

[garyrussell]

@hariraogotit Is this still a problem? Or, have you solved it?

[hariraogotit]

@garyrussell thanks for coming back to me.

Yes once I removed the below code, it worked

  @Bean
  @Primary
  KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
  kafkaListenerContainerFactory() {
    return sydneyCluster();
  }

I anyway do not want the above config.