void deleteOldVersionPath(String path, List<String> zNodeIds, int numVersionsToLeave, Comparator<String> c) {
    if (StringUtils.isEmpty(path) || zNodeIds == null) {
      LOG.warn("cannot cleanup empty path or empty list in ZK");
      return;
    }
    if (zNodeIds.size() > numVersionsToLeave) {
      Collections.sort(zNodeIds, c);
      // get the znodes to delete
      int size = zNodeIds.size();
      List<String> zNodesToDelete = zNodeIds.subList(0, zNodeIds.size() - numVersionsToLeave);
      LOG.info("Starting cleanup of barrier version zkNodes. From size=" + size + " to size " + zNodesToDelete.size() + "; numberToLeave=" + numVersionsToLeave);
      for (String znodeId : zNodesToDelete) {
        String pathToDelete = path + "/" + znodeId;
        try {
          LOG.info("deleting " + pathToDelete);
          zkClient.deleteRecursive(pathToDelete);
          metrics.deletions.inc();
        } catch (Exception e) {
          LOG.warn("delete of node " + pathToDelete + " failed.", e);
        }
      }
    }
  }
cleanup old data from ZK
@param numVersionsToLeave - number of versions to leave
samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java