docs: public static CuratorFramework startCuratorFramework(Configuration configuration) {
  		Preconditions.checkNotNull(configuration, "configuration");
  		String zkQuorum = configuration.getValue(HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM);
  
  		if (zkQuorum == null || StringUtils.isBlank(zkQuorum)) {
  			throw new RuntimeException("No valid ZooKeeper quorum has been specified. " +
  					"You can specify the quorum via the configuration key '" +
  					HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM.key() + "'.");
  		}
  
  		int sessionTimeout = configuration.getInteger(HighAvailabilityOptions.ZOOKEEPER_SESSION_TIMEOUT);
  
  		int connectionTimeout = configuration.getInteger(HighAvailabilityOptions.ZOOKEEPER_CONNECTION_TIMEOUT);
  
  		int retryWait = configuration.getInteger(HighAvailabilityOptions.ZOOKEEPER_RETRY_WAIT);
  
  		int maxRetryAttempts = configuration.getInteger(HighAvailabilityOptions.ZOOKEEPER_MAX_RETRY_ATTEMPTS);
  
  		String root = configuration.getValue(HighAvailabilityOptions.HA_ZOOKEEPER_ROOT);
  
  		String namespace = configuration.getValue(HighAvailabilityOptions.HA_CLUSTER_ID);
  
  		boolean disableSaslClient = configuration.getBoolean(SecurityOptions.ZOOKEEPER_SASL_DISABLE);
  
  		ACLProvider aclProvider;
  
  		ZkClientACLMode aclMode = ZkClientACLMode.fromConfig(configuration);
  
  		if (disableSaslClient && aclMode == ZkClientACLMode.CREATOR) {
  			String errorMessage = "Cannot set ACL role to " + aclMode + "  since SASL authentication is " +
  					"disabled through the " + SecurityOptions.ZOOKEEPER_SASL_DISABLE.key() + " property";
  			LOG.warn(errorMessage);
  			throw new IllegalConfigurationException(errorMessage);
  		}
  
  		if (aclMode == ZkClientACLMode.CREATOR) {
  			LOG.info("Enforcing creator for ZK connections");
  			aclProvider = new SecureAclProvider();
  		} else {
  			LOG.info("Enforcing default ACL for ZK connections");
  			aclProvider = new DefaultACLProvider();
  		}
  
  		String rootWithNamespace = generateZookeeperPath(root, namespace);
  
  		LOG.info("Using '{}' as Zookeeper namespace.", rootWithNamespace);
  
  		CuratorFramework cf = CuratorFrameworkFactory.builder()
  				.connectString(zkQuorum)
  				.sessionTimeoutMs(sessionTimeout)
  				.connectionTimeoutMs(connectionTimeout)
  				.retryPolicy(new ExponentialBackoffRetry(retryWait, maxRetryAttempts))
  				// Curator prepends a '/' manually and throws an Exception if the
  				// namespace starts with a '/'.
  				.namespace(rootWithNamespace.startsWith("/") ? rootWithNamespace.substring(1) : rootWithNamespace)
  				.aclProvider(aclProvider)
  				.build();
  
  		cf.start();
  
  		return cf;
  	}