/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.samza.zk;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.HashMap;
import java.util.Objects;
import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkBadVersionException;
import org.I0Itec.zkclient.exception.ZkInterruptedException;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.apache.commons.lang3.StringUtils;
import org.apache.samza.SamzaException;
import org.apache.samza.container.TaskName;
import org.apache.samza.job.model.JobModel;
import org.apache.samza.metrics.MetricsRegistry;
import org.apache.samza.runtime.LocationId;
import org.apache.samza.serializers.model.SamzaObjectMapper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Util class to help manage Zk connection and ZkClient.
 * It also provides additional utility methods for read/write/subscribe/unsubscribe access to the ZK tree.
 *
 * <p>
 *  <b>Note on ZkClient:</b>
 *  {@link ZkClient} consists of two threads - I/O thread and Event thread.
 *  I/O thread manages heartbeats to the Zookeeper server in the ensemble and handles responses to synchronous methods
 *  in Zookeeper API.
 *  Event thread typically receives all the Watcher events and delivers to registered listeners. It, also, handles
 *  responses to asynchronous methods in Zookeeper API.
 * </p>
 *
 * <p>
 *  <b>Note on Session disconnect handling:</b>
 *  After the session has timed out, and restored we may still get some notifications from before (from the old
 *  session). To avoid this, we add a currentGeneration member, which starts with 0, and is increased each time
 *  a new session is established. Current value of this member is passed to each Listener when it is created.
 *  So if the Callback from this Listener comes with an old generation id - we ignore it.
 * </p>
 *
 * <p>
 *   <b>Note on Session Management:</b>
 *   Session management, if needed, should be handled by the caller. This can be done by implementing
 *   {@link org.I0Itec.zkclient.IZkStateListener} and subscribing this listener to the current ZkClient. Note: The connection state change
 *   callbacks are invoked in the context of the Event thread of the ZkClient. So, it is advised to do non-blocking
 *   processing in the callbacks.
 * </p>
 */
public class ZkUtils {
  private static final Logger LOG = LoggerFactory.getLogger(ZkUtils.class);
  /* package private */static final String ZK_PROTOCOL_VERSION = "1.0";

  private final ZkClient zkClient;
  private volatile String ephemeralPath = null;
  private final ZkKeyBuilder keyBuilder;
  private final int connectionTimeoutMs;
  private final AtomicInteger currentGeneration;
  private final ZkUtilsMetrics metrics;
  private final int sessionTimeoutMs;

  public void incGeneration() {
    currentGeneration.incrementAndGet();
  }

  public int getGeneration() {
    return currentGeneration.get();
  }

  public ZkUtils(ZkKeyBuilder zkKeyBuilder, ZkClient zkClient, int connectionTimeoutMs, int sessionTimeOutMs, MetricsRegistry metricsRegistry) {
    this.keyBuilder = zkKeyBuilder;
    this.connectionTimeoutMs = connectionTimeoutMs;
    this.zkClient = zkClient;
    this.metrics = new ZkUtilsMetrics(metricsRegistry);
    this.currentGeneration = new AtomicInteger(0);
    this.sessionTimeoutMs = sessionTimeOutMs;
  }

  public void connect() throws ZkInterruptedException {
    boolean isConnected = zkClient.waitUntilConnected(connectionTimeoutMs, TimeUnit.MILLISECONDS);
    if (!isConnected) {
      metrics.zkConnectionError.inc();
      throw new RuntimeException("Unable to connect to Zookeeper within connectionTimeout " + connectionTimeoutMs + "ms. Shutting down!");
    }
  }

  // reset all zk-session specific state
  public synchronized void unregister() {
    ephemeralPath = null;
  }

  public ZkClient getZkClient() {
    return zkClient;
  }

  public ZkKeyBuilder getKeyBuilder() {
    return keyBuilder;
  }

  /**
   * Returns a ZK generated identifier for this client.
   * If the current client is registering for the first time, it creates an ephemeral sequential node in the ZK tree
   * If the current client has already registered and is still within the same session, it returns the already existing
   * value for the ephemeralPath
   *
   * @param data Object that should be written as data in the registered ephemeral ZK node
   * @return String representing the absolute ephemeralPath of this client in the current session
   */
  public synchronized String registerProcessorAndGetId(final ProcessorData data) {
    final long startTimeMs = System.currentTimeMillis();
    final long retryTimeOutMs = 2 * sessionTimeoutMs;
    String processorId = data.getProcessorId();
    if (ephemeralPath == null) {
      ephemeralPath = zkClient.createEphemeralSequential(keyBuilder.getProcessorsPath() + "/", data.toString());
      LOG.info("Created ephemeral path: {} for processor: {} in zookeeper.", ephemeralPath, data);
      while (true) {
        ProcessorNode processorNode = new ProcessorNode(data, ephemeralPath);
        // Determine if there are duplicate processors with this.processorId after registration.
        if (!isValidRegisteredProcessor(processorNode)) {
          long currentTimeMs = System.currentTimeMillis();
          if ((currentTimeMs - startTimeMs) < retryTimeOutMs) {
            LOG.info("Processor: {} is duplicate. Retrying registration again.", processorId);
            timeDelay(1000);
          } else {
            LOG.info("Processor: {} is duplicate. Deleting zookeeper node at path: {}.", processorId, ephemeralPath);
            zkClient.delete(ephemeralPath);
            metrics.deletions.inc();
            throw new SamzaException(String.format("Processor: %s is duplicate in the group. Registration failed.", processorId));
          }
        } else {
          break;
        }
      }
    } else {
      LOG.info("Ephemeral path: {} exists for processor: {} in zookeeper.", ephemeralPath, data);
    }
    return ephemeralPath;
  }

  public void timeDelay(int sleepTimeInMillis) {
    try {
      Thread.sleep(sleepTimeInMillis);
    } catch (InterruptedException e) {
      LOG.error("Interrupted exception on wait.", e);
      Thread.interrupted();
    }
  }

  /**
   * Deletes the ephemeral node of a processor in zookeeper.
   * @param processorId uniqueId identifying the stream processor to delete.
   */
  public synchronized void deleteProcessorNode(String processorId) {
    try {
      if (ephemeralPath != null) {
        LOG.info("Deleting the ephemeral node: {} of the processor: {} in zookeeper.", ephemeralPath, processorId);
        zkClient.delete(ephemeralPath);
      }
    } catch (Exception e) {
      LOG.error("Exception occurred on deletion of ephemeral node: {}.", ephemeralPath, e);
    }
  }

  /**
   * Determines the validity of processor registered with zookeeper.
   *
   * If there are multiple processors registered with same processorId,
   * the processor with lexicographically smallest zookeeperPath is considered valid
   * and all the remaining processors are invalid.
   *
   * Two processors will not have smallest zookeeperPath because of sequentialId guarantees
   * of zookeeper for ephemeral nodes.
   *
   * @param processor to check for validity condition in processors group.
   * @return true if the processor is valid. false otherwise.
   */
  private boolean isValidRegisteredProcessor(final ProcessorNode processor) {
    String processorId = processor.getProcessorData().getProcessorId();
    List<ProcessorNode> processorNodes = getAllProcessorNodes().stream()
                                                               .filter(processorNode -> processorNode.processorData.getProcessorId().equals(processorId))
                                                               .collect(Collectors.toList());
    // Check for duplicate processor condition(if more than one processor exist for this processorId).
    if (processorNodes.size() > 1) {
      // There exists more than processor for provided `processorId`.
      LOG.debug("Processor nodes in zookeeper: {} for processorId: {}.", processorNodes, processorId);
      // Get all ephemeral processor paths
      TreeSet<String> sortedProcessorPaths = processorNodes.stream()
                                                           .map(ProcessorNode::getEphemeralPath)
                                                           .collect(Collectors.toCollection(TreeSet::new));
      // Check if smallest path is equal to this processor's ephemeralPath.
      return sortedProcessorPaths.first().equals(processor.getEphemeralPath());
    }
    // There're no duplicate processors. This is a valid registered processor.
    return true;
  }

  /**
   * Fetches all the ephemeral processor nodes of a standalone job from zookeeper.
   * @return a list of {@link ProcessorNode}, where each ProcessorNode represents a registered stream processor.
   */
  List<ProcessorNode> getAllProcessorNodes() {
    List<String> processorZNodes = getSortedActiveProcessorsZnodes();
    LOG.debug("Active ProcessorZNodes in zookeeper: {}.", processorZNodes);
    List<ProcessorNode> processorNodes = new ArrayList<>();
    for (String processorZNode: processorZNodes) {
      String ephemeralProcessorPath = String.format("%s/%s", keyBuilder.getProcessorsPath(), processorZNode);
      String data = readProcessorData(ephemeralProcessorPath);
      if (data != null) {
        processorNodes.add(new ProcessorNode(new ProcessorData(data), ephemeralProcessorPath));
      }
    }
    return processorNodes;
  }

  public void writeTaskLocality(TaskName taskName, LocationId locationId) {
    String taskLocalityPath = String.format("%s/%s", keyBuilder.getTaskLocalityPath(), taskName);
    validatePaths(new String[] {taskLocalityPath});
    writeData(taskLocalityPath, locationId.getId());
  }

  public Map<TaskName, LocationId> readTaskLocality() {
    Map<TaskName, LocationId> taskLocality = new HashMap<>();
    String taskLocalityPath = keyBuilder.getTaskLocalityPath();
    List<String> tasks = new ArrayList<>();
    if (zkClient.exists(taskLocalityPath)) {
      tasks = zkClient.getChildren(taskLocalityPath);
    }
    for (String taskName : tasks) {
      String taskPath = String.format("%s/%s", keyBuilder.getTaskLocalityPath(), taskName);
      String locationId = zkClient.readData(taskPath, true);
      if (locationId != null) {
        taskLocality.put(new TaskName(taskName), new LocationId(locationId));
      }
    }
    return taskLocality;
  }

  /**
   * Method is used to get the <i>sorted</i> list of currently active/registered processors (znodes)
   *
   * @return List of absolute ZK node paths
   */
  public List<String> getSortedActiveProcessorsZnodes() {
    List<String> znodeIds = zkClient.getChildren(keyBuilder.getProcessorsPath());
    if (znodeIds.size() > 0) {
      Collections.sort(znodeIds);
      LOG.info("Found these children - " + znodeIds);
    }
    return znodeIds;
  }

  /**
   * Method is used to read processor's data from the znode
   * @param fullPath absolute path to the znode
   * @return processor's data
   * @throws SamzaException when fullPath doesn't exist in zookeeper
   * or problems with connecting to zookeeper.
   */
  private String readProcessorData(String fullPath) {
    try {
      String data = zkClient.readData(fullPath, true);
      metrics.reads.inc();
      return data;
    } catch (Exception e) {
      throw new SamzaException(String.format("Cannot read ZK node: %s", fullPath), e);
    }
  }

  /**
   * Method is used to get the list of currently active/registered processor ids
   * @return List of processorIds
   */
  public List<String> getSortedActiveProcessorsIDs() {
    return getActiveProcessorsIDs(getSortedActiveProcessorsZnodes());
  }

  /**
   * Method is used to get the <i>sorted</i> list of processors ids for a given list of znodes
   * @param znodeIds - list of relative paths of the children's znodes
   * @return List of processor ids for a given list of znodes
   */
  public List<String> getActiveProcessorsIDs(List<String> znodeIds) {
    String processorPath = keyBuilder.getProcessorsPath();
    List<String> processorIds = new ArrayList<>(znodeIds.size());
    if (znodeIds.size() > 0) {
      for (String child : znodeIds) {
        String fullPath = String.format("%s/%s", processorPath, child);
        String processorData = readProcessorData(fullPath);
        if (processorData != null) {
          processorIds.add(new ProcessorData(processorData).getProcessorId());
        }
      }
      Collections.sort(processorIds);
      LOG.info("Found these children - " + znodeIds);
      LOG.info("Found these processorIds - " + processorIds);
    }
    return processorIds;
  }

  /* Wrapper for standard I0Itec methods */
  public void unsubscribeDataChanges(String path, IZkDataListener dataListener) {
    zkClient.unsubscribeDataChanges(path, dataListener);
  }

  public void subscribeDataChanges(String path, IZkDataListener dataListener) {
    zkClient.subscribeDataChanges(path, dataListener);
    metrics.subscriptions.inc();
  }

  public void subscribeChildChanges(String path, IZkChildListener listener) {
    zkClient.subscribeChildChanges(path, listener);
    metrics.subscriptions.inc();
  }

  public void unsubscribeChildChanges(String path, IZkChildListener childListener) {
    zkClient.unsubscribeChildChanges(path, childListener);
  }

  public void writeData(String path, Object object) {
    zkClient.writeData(path, object);
    metrics.writes.inc();
  }

  public boolean exists(String path) {
    return zkClient.exists(path);
  }

  public void close() {
    try {
      zkClient.close();
    } catch (ZkInterruptedException e) {
      LOG.warn("Interrupted when closing zkClient. Clearing the interrupted status and retrying.", e);
      Thread.interrupted();
      zkClient.close();
      Thread.currentThread().interrupt();
    }
  }

  /**
   * A generation-aware {@link IZkChildListener} that only responds to events that occur in the current-generation.
   * Each generation is identified by a generation-id which is scoped to the currently active Zk session and
   * is incremented each time a session expires. This ensures that events corresponding to the previous generation
   * are not acted on.
   */
  public abstract static class GenerationAwareZkChildListener implements IZkChildListener {
    private final int generation;
    private final ZkUtils zkUtils;
    private final String listenerName;

    public GenerationAwareZkChildListener(ZkUtils zkUtils, String listenerName) {
      generation = zkUtils.getGeneration();
      this.zkUtils = zkUtils;
      this.listenerName = listenerName;
    }

    @Override
    public void handleChildChange(String barrierParticipantPath, List<String> participantIds) throws Exception {
      int currentGeneration = zkUtils.getGeneration();
      if (currentGeneration != generation) {
        LOG.warn(String.format("Skipping handleChildChange for %s from wrong generation. Current generation: %s; " +
            "Callback generation: %s", listenerName, currentGeneration, generation));
        return;
      }
      doHandleChildChange(barrierParticipantPath, participantIds);
    }

    public abstract void doHandleChildChange(String path, List<String> children) throws Exception;
  }

  /**
   * A generation-aware {@link IZkDataListener} that only responds to events that occur in the current-generation.
   * Each generation is identified by a generation-id which is scoped to the currently active Zk session and
   * is incremented each time a session expires. This ensures that events corresponding to the previous generation
   * are not acted on.
   */
  public abstract static class GenerationAwareZkDataListener implements IZkDataListener {
    private final int generation;
    private final ZkUtils zkUtils;
    private final String listenerName;

    public GenerationAwareZkDataListener(ZkUtils zkUtils, String listenerName) {
      generation = zkUtils.getGeneration();
      this.zkUtils = zkUtils;
      this.listenerName = listenerName;
    }

    @Override
    public void handleDataChange(String path, Object data) {
      if (!isValid()) {
        LOG.warn(String.format("Skipping handleDataChange for %s from wrong generation. Current generation: %s; " +
            "Callback generation: %s", listenerName, zkUtils.getGeneration(), generation));
      } else {
        doHandleDataChange(path, data);
      }
    }

    public void handleDataDeleted(String dataPath) throws Exception {
      if (!isValid()) {
        LOG.warn(String.format("Skipping handleDataDeleted for %s from wrong generation. Current generation: %s; " +
            "Callback generation: %s", listenerName, zkUtils.getGeneration(), generation));
      } else {
        doHandleDataDeleted(dataPath);
      }
    }

    public abstract void doHandleDataChange(String path, Object data);

    public abstract void doHandleDataDeleted(String path);

    private boolean isValid() {
      return generation == zkUtils.getGeneration();
    }
  }

  /**
    * subscribe for changes of JobModel version
    * @param dataListener describe this
    */
  public void subscribeToJobModelVersionChange(GenerationAwareZkDataListener dataListener) {
    LOG.info(" subscribing for jm version change at:" + keyBuilder.getJobModelVersionPath());
    zkClient.subscribeDataChanges(keyBuilder.getJobModelVersionPath(), dataListener);
    metrics.subscriptions.inc();
  }

  /**
   * Publishes new job model into ZK.
   * This call should FAIL if the node already exists.
   * @param jobModelVersion  version of the jobModeL to publish
   * @param jobModel jobModel to publish
   *
   */
  public void publishJobModel(String jobModelVersion, JobModel jobModel) {
    try {
      ObjectMapper mmapper = SamzaObjectMapper.getObjectMapper();
      String jobModelStr = mmapper.writerWithDefaultPrettyPrinter().writeValueAsString(jobModel);
      LOG.info("jobModelAsString=" + jobModelStr);
      zkClient.createPersistent(keyBuilder.getJobModelPath(jobModelVersion), jobModelStr);
      LOG.info("wrote jobModel path =" + keyBuilder.getJobModelPath(jobModelVersion));
    } catch (Exception e) {
      LOG.error("JobModel publish failed for version=" + jobModelVersion, e);
      throw new SamzaException(e);
    }
  }

  /**
   * get the job model from ZK by version
   * @param jobModelVersion jobModel version to get
   * @return job model for this version
   */
  public JobModel getJobModel(String jobModelVersion) {
    LOG.info("Read the model ver=" + jobModelVersion + " from " + keyBuilder.getJobModelPath(jobModelVersion));
    Object data = zkClient.readData(keyBuilder.getJobModelPath(jobModelVersion));
    metrics.reads.inc();
    ObjectMapper mmapper = SamzaObjectMapper.getObjectMapper();
    JobModel jm;
    try {
      jm = mmapper.readValue((String) data, JobModel.class);
    } catch (IOException e) {
      throw new SamzaException("failed to read JobModel from ZK", e);
    }
    return jm;
  }

  /**
   * read the jobmodel version from ZK
   * @return jobmodel version as a string
   */
  public String getJobModelVersion() {
    String jobModelVersion = zkClient.readData(keyBuilder.getJobModelVersionPath(), true);
    metrics.reads.inc();
    return jobModelVersion;
  }

  /**
   * Read the version of the job model that is the most recent agreed version by the quorum.
   * @return most recent active job model version
   */
  public String getLastActiveJobModelVersion() {
    String lastActiveJobModelVersion = zkClient.readData(keyBuilder.getActiveJobModelVersionPath(), true);
    metrics.reads.inc();
    return lastActiveJobModelVersion;
  }

  /**
   * Generates the next JobModel version that should be used by a processor group in a rebalancing phase
   * for coordination.
   * @param currentJobModelVersion the current version of JobModel.
   * @return the next JobModel version.
   */
  public String getNextJobModelVersion(String currentJobModelVersion) {
    if (currentJobModelVersion == null) {
      return  "1";
    } else {
      /**
       * There's inconsistency between the maximum published jobModel version and value stored in jobModelVersion
       * zookeeper node. Short term fix is to read all published jobModel versions and choose the maximum version. If there's a
       * inconsistency, update the jobModelVersionPath with maximum published jobModelVersion.
       */
      List<String> publishedJobModelVersions = zkClient.getChildren(keyBuilder.getJobModelPathPrefix());
      metrics.reads.inc(publishedJobModelVersions.size());
      String maxPublishedJMVersion = publishedJobModelVersions.stream()
                                                              .max(Comparator.comparingInt(Integer::valueOf)).orElse("0");
      return Integer.toString(Math.max(Integer.valueOf(currentJobModelVersion), Integer.valueOf(maxPublishedJMVersion)) + 1);
    }
  }

  /**
   * publish the version number of the next JobModel
   * @param oldVersion - used to validate, that no one has changed the version in the meanwhile.
   * @param newVersion - new version.
   */
  public void publishJobModelVersion(String oldVersion, String newVersion) {
    Stat stat = new Stat();
    String currentVersion = zkClient.readData(keyBuilder.getJobModelVersionPath(), stat);
    metrics.reads.inc();
    LOG.info("publishing new version: " + newVersion + "; oldVersion = " + oldVersion + "(" + stat
        .getVersion() + ")");

    if (currentVersion != null && !currentVersion.equals(oldVersion)) {
      throw new SamzaException(
          "Someone changed JobModelVersion while the leader was generating one: expected" + oldVersion + ", got " + currentVersion);
    }
    // data version is the ZK version of the data from the ZK.
    int dataVersion = stat.getVersion();
    try {
      stat = zkClient.writeDataReturnStat(keyBuilder.getJobModelVersionPath(), newVersion, dataVersion);
      metrics.writes.inc();
    } catch (Exception e) {
      String msg = "publish job model version failed for new version = " + newVersion + "; old version = " + oldVersion;
      LOG.error(msg, e);
      throw new SamzaException(msg, e);
    }
    LOG.info("published new version: " + newVersion + "; expected data version = " + (dataVersion + 1) +
        "(actual data version after update = " + stat.getVersion() + ")");
  }

  /**
   * Publish to most recent job model version that is agreed by the quorum.
   * @param version active job model version
   */
  public void publishActiveJobModelVersion(String version) {
    try {
      zkClient.writeData(keyBuilder.getActiveJobModelVersionPath(), version);
      metrics.writes.inc();
      LOG.info("Published the active job model version = {} to zookeeper successfully.", version);
    } catch (Exception e) {
      /*
       * Failure to write the most recent job model version has the following implications
       *  1. Processors no longer benefit from the optimization to start the container upon restarts based
       *     on the most recent active job model. It is useful in scenarios where processors leave the quorum and
       *     comeback within debounce time and the work assignment for new quorum remains unchanged.
       *  2. During rolling upgrades, processors that are upgraded initially will wait for the min(deployment
       *     window, T + debounce time) where T is the time at which the last change notification of the quorum was received
       *     by the leader.
       *
       *  That said, failures don't impact correctness and is better to continue processing as opposed to bringing down
       *  the processor as a fatal error.
       */
      LOG.warn("Failed to persist the active job model version = {} due to {}", version, e);
    }
  }

  // validate that Zk protocol currently used by the job is the same as in this participant
  public void validateZkVersion() {

    // Version of the protocol is written into root znode. If root does not exist yet we need to create one.
    String rootPath = keyBuilder.getRootPath();
    if (!zkClient.exists(rootPath)) {
      try {
        // attempt to create the root with the correct version
        zkClient.createPersistent(rootPath, ZK_PROTOCOL_VERSION);
        LOG.info("Created zk root node: " + rootPath + " with zk version " + ZK_PROTOCOL_VERSION);
        return;
      } catch (ZkNodeExistsException e) {
        // ignoring
        LOG.warn("root path " + rootPath + " already exists.");
      }
    }
    // if exists, verify the version
    Stat stat = new Stat();
    String version = zkClient.readData(rootPath, stat);
    if (version == null) {
      // for backward compatibility, if no value - assume 1.0
      try {
        zkClient.writeData(rootPath, "1.0", stat.getVersion());
      } catch (ZkBadVersionException e) {
        // if the write failed with ZkBadVersionException it means someone else already wrote a version, so we can ignore it.
      }
      // re-read the updated version
      version = zkClient.readData(rootPath);
    }
    LOG.info("Current version for zk root node: " + rootPath + " is " + version + ", expected version is " + ZK_PROTOCOL_VERSION);
    if (!version.equals(ZK_PROTOCOL_VERSION)) {
      throw new SamzaException("ZK Protocol mismatch. Expected " + ZK_PROTOCOL_VERSION + "; found " + version);
    }
  }

  /**
   * verify that given paths exist in ZK
   * @param paths - paths to verify or create
   */
  public void validatePaths(String[] paths) {
    for (String path : paths) {
      if (!zkClient.exists(path)) {
        zkClient.createPersistent(path, true);
      }
    }
  }

  /**
   * subscribe to the changes in the list of processors in ZK
   * @param listener - will be called when a processor is added or removed.
   */
  public void subscribeToProcessorChange(IZkChildListener listener) {
    LOG.info("Subscribing for child change at:" + keyBuilder.getProcessorsPath());
    zkClient.subscribeChildChanges(keyBuilder.getProcessorsPath(), listener);
    metrics.subscriptions.inc();
  }

  /**
   * cleanup old data from ZK
   * @param numVersionsToLeave - number of versions to leave
   */
  public void cleanupZK(int numVersionsToLeave) {
    deleteOldBarrierVersions(numVersionsToLeave);
    deleteOldJobModels(numVersionsToLeave);
  }

  void deleteOldJobModels(int numVersionsToLeave) {
    // read current list of JMs
    String path = keyBuilder.getJobModelPathPrefix();
    LOG.info("About to delete jm path=" + path);
    List<String> znodeIds = zkClient.getChildren(path);
    deleteOldVersionPath(path, znodeIds, numVersionsToLeave, new Comparator<String>() {
      @Override
      public int compare(String o1, String o2) {
        // jm version name format is <num>
        return Integer.valueOf(o1) - Integer.valueOf(o2);
      }
    });
  }

  void deleteOldBarrierVersions(int numVersionsToLeave) {
    // read current list of barriers
    String path = keyBuilder.getJobModelVersionBarrierPrefix();
    LOG.info("About to delete old barrier paths from " + path);
    List<String> znodeIds = zkClient.getChildren(path);
    LOG.info("List of all zkNodes: " + znodeIds);
    deleteOldVersionPath(path, znodeIds, numVersionsToLeave,  new Comparator<String>() {
      @Override
      public int compare(String o1, String o2) {
        // barrier's name format is barrier_<num>
        return ZkBarrierForVersionUpgrade.getVersion(o1) - ZkBarrierForVersionUpgrade.getVersion(o2);
      }
    });
  }

  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);
        }
      }
    }
  }
  /**
   * Represents zookeeper processor node.
   */
  static class ProcessorNode {
    private final ProcessorData processorData;

    // Ex: /test/processors/0000000000
    private final String ephemeralProcessorPath;

    ProcessorNode(ProcessorData processorData, String ephemeralProcessorPath) {
      this.processorData = processorData;
      this.ephemeralProcessorPath = ephemeralProcessorPath;
    }

    ProcessorData getProcessorData() {
      return processorData;
    }

    String getEphemeralPath() {
      return ephemeralProcessorPath;
    }

    @Override
    public String toString() {
      return String.format("[ProcessorData: %s, ephemeralProcessorPath: %s]", processorData, ephemeralProcessorPath);
    }

    @Override
    public int hashCode() {
      return Objects.hash(processorData, ephemeralProcessorPath);
    }

    @Override
    public boolean equals(Object obj) {
      if (obj == null) return false;
      if (getClass() != obj.getClass()) return false;
      final ProcessorNode other = (ProcessorNode) obj;
      return Objects.equals(processorData, other.processorData) && Objects.equals(ephemeralProcessorPath, other.ephemeralProcessorPath);
    }
  }
}