code: @Override
  	public void connect() {
  		try {
  
  			// Create Kafka clients
  			disconnect();
  			StringBuilder urlList = new StringBuilder(128);
  			for (String url : urls) {
  				if (urlList.length() > 0) {
  					urlList.append(',');
  				}
  
  				// Add default port
  				if (url.indexOf(':') == -1) {
  					url = url + ":9092";
  				}
  
  				// Remove protocol prefix ("tcp://127.0.0.0")
  				int i = url.indexOf("://");
  				if (i > -1 && i < url.length() - 4) {
  					url = url.substring(i + 3);
  				}
  				urlList.append(url);
  			}
  			producerProperties.put("bootstrap.servers", urlList.toString());
  			consumerProperties.put("bootstrap.servers", urlList.toString());
  
  			// Set unique "node ID" as Kafka "group ID"
  			consumerProperties.setProperty("group.id", nodeID);
  
  			// Create producer
  			ByteArraySerializer byteArraySerializer = new ByteArraySerializer();
  			producer = new KafkaProducer<>(producerProperties, byteArraySerializer, byteArraySerializer);
  
  			// Start reader loop
  			poller = new KafkaPoller(this);
  			executor = Executors.newSingleThreadExecutor();
  			executor.execute(poller);
  
  			// Start subscribing channels...
  			connected();
  
  		} catch (Exception cause) {
  			reconnect(cause);
  		}
  	}
docs: --- CONNECT ---