code: @Override
  	public void connect() {
  		try {
  
  			// Create MQTT client config
  			final MqttClientConfig config = new MqttClientConfig();
  			String uri = url;
  			if (uri.indexOf(':') == -1) {
  				uri = uri + ":1883";
  			}
  			uri = uri.replace("mqtt://", "tcp://");
  			if (url.indexOf("://") == -1) {
  				uri = "tcp://" + uri;
  			}
  
  			// Set properties
  			config.setReconnectionStrategy(new NullReconnectStrategy());
  			config.setKeepAliveSeconds(keepAliveSeconds);
  			config.setConnectTimeoutSeconds(connectTimeoutSeconds);
  			config.setMessageResendIntervalSeconds(messageResendIntervalSeconds);
  			config.setBlockingTimeoutSeconds(blockingTimeoutSeconds);
  			config.setMaxInFlightMessages(maxInFlightMessages);
  
  			// Create MQTT client
  			disconnect();
  			client = new AsyncMqttClient(uri, this, executor, config);
  			client.connect(nodeID + '-' + broker.getConfig().getUidGenerator().nextUID(), cleanSession, username,
  					password);
  
  		} catch (Exception cause) {
  			reconnect(cause);
  		}
  	}
docs: --- CONNECT ---