Styjun

I have two Kafka clusters, the IPs for which I am fetching dynamically from database. I am using @KafkaListener for creating listeners. Now I want to create multiple Kafka listeners at runtime depending on the bootstrap server attribute(comma-separated values), each one listening to a cluster. Can you please suggest me how do I achieve this?

Spring-boot: 2.1.3.RELEASE
Kafka-2.0.1
Java-8

Your requirements are not clear but, assuming you want the same listener configuration to listen to multiple clusters, here is one solution. i.e. make the listener bean a prototype and mutate the container factory for each instance...

@SpringBootApplication
@EnableConfigurationProperties(ClusterProperties.class)
public class So55311070Application 

 public static void main(String[] args) 
 SpringApplication.run(So55311070Application.class, args);
 

 private final Map<String, MyListener> listeners = new HashMap<>();

 @Bean
 public ApplicationRunner runner(ClusterProperties props, ConsumerFactory<Object, Object> cf,
 ConcurrentKafkaListenerContainerFactory<Object, Object> containerFactory,
 ApplicationContext context, KafkaListenerEndpointRegistry registry) 

 return args -> 
 AtomicInteger instance = new AtomicInteger();
 Arrays.stream(props.getClusters()).forEach(cluster -> 
 Map<String, Object> consumerProps = new HashMap<>(cf.getConfigurationProperties());
 consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster);
 String groupId = "group" + instance.getAndIncrement();
 consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
 containerFactory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerProps));
 this.listeners.put(groupId, context.getBean("listener", MyListener.class));
 );
 registry.getListenerContainers().forEach(c -> System.out.println(c.getGroupId())); // 2.2.5 snapshot only
 ;
 

 @Bean
 @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
 public MyListener listener() 
 return new MyListener();
 



class MyListener 

 @KafkaListener(topics = "so55311070")
 public void listen(String in) 
 System.out.println(in);
 

@ConfigurationProperties(prefix = "kafka")
public class ClusterProperties 

 private String[] clusters;

 public String[] getClusters() 
 return this.clusters;
 

 public void setClusters(String[] clusters) 
 this.clusters = clusters;
 

kafka.clusters=localhost:9092,localhost:9093

spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=false

Result

group0
group1
...
2019-03-23 11:43:25.993 INFO 74869 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer 
 : partitions assigned: [so55311070-0]
2019-03-23 11:43:25.994 INFO 74869 --- [ntainer#1-0-C-1] o.s.k.l.KafkaMessageListenerContainer 
 : partitions assigned: [so55311070-0]

EDIT

Add code to retry starting failed containers.

It turns out we don't need a local map of listeners, the registry has a map of all containers, including the ones that failed to start.

@SpringBootApplication
@EnableConfigurationProperties(ClusterProperties.class)
public class So55311070Application 

 public static void main(String[] args) 
 SpringApplication.run(So55311070Application.class, args);
 

 private boolean atLeastOneFailure;

 private ScheduledFuture<?> restartTask;

 @Bean
 public ApplicationRunner runner(ClusterProperties props, ConsumerFactory<Object, Object> cf,
 ConcurrentKafkaListenerContainerFactory<Object, Object> containerFactory,
 ApplicationContext context, KafkaListenerEndpointRegistry registry, TaskScheduler scheduler) 

 return args -> 
 AtomicInteger instance = new AtomicInteger();
 Arrays.stream(props.getClusters()).forEach(cluster -> 
 Map<String, Object> consumerProps = new HashMap<>(cf.getConfigurationProperties());
 consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster);
 String groupId = "group" + instance.getAndIncrement();
 consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
 attemptStart(containerFactory, context, consumerProps, groupId);
 );
 registry.getListenerContainers().forEach(c -> System.out.println(c.getGroupId())); // 2.2.5 snapshot only
 if (this.atLeastOneFailure) 
 Runnable rescheduleTask = () -> 
 registry.getListenerContainers().forEach(c -> 
 this.atLeastOneFailure = false;
 if (!c.isRunning()) 
 System.out.println("Attempting restart of " + c.getGroupId());
 try 
 c.start();
 
 catch (Exception e) 
 System.out.println("Failed to start " + e.getMessage());
 this.atLeastOneFailure = true;
 
 
 );
 if (!this.atLeastOneFailure) 
 this.restartTask.cancel(false);
 
 ;
 this.restartTask = scheduler.scheduleAtFixedRate(rescheduleTask,
 Instant.now().plusSeconds(60),
 Duration.ofSeconds(60));
 
 ;
 

 private void attemptStart(ConcurrentKafkaListenerContainerFactory<Object, Object> containerFactory,
 ApplicationContext context, Map<String, Object> consumerProps, String groupId) 

 containerFactory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerProps));
 try 
 context.getBean("listener", MyListener.class);
 
 catch (BeanCreationException e) 
 this.atLeastOneFailure = true;
 
 

 @Bean
 @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
 public MyListener listener() 
 return new MyListener();
 

 @Bean
 public TaskScheduler scheduler() 
 return new ThreadPoolTaskScheduler();
 



class MyListener 

 @KafkaListener(topics = "so55311070")
 public void listen(String in) 
 System.out.println(in);