code: public KafkaClient init() throws Exception {
          if (executorService == null) {
              int numThreads = Math.min(Math.max(Runtime.getRuntime().availableProcessors(), 1), 4);
              executorService = Executors.newFixedThreadPool(numThreads);
              myOwnExecutorService = true;
          } else {
              myOwnExecutorService = false;
          }
  
          if (metadataConsumer == null) {
              metadataConsumer = KafkaHelper.createKafkaConsumer(getKafkaBootstrapServers(),
                      getMetadataConsumerGroupId(), false, true);
          }
  
          if (cacheProducers == null) {
              cacheProducers = new ConcurrentHashMap<>();
              for (ProducerType type : ProducerType.ALL_TYPES) {
                  cacheProducers.put(type, KafkaHelper.createKafkaProducer(type,
                          kafkaBootstrapServers, producerProperties));
              }
          }
  
          return this;
      }
docs: Init method.