public static ZooKeeperClientConfig createConfig(final NiFiProperties nifiProperties) {
    final String connectString = nifiProperties.getProperty(NiFiProperties.ZOOKEEPER_CONNECT_STRING);
    if (connectString == null || connectString.trim().isEmpty()) {
        throw new IllegalStateException("The '" + NiFiProperties.ZOOKEEPER_CONNECT_STRING + "' property is not set in nifi.properties");
    }
    final String cleanedConnectString = cleanConnectString(connectString);
    if (cleanedConnectString.isEmpty()) {
        throw new IllegalStateException("The '" + NiFiProperties.ZOOKEEPER_CONNECT_STRING + "' property is set in nifi.properties but needs to be in pairs of host:port separated by commas");
    }
    final long sessionTimeoutMs = getTimePeriod(nifiProperties, NiFiProperties.ZOOKEEPER_SESSION_TIMEOUT, NiFiProperties.DEFAULT_ZOOKEEPER_SESSION_TIMEOUT);
    final long connectionTimeoutMs = getTimePeriod(nifiProperties, NiFiProperties.ZOOKEEPER_CONNECT_TIMEOUT, NiFiProperties.DEFAULT_ZOOKEEPER_CONNECT_TIMEOUT);
    final String rootPath = nifiProperties.getProperty(NiFiProperties.ZOOKEEPER_ROOT_NODE, NiFiProperties.DEFAULT_ZOOKEEPER_ROOT_NODE);

    try {
        PathUtils.validatePath(rootPath);
    } catch (final IllegalArgumentException e) {
        throw new IllegalArgumentException("The '" + NiFiProperties.ZOOKEEPER_ROOT_NODE + "' property in nifi.properties is set to an illegal value: " + rootPath);
    }

    return new ZooKeeperClientConfig(cleanedConnectString, (int) sessionTimeoutMs, (int) connectionTimeoutMs, rootPath);
}
 

@VisibleForTesting
public static String
validateAndGetFullLedgerAllocatorPoolPath(DistributedLogConfiguration conf, URI uri) throws IOException {
    String poolPath = conf.getLedgerAllocatorPoolPath();
    LOG.info("PoolPath is {}", poolPath);
    if (null == poolPath || !poolPath.startsWith(".") || poolPath.endsWith("/")) {
        LOG.error("Invalid ledger allocator pool path specified when enabling ledger allocator pool: {}", poolPath);
        throw new IOException("Invalid ledger allocator pool path specified : " + poolPath);
    }
    String poolName = conf.getLedgerAllocatorPoolName();
    if (null == poolName) {
        LOG.error("No ledger allocator pool name specified when enabling ledger allocator pool.");
        throw new IOException("No ledger allocator name specified when enabling ledger allocator pool.");
    }
    String rootPath = uri.getPath() + "/" + poolPath + "/" + poolName;
    try {
        PathUtils.validatePath(rootPath);
    } catch (IllegalArgumentException iae) {
        LOG.error("Invalid ledger allocator pool path specified when enabling ledger allocator pool: {}", poolPath);
        throw new IOException("Invalid ledger allocator pool path specified : " + poolPath);
    }
    return rootPath;
}
 

static String validateAndGetFullLedgerAllocatorPoolPath(DistributedLogConfiguration conf, URI uri) throws IOException {
    String poolPath = conf.getLedgerAllocatorPoolPath();
    LOG.info("PoolPath is {}", poolPath);
    if (null == poolPath || !poolPath.startsWith(".") || poolPath.endsWith("/")) {
        LOG.error("Invalid ledger allocator pool path specified when enabling ledger allocator pool : {}", poolPath);
        throw new IOException("Invalid ledger allocator pool path specified : " + poolPath);
    }
    String poolName = conf.getLedgerAllocatorPoolName();
    if (null == poolName) {
        LOG.error("No ledger allocator pool name specified when enabling ledger allocator pool.");
        throw new IOException("No ledger allocator name specified when enabling ledger allocator pool.");
    }
    String rootPath = uri.getPath() + "/" + poolPath + "/" + poolName;
    try {
        PathUtils.validatePath(rootPath);
    } catch (IllegalArgumentException iae) {
        LOG.error("Invalid ledger allocator pool path specified when enabling ledger allocator pool : {}", poolPath);
        throw new IOException("Invalid ledger allocator pool path specified : " + poolPath);
    }
    return rootPath;
}
 

public byte[] put(final String key, final byte[] value) throws InterruptedException {
  Preconditions.checkArgument(key.indexOf('/') == -1);
  PathUtils.validatePath(ZKPaths.makePath(path, key));
  final byte[] prev;
  synchronized (lock) {
    final Map<String, byte[]> mutable = Maps.newHashMap(entries.get());
    prev = mutable.put(key, value);
    try {
      entries.set(ImmutableMap.copyOf(mutable));
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }
  reactor.signal();
  return prev;
}
 

private byte[] remove(final String key) throws InterruptedException {
  Preconditions.checkArgument(key.indexOf('/') == -1);
  PathUtils.validatePath(ZKPaths.makePath(path, key));
  final byte[] value;
  synchronized (lock) {
    final Map<String, byte[]> mutable = Maps.newHashMap(entries.get());
    value = mutable.remove(key);
    try {
      entries.set(ImmutableMap.copyOf(mutable));
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }
  reactor.signal();
  return value;
}
 

/**
 * Validate ZK path with the path itself included in
 * the exception text
 * @param path path to validate
 * @return the path parameter
 * @throws InvalidPathnameException if the pathname is invalid.
 */
public static String validateZKPath(String path) throws
    InvalidPathnameException {
  try {
    PathUtils.validatePath(path);

  } catch (IllegalArgumentException e) {
    throw new InvalidPathnameException(path,
        "Invalid Path \"" + path + "\" : " + e, e);
  }
  return path;
}
 

/**
 * Assert the path is valid by ZK rules
 * @param path path to check
 */
public static void assertValidZKPath(String path) {
  try {
    PathUtils.validatePath(path);
  } catch (IllegalArgumentException e) {
    throw new IllegalArgumentException("Invalid Path " + path + ": " + e, e);
  }
}
 

/**
 * Validate ZK path with the path itself included in
 * the exception text
 * @param path path to validate
 * @return the path parameter
 * @throws InvalidPathnameException if the pathname is invalid.
 */
public static String validateZKPath(String path) throws
    InvalidPathnameException {
  try {
    PathUtils.validatePath(path);

  } catch (IllegalArgumentException e) {
    throw new InvalidPathnameException(path,
        "Invalid Path \"" + path + "\" : " + e, e);
  }
  return path;
}
 

/**
 * Assert the path is valid by ZK rules
 * @param path path to check
 */
public static void assertValidZKPath(String path) {
  try {
    PathUtils.validatePath(path);
  } catch (IllegalArgumentException e) {
    throw new IllegalArgumentException("Invalid Path " + path + ": " + e, e);
  }
}
 

private String toPath(String key) {
    checkArgument(key.indexOf('/') == -1, "Keys may not contain '/'.");
    // The key may contain special characters which are invalid in ZooKeeper paths.  Encode them
    String encodedKey = encodeKey(key);
    String path = ZKPaths.makePath(_zkPath, encodedKey);
    PathUtils.validatePath(path);
    return path;
}
 

/**
 * Validates a ZooKeeper path.
 */
public static String validatePath(String path, String name) {
    requireNonNull(path, name);
    if (path.indexOf('/') > 0) {
        throw new IllegalArgumentException(name + " cannot have '/'. " + name + ": " + path);
    }
    try {
        // Simply prepend '/' to validate the path.
        PathUtils.validatePath('/' + path);
    } catch (IllegalArgumentException e) {
        throw new IllegalArgumentException(name + ": " + path + " (reason: " + e.getMessage() + ')');
    }
    return path;
}
 

private static String validateZNodePath(String znodePath) {
    try {
        PathUtils.validatePath(znodePath);
    } catch (IllegalArgumentException e) {
        throw new IllegalArgumentException("znodePath: " + znodePath +
                                           " (reason: " + e.getMessage() + ')');
    }
    return znodePath;
}
 

private static void validate(String nodeName)
{
    if ( isParameter(Objects.requireNonNull(nodeName, "nodeName cannot be null")) )
    {
        return;
    }
    if ( nodeName.equals(PATH_SEPARATOR) )
    {
        return;
    }
    PathUtils.validatePath(PATH_SEPARATOR + nodeName);
}
 

public static ZooKeeperClientConfig createConfig(final NiFiProperties nifiProperties) {
    final String connectString = nifiProperties.getProperty(NiFiProperties.ZOOKEEPER_CONNECT_STRING);
    if (connectString == null || connectString.trim().isEmpty()) {
        throw new IllegalStateException("The '" + NiFiProperties.ZOOKEEPER_CONNECT_STRING + "' property is not set in nifi.properties");
    }
    final String cleanedConnectString = cleanConnectString(connectString);
    if (cleanedConnectString.isEmpty()) {
        throw new IllegalStateException("The '" + NiFiProperties.ZOOKEEPER_CONNECT_STRING +
                "' property is set in nifi.properties but needs to be in pairs of host:port separated by commas");
    }
    final long sessionTimeoutMs = getTimePeriod(nifiProperties, NiFiProperties.ZOOKEEPER_SESSION_TIMEOUT, NiFiProperties.DEFAULT_ZOOKEEPER_SESSION_TIMEOUT);
    final long connectionTimeoutMs = getTimePeriod(nifiProperties, NiFiProperties.ZOOKEEPER_CONNECT_TIMEOUT, NiFiProperties.DEFAULT_ZOOKEEPER_CONNECT_TIMEOUT);
    final String rootPath = nifiProperties.getProperty(NiFiProperties.ZOOKEEPER_ROOT_NODE, NiFiProperties.DEFAULT_ZOOKEEPER_ROOT_NODE);
    final String authType = nifiProperties.getProperty(NiFiProperties.ZOOKEEPER_AUTH_TYPE,NiFiProperties.DEFAULT_ZOOKEEPER_AUTH_TYPE);
    final String authPrincipal = nifiProperties.getKerberosServicePrincipal();
    final String removeHostFromPrincipal = nifiProperties.getProperty(NiFiProperties.ZOOKEEPER_KERBEROS_REMOVE_HOST_FROM_PRINCIPAL,
            NiFiProperties.DEFAULT_ZOOKEEPER_KERBEROS_REMOVE_HOST_FROM_PRINCIPAL);
    final String removeRealmFromPrincipal = nifiProperties.getProperty(NiFiProperties.ZOOKEEPER_KERBEROS_REMOVE_REALM_FROM_PRINCIPAL,
            NiFiProperties.DEFAULT_ZOOKEEPER_KERBEROS_REMOVE_REALM_FROM_PRINCIPAL);

    try {
        PathUtils.validatePath(rootPath);
    } catch (final IllegalArgumentException e) {
        throw new IllegalArgumentException("The '" + NiFiProperties.ZOOKEEPER_ROOT_NODE + "' property in nifi.properties is set to an illegal value: " + rootPath);
    }

    return new ZooKeeperClientConfig(cleanedConnectString, (int) sessionTimeoutMs, (int) connectionTimeoutMs, rootPath, authType, authPrincipal, removeHostFromPrincipal, removeRealmFromPrincipal);
}
 

public static String buildPath(String parent, String firstPart, String... restOfParts) {
    PathUtils.validatePath(parent);
    if (firstPart.contains("/")) throw new IllegalArgumentException("only parent may contain / character");
    String path = (parent.equals("/") ? parent : parent+"/")+firstPart;
    for (String part : restOfParts) {
        if (part.contains("/")) throw new IllegalArgumentException("only parent may contain / character");
        path = path+"/"+part;
    }
    PathUtils.validatePath(path);
    return path;
}
 

public static String getName(String path) {
    PathUtils.validatePath(path);
    if (path.equals("/")) {
        throw new IllegalArgumentException("name of / is undefined");
    }
    final int index = path.lastIndexOf('/');
    return path.substring(index+1);
}
 

public static String getParent(String path) {
    PathUtils.validatePath(path);
    if (path.equals("/")) {
        throw new IllegalArgumentException("parent of / is undefined");
    }
    final int index = path.lastIndexOf('/');
    if (index == 0) {
        return "/";
    }
    return path.substring(0, index);
}
 

public MesosLogStreamModule(Options options, ZooKeeperConfig zkClientConfig) {
  this.options = options;
  requireArg(options.logPath, "native_log_file_path");
  requireArg(options.zkLogGroupPath, "native_log_zk_group_path");
  PathUtils.validatePath(options.zkLogGroupPath);
  this.zkClientConfig = zkClientConfig;
}
 

@Override
public void rdelete(String path) throws RegistryException {
  checkConnected();
  try {
    PathUtils.validatePath(path);
    List<String> tree = ZKUtil.listSubTreeBFS(zkClient, realPath(path));
    for (int i = tree.size() - 1; i >= 0 ; --i) {
      //Delete the leaves first and eventually get rid of the root
      zkClient.delete(tree.get(i), -1); //Delete all versions of the node with -1.
    }
  } catch (InterruptedException | KeeperException e) {
    throw new RegistryException(ErrorCode.Unknown, e) ;
  }
}
 

public List<String> findDencendantRealPaths(String path) throws RegistryException {
  checkConnected();
  try {
    PathUtils.validatePath(realPath(path));
    return ZKUtil.listSubTreeBFS(zkClient, realPath(path));
  } catch (InterruptedException | KeeperException e) {
    throw new RegistryException(ErrorCode.Unknown, e) ;
  }
}
 

public void rcopy(String path, String toPath) throws RegistryException {
  try {
    PathUtils.validatePath(path);
    List<String> tree = ZKUtil.listSubTreeBFS(zkClient, realPath(path));
    for (int i = 0; i < tree.size(); i++) {
      String selPath = tree.get(i);
      String selToPath = selPath.replace(path, toPath);
      byte[] data = zkClient.getData(selPath, false, new Stat()) ;
      zkClient.create(selToPath, data, DEFAULT_ACL, toCreateMode(NodeCreateMode.PERSISTENT)) ;
    }
  } catch (InterruptedException | KeeperException e) {
    throw new RegistryException(ErrorCode.Unknown, e) ;
  }
}
 

@Override
public int parseArguments(Parameters params) throws CmdLineException {
    String param = params.getParameter(0);
    try {
        PathUtils.validatePath(param);
        setter.addValue(param);
        return 1;
    } catch (IllegalArgumentException e) {
        throw new CmdLineException(owner,
                String.format("\"%s\" is not a valid value for \"%s\"", param, params.getParameter(-1)));
    }
}
 

private static void zkCopy(HelixZkClient srcClient, String srcRootPath, HelixZkClient dstClient,
    String dstRootPath) {
  // Strip off tailing "/"
  if (!srcRootPath.equals("/") && srcRootPath.endsWith("/")) {
    srcRootPath = srcRootPath.substring(0, srcRootPath.length() - 1);
  }

  if (!dstRootPath.equals("/") && dstRootPath.endsWith("/")) {
    dstRootPath = dstRootPath.substring(0, dstRootPath.length() - 1);
  }

  // Validate paths
  PathUtils.validatePath(srcRootPath);
  PathUtils.validatePath(dstRootPath);

  if (srcRootPath.equals(dstRootPath)) {
    logger.info("srcPath == dstPath. Skip copying");
    return;
  }

  if (srcRootPath.startsWith(dstRootPath) || dstRootPath.startsWith(srcRootPath)) {
    throw new IllegalArgumentException(
        "srcPath/dstPath can't be prefix of dstPath/srcPath, was srcPath: " + srcRootPath
            + ", dstPath: " + dstRootPath);
  }

  // Recursive copy using BFS
  List<String> queue = new LinkedList<String>();
  String root = "";
  copy(srcClient, srcRootPath, dstClient, dstRootPath, Arrays.asList(root));

  queue.add(root);
  while (!queue.isEmpty()) {
    String path = queue.remove(0);
    String fromPath = concatenate(srcRootPath, path);

    List<String> children = srcClient.getChildren(fromPath);
    List<String> paths = new ArrayList<String>();
    if (children != null && children.size() > 0) {
      for (String child : children) {
        String childPath = concatenate(path, child);
        paths.add(childPath);
      }
      copy(srcClient, srcRootPath, dstClient, dstRootPath, paths);
      queue.addAll(paths);
    }
  }
}
 

public PathFactory(final String base, final String... parts) {
  final String joined = join(base, parts);
  this.base = joined.startsWith("/") ? joined : "/" + joined;
  PathUtils.validatePath(base);
}
 

/**
 * Validate and return a normalized zookeeper path which doesn't contain consecutive slashes and
 * never ends with a slash (except for root path).
 *
 * @param path the path to be normalized
 * @return normalized path string
 */
static String normalizePath(String path) {
  String normalizedPath = path.replaceAll("//+", "/").replaceFirst("(.+)/$", "$1");
  PathUtils.validatePath(normalizedPath);
  return normalizedPath;
}