/**
 * Sets the acl on path
 *
 * @param path
 * @param acl
 *            List of ACL permissions to assign to the path.
 * @throws ZkException
 *             if any ZooKeeper exception occurred
 * @throws RuntimeException
 *             if any other exception occurs
 */
public void setAcl(final String path, final List<ACL> acl) throws ZkException {
    if (path == null) {
        throw new NullPointerException("Missing value for path");
    }

    if (acl == null || acl.size() == 0) {
        throw new NullPointerException("Missing value for ACL");
    }

    if (!exists(path)) {
        throw new RuntimeException("trying to set acls on non existing node " + path);
    }

    retryUntilConnected(new Callable<Void>() {
        @Override
        public Void call() throws Exception {
            Stat stat = new Stat();
            _connection.readData(path, stat, false);
            _connection.setAcl(path, acl, stat.getAversion());
            return null;
        }
    });
}
 

/**
 * Gets the acl on path
 *
 * @param path
 * @return an entry instance with key = list of acls on node and value = stats.
 * @throws ZkException
 *             if any ZooKeeper exception occurred
 * @throws RuntimeException
 *             if any other exception occurs
 */
public Map.Entry<List<ACL>, Stat> getAcl(final String path) throws ZkException {
    if (path == null) {
        throw new NullPointerException("Missing value for path");
    }

    if (!exists(path)) {
        throw new RuntimeException("trying to get acls on non existing node " + path);
    }

    return retryUntilConnected(new Callable<Map.Entry<List<ACL>, Stat>>() {
        @Override
        public Map.Entry<List<ACL>, Stat> call() throws Exception {
            return _connection.getAcl(path);
        }
    });
}
 

/**
 * Create a persistent Sequential node.
 *
 * @param path
 * @param data
 * @param createParents if true all parent dirs are created as well and no
 * {@link ZkNodeExistsException} is thrown in case the path already exists
 * @throws ZkInterruptedException if operation was interrupted, or a
 * required reconnection got interrupted
 * @throws IllegalArgumentException if called from anything except the
 * ZooKeeper event thread
 * @throws ZkException if any ZooKeeper exception occurred
 * @throws RuntimeException if any other exception occurs
 */
public String createPersistentSequential(String path, Object data, boolean createParents)
                                                                                         throws ZkInterruptedException,
                                                                                         IllegalArgumentException,
                                                                                         ZkException,
                                                                                         RuntimeException {
    try {
        return create(path, data, CreateMode.PERSISTENT_SEQUENTIAL);
    } catch (ZkNoNodeException e) {
        if (!createParents) {
            throw e;
        }
        String parentDir = path.substring(0, path.lastIndexOf('/'));
        createPersistent(parentDir, createParents);
        return createPersistentSequential(path, data, createParents);
    }
}
 

/**
 * Create a persistent Sequential node.
 *
 * @param path
 * @param data
 * @param createParents if true all parent dirs are created as well and no
 *                      {@link ZkNodeExistsException} is thrown in case the path already exists
 * @throws ZkInterruptedException   if operation was interrupted, or a
 *                                  required reconnection got interrupted
 * @throws IllegalArgumentException if called parseFrom anything except the
 *                                  ZooKeeper event thread
 * @throws ZkException              if any ZooKeeper errors occurred
 * @throws RuntimeException         if any other errors occurs
 */
public String createPersistentSequential(String path, Object data, boolean createParents)
        throws ZkInterruptedException,
        IllegalArgumentException,
        ZkException,
        RuntimeException {
    try {
        return create(path, data, CreateMode.PERSISTENT_SEQUENTIAL);
    } catch (ZkNoNodeException e) {
        if (!createParents) {
            throw e;
        }
        String parentDir = path.substring(0, path.lastIndexOf('/'));
        createPersistent(parentDir, createParents);
        return createPersistentSequential(path, data, createParents);
    }
}
 

/**
 * Sets the acl on path
 *
 * @param path
 * @param acl
 *            List of ACL permissions to assign to the path.
 * @throws ZkException
 *             if any ZooKeeper exception occurred
 * @throws RuntimeException
 *             if any other exception occurs
 */
public void setAcl(final String path, final List<ACL> acl) throws ZkException {
    if (path == null) {
        throw new NullPointerException("Missing value for path");
    }

    if (acl == null || acl.size() == 0) {
        throw new NullPointerException("Missing value for ACL");
    }

    if (!exists(path)) {
        throw new RuntimeException("trying to set acls on non existing node " + path);
    }

    retryUntilConnected(new Callable<Void>() {
        @Override
        public Void call() throws Exception {
            Stat stat = new Stat();
            _connection.readData(path, stat, false);
            _connection.setAcl(path, acl, stat.getAversion());
            return null;
        }
    });
}
 

/**
 * Gets the acl on path
 *
 * @param path
 * @return an entry instance with key = list of acls on node and value = stats.
 * @throws ZkException
 *             if any ZooKeeper exception occurred
 * @throws RuntimeException
 *             if any other exception occurs
 */
public Map.Entry<List<ACL>, Stat> getAcl(final String path) throws ZkException {
    if (path == null) {
        throw new NullPointerException("Missing value for path");
    }

    if (!exists(path)) {
        throw new RuntimeException("trying to get acls on non existing node " + path);
    }

    return retryUntilConnected(new Callable<Map.Entry<List<ACL>, Stat>>() {
        @Override
        public Map.Entry<List<ACL>, Stat> call() throws Exception {
            return _connection.getAcl(path);
        }
    });
}
 

/**
 * Returns an {@link Authorizer} to make {@link Acl} requests
 *
 * @return an {@link Authorizer} to make {@link Acl} requests
 *
 * @throws AdminOperationException
 *      if there is an issue creating the authorizer
 */
public Authorizer getAuthorizer() {
    if (authorizer == null) {
        ZKConfig zkConfig = new ZKConfig(new VerifiableProperties(properties));

        Map<String, Object> authorizerProps = new HashMap<>();
        authorizerProps.put(ZKConfig.ZkConnectProp(), zkConfig.zkConnect());
        authorizerProps.put(ZKConfig.ZkConnectionTimeoutMsProp(), zkConfig.zkConnectionTimeoutMs());
        authorizerProps.put(ZKConfig.ZkSessionTimeoutMsProp(), zkConfig.zkSessionTimeoutMs());
        authorizerProps.put(ZKConfig.ZkSyncTimeMsProp(), zkConfig.zkSyncTimeMs());

        try {
            Authorizer simpleAclAuthorizer = new SimpleAclAuthorizer();
            simpleAclAuthorizer.configure(authorizerProps);
            authorizer = simpleAclAuthorizer;
        } catch (ZkException | ZooKeeperClientException e) {
            throw new AdminOperationException("Unable to create authorizer", e);
        }
    }

    return authorizer;
}
 

/**
 * Create a persistent Sequential node.
 *
 * @param path
 * @param data
 * @param createParents if true all parent dirs are created as well and no
 * {@link ZkNodeExistsException} is thrown in case the path already exists
 * @throws ZkInterruptedException if operation was interrupted, or a
 * required reconnection got interrupted
 * @throws IllegalArgumentException if called from anything except the
 * ZooKeeper event thread
 * @throws ZkException if any ZooKeeper exception occurred
 * @throws RuntimeException if any other exception occurs
 */
public String createPersistentSequential(String path, Object data, boolean createParents)
                                                                                         throws ZkInterruptedException,
                                                                                         IllegalArgumentException,
                                                                                         ZkException,
                                                                                         RuntimeException {
    try {
        return create(path, data, CreateMode.PERSISTENT_SEQUENTIAL);
    } catch (ZkNoNodeException e) {
        if (!createParents) {
            throw e;
        }
        String parentDir = path.substring(0, path.lastIndexOf('/'));
        createPersistent(parentDir, createParents);
        return createPersistentSequential(path, data, createParents);
    }
}
 

private void syncFromZooKeeper() throws ZkException, IOException {
    ConcurrentHashMap<String/*zk path*/, Set<IZkDataListener>> listeners = zkClient.getDataListener();
    Enumeration<String> paths = listeners.keys();
    while (paths.hasMoreElements()) {
        String path = paths.nextElement();
        initPathData(path);
    }
}
 

private void createNode(String path) throws ZkException {
    if (StringUtils.isNoneBlank(path) && !zkClient.exists(path)) {
        String parentPath = path.substring(0, path.lastIndexOf("/"));
        createNode(parentPath);
        this.zkClient.create(path, "carrera".getBytes(), CreateMode.PERSISTENT);
    }
}
 

public void setData(String path, String data, int expectedVersion) throws ZkException {
    try {
        this.zkClient.writeData(path, data.getBytes(), expectedVersion);
    } catch (ZkNoNodeException e) {
        createNode(path);
        this.zkClient.writeData(path, data.getBytes());
    }
}
 

private void initPathData(String path) throws ZkException, IOException {
    if (zkClient.exists(path)) {
        Stat stat = new Stat();
        getData(path, stat);
        String localConfigPath = getLocalConfigPath(path);
        LOGGER.debug("ParameterDynamicZookeeper - SyncFromZooKeeper zkPath:{}, localPath:{}, version:{}",
                path, localConfigPath, stat.getVersion());
    } else {
        LOGGER.error("ParameterDynamicZookeeper Error - SyncFromZooKeeper Lost A Config File [{}]", path);
    }
}
 

private boolean isZkSaslEnabled() {
    boolean isSecurityEnabled = false;
    boolean zkSaslEnabled = Boolean.parseBoolean(System.getProperty(ZK_SASL_CLIENT, "true"));
    String zkLoginContextName = System.getProperty(ZK_LOGIN_CONTEXT_NAME_KEY, "Client");

    if (!zkSaslEnabled) {
        LOG.warn("Client SASL has been explicitly disabled with " + ZK_SASL_CLIENT);
        return false;
    }

    String loginConfigFile = System.getProperty(JAVA_LOGIN_CONFIG_PARAM);
    if (loginConfigFile != null && loginConfigFile.length() > 0) {
        LOG.info("JAAS File name: " + loginConfigFile);
        File configFile = new File(loginConfigFile);
        if (!configFile.canRead()) {
            throw new IllegalArgumentException("File " + loginConfigFile + "cannot be read.");
        }

        try {
            Configuration loginConf = Configuration.getConfiguration();
            isSecurityEnabled = loginConf.getAppConfigurationEntry(zkLoginContextName) != null;
        } catch (Exception e) {
            throw new ZkException(e);
        }
    }
    return isSecurityEnabled;
}
 

public void configMutliCluster(ZooKeeper zk) {
    if (_serversList.size() == 1) {
        return;
    }
    String cluster1 = _serversList.get(0);
    try {
        if (_serversList.size() > 1) {
            // 强制的声明accessible
            ReflectionUtils.makeAccessible(clientCnxnField);
            ReflectionUtils.makeAccessible(hostProviderField);
            ReflectionUtils.makeAccessible(serverAddressesField);

            // 添加第二组集群列表
            for (int i = 1; i < _serversList.size(); i++) {
                String cluster = _serversList.get(i);
                // 强制获取zk中的地址信息
                ClientCnxn cnxn = (ClientCnxn) ReflectionUtils.getField(clientCnxnField, zk);
                HostProvider hostProvider = (HostProvider) ReflectionUtils.getField(hostProviderField, cnxn);
                List<InetSocketAddress> serverAddrs = (List<InetSocketAddress>) ReflectionUtils.getField(serverAddressesField,
                    hostProvider);
                // 添加第二组集群列表
                serverAddrs.addAll(new ConnectStringParser(cluster).getServerAddresses());
            }
        }
    } catch (Exception e) {
        try {
            if (zk != null) {
                zk.close();
            }
        } catch (InterruptedException ie) {
            // ignore interrupt
        }
        throw new ZkException("zookeeper_create_error, serveraddrs=" + cluster1, e);
    }

}
 

/**
 * Create a persistent Sequential node.
 *
 * @param path
 * @param createParents if true all parent dirs are created as well and no
 * {@link ZkNodeExistsException} is thrown in case the path already exists
 * @throws ZkInterruptedException if operation was interrupted, or a
 * required reconnection got interrupted
 * @throws IllegalArgumentException if called from anything except the
 * ZooKeeper event thread
 * @throws ZkException if any ZooKeeper exception occurred
 * @throws RuntimeException if any other exception occurs
 */
public String createPersistentSequential(String path, boolean createParents) throws ZkInterruptedException,
                                                                            IllegalArgumentException, ZkException,
                                                                            RuntimeException {
    try {
        return create(path, null, CreateMode.PERSISTENT_SEQUENTIAL);
    } catch (ZkNoNodeException e) {
        if (!createParents) {
            throw e;
        }
        String parentDir = path.substring(0, path.lastIndexOf('/'));
        createPersistent(parentDir, createParents);
        return createPersistentSequential(path, createParents);
    }
}
 

/**
 * Create a persistent Sequential node.
 *
 * @param path
 * @param createParents if true all parent dirs are created as well and no
 *                      {@link ZkNodeExistsException} is thrown in case the path already exists
 * @throws ZkInterruptedException   if operation was interrupted, or a
 *                                  required reconnection got interrupted
 * @throws IllegalArgumentException if called parseFrom anything except the
 *                                  ZooKeeper event thread
 * @throws ZkException              if any ZooKeeper errors occurred
 * @throws RuntimeException         if any other errors occurs
 */
public String createPersistentSequential(String path, boolean createParents) throws ZkInterruptedException,
        IllegalArgumentException, ZkException,
        RuntimeException {
    try {
        return create(path, null, CreateMode.PERSISTENT_SEQUENTIAL);
    } catch (ZkNoNodeException e) {
        if (!createParents) {
            throw e;
        }
        String parentDir = path.substring(0, path.lastIndexOf('/'));
        createPersistent(parentDir, createParents);
        return createPersistentSequential(path, createParents);
    }
}
 

private void syncFromZooKeeper() throws ZkException, IOException {
    ConcurrentHashMap<String/*zk path*/, Set<IZkDataListener>> listeners = zkClient.getDataListener();
    Enumeration<String> paths = listeners.keys();
    while (paths.hasMoreElements()) {
        String path = paths.nextElement();
        initPathData(path);
    }
}
 

private void createNode(String path) throws ZkException {
    if (StringUtils.isNoneBlank(path) && !zkClient.exists(path)) {
        String parentPath = path.substring(0, path.lastIndexOf("/"));
        createNode(parentPath);
        this.zkClient.create(path, "carrera".getBytes(), CreateMode.PERSISTENT);
    }
}
 

public void setData(String path, String data, int expectedVersion) throws ZkException {
    try {
        this.zkClient.writeData(path, data.getBytes(), expectedVersion);
    } catch (ZkNoNodeException e) {
        createNode(path);
        this.zkClient.writeData(path, data.getBytes());
    }
}
 

private void initPathData(String path) throws ZkException, IOException {
    if (zkClient.exists(path)) {
        Stat stat = new Stat();
        getData(path, stat);
        String localConfigPath = getLocalConfigPath(path);
        LOGGER.debug("ParameterDynamicZookeeper - SyncFromZooKeeper zkPath:{}, localPath:{}, version:{}",
                path, localConfigPath, stat.getVersion());
    } else {
        LOGGER.error("ParameterDynamicZookeeper Error - SyncFromZooKeeper Lost A Config File [{}]", path);
    }
}
 

private boolean isZkSaslEnabled() {
    boolean isSecurityEnabled = false;
    boolean zkSaslEnabled = Boolean.parseBoolean(System.getProperty(ZK_SASL_CLIENT, "true"));
    String zkLoginContextName = System.getProperty(ZK_LOGIN_CONTEXT_NAME_KEY, "Client");

    if (!zkSaslEnabled) {
        LOG.warn("Client SASL has been explicitly disabled with " + ZK_SASL_CLIENT);
        return false;
    }

    String loginConfigFile = System.getProperty(JAVA_LOGIN_CONFIG_PARAM);
    if (loginConfigFile != null && loginConfigFile.length() > 0) {
        LOG.info("JAAS File name: " + loginConfigFile);
        File configFile = new File(loginConfigFile);
        if (!configFile.canRead()) {
            throw new IllegalArgumentException("File " + loginConfigFile + "cannot be read.");
        }

        try {
            Configuration loginConf = Configuration.getConfiguration();
            isSecurityEnabled = loginConf.getAppConfigurationEntry(zkLoginContextName) != null;
        } catch (Exception e) {
            throw new ZkException(e);
        }
    }
    return isSecurityEnabled;
}
 

/**
 * Returns all {@link Acl}s defined in the Kafka cluster
 *
 * @return unmodifiable map of all {@link Acl}s defined in the Kafka cluster
 *
 * @throws AdminOperationException
 *      if there is an issue reading the {@link Acl}s
 */
public Map<Resource, Set<Acl>> getAcls() {
    LOG.debug("Fetching all ACLs");
    try {
        return convertKafkaAclMap(getAuthorizer().getAcls());
    } catch (ZkException | ZooKeeperClientException e) {
        throw new AdminOperationException("Unable to retrieve all ACLs", e);
    }
}
 

/**
 * Returns all {@link Acl}s associated to the given {@link KafkaPrincipal}
 *
 * @param principal
 *      the {@link KafkaPrincipal} to look up {@link Acl}s for
 * @return unmodifiable map of all {@link Acl}s associated to the given {@link KafkaPrincipal}
 * @throws IllegalArgumentException
 *      if principal is {@code null}
 * @throws AdminOperationException
 *      if there is an issue reading the {@link Acl}s
 */
public Map<Resource, Set<Acl>> getAcls(KafkaPrincipal principal) {
    if (principal == null)
        throw new IllegalArgumentException("principal cannot be null");

    LOG.debug("Fetching all ACLs for principal [{}]", principal);

    try {
        return convertKafkaAclMap(getAuthorizer().getAcls(principal));
    } catch (ZkException | ZooKeeperClientException e) {
        throw new AdminOperationException("Unable to retrieve ACLs for principal: " + principal, e);
    }
}
 

/**
 * Returns all {@link Acl}s associated to the given {@link Resource}
 *
 * @param resource
 *      the {@link Resource} to look up {@link Acl}s for
 * @return unmodifiable set of all {@link Acl}s associated to the given {@link Resource}
 * @throws IllegalArgumentException
 *      if resource is {@code null}
 * @throws AdminOperationException
 *      if there is an issue reading the {@link Acl}s
 */
public Set<Acl> getAcls(Resource resource) {
    if (resource == null)
        throw new IllegalArgumentException("resource cannot be null");

    LOG.debug("Fetching all ACLs for resource [{}]", resource);

    try {
        return Collections.unmodifiableSet(convertToJavaSet(getAuthorizer().getAcls(resource).iterator()));
    } catch (ZkException | ZooKeeperClientException e) {
        throw new AdminOperationException("Unable to retrieve ACLs for resource: " + resource, e);
    }
}
 

/**
 * Adds the given {@link Acl}s to the {@link Resource}
 *
 * @param acls
 *      the {@link Acl}s to add
 * @param resource
 *      the {@link Resource} to add the {@link Acl}s to
 * @throws IllegalArgumentException
 *      if acls or resource is {@code null}
 * @throws AdminOperationException
 *      if there is an issue adding the {@link Acl}s
 */
public void addAcls(Set<Acl> acls, Resource resource) {
    if (acls == null)
        throw new IllegalArgumentException("acls cannot be null");
    if (resource == null)
        throw new IllegalArgumentException("resource cannot be null");

    LOG.debug("Adding ACLs [{}] for resource [{}]", acls, resource);

    try {
        getAuthorizer().addAcls(toImmutableScalaSet(acls), resource);
    } catch (ZkException | ZooKeeperClientException | IllegalStateException e) {
        throw new AdminOperationException("Unable to add ACLs for resource: " + resource, e);
    }
}
 

/**
 * Removes the given {@link Acl}s from the {@link Resource}
 *
 * @param acls
 *      the {@link Acl}s to remove
 * @param resource
 *      the {@link Resource} to remove the {@link Acl}s from
 * @throws IllegalArgumentException
 *      if acls or resource is {@code null}
 * @throws AdminOperationException
 *      if there is an issue removing the {@link Acl}s
 */
public void removeAcls(Set<Acl> acls, Resource resource) {
    if (acls == null)
        throw new IllegalArgumentException("acls cannot be null");
    if (resource == null)
        throw new IllegalArgumentException("resource cannot be null");

    LOG.debug("Removing ACLs [{}] for resource [{}]", acls, resource);

    try {
        getAuthorizer().removeAcls(toImmutableScalaSet(acls), resource);
    } catch (ZkException | ZooKeeperClientException e) {
        throw new AdminOperationException("Unable to remove ACLs for resource: " + resource, e);
    }
}
 

@Override
protected Registry createRegistry(URL registryUrl) {
    try {
        int timeout = registryUrl.getIntParameter(URLParam.registryConnectTimeout.getName(), URLParam.registryConnectTimeout.getIntValue());
        int sessionTimeout =
                registryUrl.getIntParameter(URLParam.registrySessionTimeout.getName(),
                        URLParam.registrySessionTimeout.getIntValue());
        ZkClient zkClient = new ZkClient(registryUrl.getParameter(URLParam.registryAddress.getName()), sessionTimeout, timeout);
        return new ZookeeperRegistry(registryUrl, zkClient);
    } catch (ZkException e) {
        throw e;
    }
}
 

/**
    * 创建一个持久的顺序节点
    */
public String createPersistentSequential(String path, boolean createParents)
		throws ZkInterruptedException, IllegalArgumentException, ZkException, RuntimeException {
       try {
           return create(path, null, CreateMode.PERSISTENT_SEQUENTIAL);
       } catch (ZkNoNodeException e) {
           if (!createParents) {
               throw e;
           }
           String parentDir = path.substring(0, path.lastIndexOf('/'));
           createPersistent(parentDir, createParents);
           return createPersistentSequential(path, createParents);
       }
   }
 

public String createPersistentSequential(String path, Object data, boolean createParents)
throws ZkInterruptedException, IllegalArgumentException, ZkException, RuntimeException {
     try {
         return create(path, data, CreateMode.PERSISTENT_SEQUENTIAL);
     } catch (ZkNoNodeException e) {
         if (!createParents) {
             throw e;
         }
         String parentDir = path.substring(0, path.lastIndexOf('/'));
         createPersistent(parentDir, createParents);
         return createPersistentSequential(path, data, createParents);
     }
 }
 

/**
 * Gracefully disconnect from ZooKeeper
 */
public void disconnect() {

  if (_zkclient != null) {
    try {
      // remove the liveinstance node
      String liveInstancePath = KeyBuilder.liveInstance(_cluster, _liveInstanceName);
      LOG.info("deleting live instance node: " + liveInstancePath);
      _zkclient.delete(liveInstancePath);

      // NOTE: we should not delete the instance node which still holds the
      // assigned tasks. Coordinator will call cleanUpDeadInstanceDataAndOtherUnusedTasks
      // to do an ad-hoc cleanUp once the tasks haven been properly handled
      // per the strategy (reassign or discard).
    } catch (ZkException zke) {
      // do nothing, best effort clean up
    } finally {
      if (_assignmentList != null) {
        _assignmentList.close();
        _assignmentList = null;
      }
      _zkclient.close();
      _zkclient = null;
      _leaderElectionListener = null;
    }
  }
  // isLeader will be reinitialized when we reconnect
}