flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java
/**
	 * Starts a {@link CuratorFramework} instance and connects it to the given ZooKeeper
	 * quorum.
	 *
	 * @param configuration {@link Configuration} object containing the configuration values
	 * @return {@link CuratorFramework} instance
	 */
Starts a  CuratorFramework instance and connects it to the given ZooKeeper
quorum.

@param configuration  Configuration object containing the configuration values
@return  CuratorFramework instance
Starts a CuratorFramework instance and connects it to the given ZooKeeper quorum . @param configuration Configuration object containing the configuration values @return CuratorFramework instance
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))
				.namespace(rootWithNamespace.startsWith("/") ? rootWithNamespace.substring(1) : rootWithNamespace)
				.aclProvider(aclProvider)
				.build();
		cf.start();
		return cf;
	}
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 ) ) . namespace ( rootWithNamespace . startsWith ( "/" ) ? rootWithNamespace . substring ( 1 ) : rootWithNamespace ) . aclProvider ( aclProvider ) . build ( ) ; cf . start ( ) ; return cf ; }
Starts a {@link CuratorFramework} instance and connects it to the given ZooKeeper
 quorum.
Starts a { @link CuratorFramework } instance and connects it to the given ZooKeeper quorum .
// Curator prepends a '/' manually and throws an Exception if the // namespace starts with a '/'.