Getting KafkaConsumer is not safe for multi-threaded access error when I use CuratorFrameworkFactory.newClient()

Kafka Javdocs clearly mentions that :

The Kafka consumer is NOT thread-safe. All network I/O happens in the thread of the application making the call. It is the responsibility of the user to ensure that multi-threaded access is properly synchronized. Un-synchronized access will result in ConcurrentModificationException.

More details present at http://kafka.apache.org/21/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#multithreaded

here are some example to handle several topics thread-safe.

1. One topic Consumer Per Thread, avoiding multi-threaded access error

   class Consumer implements Runnable {
       private volatile boolean exit = false;
   
       private String topic;
   
       Consumer(String topic){
           this.topic=topic;
       }
   
       @Override
       public void run() {
           KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(props);
   
           try{
               consumer.subscribe(Collections.singletonList(topic));
   
               while (!exit) {
                   ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
   
                   for (ConsumerRecord<String, String> record : records){
                       System.out.print("Topic: " + record.topic() + ", ");
                       System.out.print("Partition: " + record.partition() + ",");
                       System.out.print("Key:" + record.key() + ", ");
                       System.out.println("Value: " + record.value() + ", ");
   
                       //do your stuff here
   
                   }
               }
           } catch (Exception e){
               e.printStackTrace();
           }finally {
               consumer.close();
           }
       }
       public void stop() {
          exit = true;
       }
    }
   

2. One consumer with dynamic topic add, and multi-threaded access synchronized.

   class Consumer implements Runnable {
       private volatile boolean exit = false;
   
       private final List<String> topics;
       private final KafkaConsumer<String, String> consumer;
   
       Consumer(){
           topics = new ArrayList<>();
           consumer = new KafkaConsumer<String,String>(props);
       }
   
       public void addTopic(String newTopic){
            synchronized(this) {
               topics.add(newTopic);
               consumer.subscribe(topics);
           }
       }
   
       @Override
       public void run() {
           try{
               while (!exit) {
                   synchronized(this) {
                       ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                       for (ConsumerRecord<String, String> record : records){
                           System.out.print("Topic: " + record.topic() + ", ");
                           System.out.print("Partition: " + record.partition() + ",");
                           System.out.print("Key:" + record.key() + ", ");
                           System.out.println("Value: " + record.value() + ", ");
   
                           //do your stuff here
                       }
                   }
   
                   //IMPORTANT: sleep to avoid lock addTopic method
                   try { Thread.sleep(200); } catch (Exception ex) { }
               }
           } catch (Exception e){
               e.printStackTrace();
           }finally {
               consumer.close();
           }
       }
       public void stop() {
          exit = true;
       }
    }
   

I hope it helps