/**
 * 
 * @param path
 * @param content
 * @param createMode
 * @throws Exception
 */
public static void createOrUpdate(String path, byte[] content, CreateMode createMode) throws Exception {
	if (!createMode.isSequential()) {
		if (exists(path)) {
			log.info("节点已经存在:{}", path);
			if (content != null) {
				setData(path, content);
			}
			return;
		}
	}

	try {
		zkclient.create().creatingParentsIfNeeded().withMode(createMode).forPath(path, content);
	} catch (NodeExistsException e) {
		//			node exists skip it
		//			log.error(e.toString(), e);
	}

	return;
}
 

private void createInstanceNodeForInitialization() throws Exception {
	String parent = String.format("%s/%s/instances", CONSTANTS_ROOT_PATH, CommonUtils.getApplication(this.endpoint));
	String path = String.format("%s/%s", parent, this.endpoint);
	byte[] versionByteArray = ByteUtils.longToByteArray(this.instanceVersion);
	try {
		this.curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath(path, versionByteArray);
	} catch (NodeExistsException error) {
		try {
			this.instanceLock.lock();
			this.instanceInited = false;
			this.curatorFramework.checkExists().usingWatcher(this).inBackground(this).forPath(path);

			this.instanceCond.await(this.initializeWaitingSeconds, TimeUnit.SECONDS);
			if (this.instanceInited == false) {
				throw error;
			}
		} catch (InterruptedException ignore) {
			// ignore
		} finally {
			this.instanceLock.unlock();
		}
	}
}
 

public static String createIfAbsent(ZooKeeper zk, String path, byte[] data, CreateMode createMode, boolean gc)
        throws KeeperException, InterruptedException {
    String pathCreated = null;
    try {
        pathCreated = zk.create(path, data, Ids.OPEN_ACL_UNSAFE, createMode);
    } catch (NodeExistsException e) {
        // OK
        LOG.debug("Create skipped for existing znode: path={}", path);
    }
    // reset if what exists is the ephemeral garbage.
    if (gc && (pathCreated == null) && CreateMode.EPHEMERAL.equals(createMode)) {
        Stat stat = zk.exists(path, false);
        if (stat != null && zk.getSessionId() != stat.getEphemeralOwner()) {
            deleteIfExists(zk, path, -1);
            pathCreated = zk.create(path, data, Ids.OPEN_ACL_UNSAFE, createMode);
        }
    }
    return pathCreated;
}
 

@Override
public void createInputConfig(String clusterName, String serviceName, String inputConfig) throws Exception {
  String nodePath = String.format("/%s/input/%s", clusterName, serviceName);
  try {
    client.create().creatingParentContainersIfNeeded().withACL(LogSearchConfigZKHelper.getAcls(properties)).forPath(nodePath, inputConfig.getBytes());
    logger.info("Uploaded input config for the service " + serviceName + " for cluster " + clusterName);
  } catch (NodeExistsException e) {
    logger.debug("Did not upload input config for service " + serviceName + " as it was already uploaded by another Log Feeder");
  }
}
 

/**
 * 
 * @param path
 * @param bs
 * @throws Exception
 */
public static void setData(String path, byte[] bs) throws Exception {
	if (bs != null) {
		if (Zk.exists(path)) {
			zkclient.setData().forPath(path, bs);
		} else {
			try {
				zkclient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path, bs);
			} catch (NodeExistsException e) {
				//节点已经存在, skip it
				//log.error(e.toString(), e);
			}
		}
	}
}
 

public void createPersistent(String path) {
    try {
        client.create().forPath(path);
    } catch (NodeExistsException ignore) {
    } catch (Exception e) {
        throw new IllegalStateException(e.getMessage(), e);
    }
}
 

public void createEphemeral(String path) {
    try {
        client.create().withMode(CreateMode.EPHEMERAL).forPath(path);
    } catch (NodeExistsException ignore) {
    } catch (Exception e) {
        throw new IllegalStateException(e.getMessage(), e);
    }
}
 

public void atomicUpdate(String path, BiFunction<Stat , byte[], byte[]> editor) throws KeeperException, InterruptedException {
  for (; ; ) {
    byte[] modified = null;
    byte[] zkData = null;
    Stat s = new Stat();
    try {
      if (exists(path, true)) {
        zkData = getData(path, null, s, true);
        modified = editor.apply(s, zkData);
        if (modified == null) {
          //no change , no need to persist
          return;
        }
        setData(path, modified, s.getVersion(), true);
        break;
      } else {
        modified = editor.apply(s,null);
        if (modified == null) {
          //no change , no need to persist
          return;
        }
        create(path, modified, CreateMode.PERSISTENT, true);
        break;
      }
    } catch (KeeperException.BadVersionException | KeeperException.NodeExistsException e) {
      continue;
    }
  }


}
 

public void ensureExists(final String path, final byte[] data,
    CreateMode createMode, final SolrZkClient zkClient, int skipPathParts) throws KeeperException, InterruptedException {
  
  if (zkClient.exists(path, true)) {
    return;
  }
  try {
    zkClient.makePath(path, data, createMode, null, true, true, skipPathParts);
  } catch (NodeExistsException e) {
    // it's okay if another beats us creating the node
  }
  
}
 

/**
 * Puts an element in the map only if there isn't one with the same trackingId already
 * @return True if the the element was added. False if it wasn't (because the key already exists)
 */
public boolean putIfAbsent(String trackingId, byte[] data) throws KeeperException, InterruptedException {
  try {
    zookeeper.makePath(dir + "/" + PREFIX + trackingId, data, CreateMode.PERSISTENT, null, true, true);
    return true;
  } catch (NodeExistsException e) {
    return false;
  }
}
 

private void initializeSubsystemRollbackDirectory() throws Exception {
	String parent = String.format("%s/%s/rollback", CONSTANTS_ROOT_PATH, CommonUtils.getApplication(this.endpoint));
	try {
		this.curatorFramework.create() //
				.creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath(parent);
	} catch (NodeExistsException nex) {
		logger.debug("Path exists(path= {})!", parent); // ignore
	}
}
 

private void initializeClusterInstancesDirectory() throws Exception {
	String parent = String.format("%s/%s/instances", CONSTANTS_ROOT_PATH, CommonUtils.getApplication(this.endpoint));
	try {
		this.curatorFramework.create() //
				.creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath(parent);
	} catch (NodeExistsException nex) {
		logger.debug("Path exists(path= {})!", parent); // ignore
	}
}
 

/**
 *
 * @param zk
 * @param path
 *            path of the z-node used to track the generators ids
 * @param prefix
 *            prefix to prepend to the generated id. Having a unique prefix can make the id globally unique
 * @throws Exception
 */
public DistributedIdGenerator(ZooKeeper zk, String path, String prefix) throws Exception {
    this.prefix = prefix;
    this.counter = new AtomicLong(0);

    // Create base path if it doesn't exist
    if (zk.exists(path, false) == null) {
        try {
            ZkUtils.createFullPathOptimistic(zk, path, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.PERSISTENT);
        } catch (NodeExistsException e) {
            // Ok
        }
    }

    // Create an ephemeral sequential z-node that will have a name containing a unique number. We'll use this number
    // as a prefix for all the generated ids, in addition to the specified prefix.
    String createdPath = zk.create(path + "/-", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
            CreateMode.EPHEMERAL_SEQUENTIAL);

    // Parse the sequential z-node name and extract the unique number
    String[] parts = createdPath.split("/");
    String name = parts[parts.length - 1].replace('-', ' ').trim();

    this.generatorInstanceId = Integer.parseInt(name);
    log.info("Created sequential node at {} -- Generator Id is {}-{}", createdPath, prefix, generatorInstanceId);
}
 

@VisibleForTesting
public void init() throws KeeperException, InterruptedException {
    try {
        if (zooKeeper.exists(SchemaPath, false) == null) {
            zooKeeper.create(SchemaPath, new byte[]{}, Acl, CreateMode.PERSISTENT);
        }
    } catch (KeeperException.NodeExistsException error) {
        // race on startup, ignore.
    }
}
 

/**
 * a wrapper for ZkUtils.createFullPathOptimistic but ignore exception of node exists
 */
private static void createZkNode(ZooKeeper zkc, String path,
                                 byte[] data, final List<ACL> acl, final CreateMode createMode)
    throws KeeperException, InterruptedException {

    try {
        ZkUtils.createFullPathOptimistic(zkc, path, data, acl, createMode);
    } catch (NodeExistsException e) {
        // Ignore
    }
}
 

/**
 * Check if the provided <i>path</i> exists or not and wait it expired if possible.
 *
 * @param zk the zookeeper client instance
 * @param path the zookeeper path
 * @param sessionTimeoutMs session timeout in milliseconds
 * @return true if path exists, otherwise return false
 * @throws KeeperException when failed to access zookeeper
 * @throws InterruptedException interrupted when waiting for znode to be expired
 */
public static boolean checkNodeAndWaitExpired(ZooKeeper zk,
                                              String path,
                                              long sessionTimeoutMs) throws KeeperException, InterruptedException {
    final CountDownLatch prevNodeLatch = new CountDownLatch(1);
    Watcher zkPrevNodeWatcher = watchedEvent -> {
        // check for prev node deletion.
        if (EventType.NodeDeleted == watchedEvent.getType()) {
            prevNodeLatch.countDown();
        }
    };
    Stat stat = zk.exists(path, zkPrevNodeWatcher);
    if (null != stat) {
        // if the ephemeral owner isn't current zookeeper client
        // wait for it to be expired
        if (stat.getEphemeralOwner() != zk.getSessionId()) {
            log.info("Previous znode : {} still exists, so waiting {} ms for znode deletion",
                path, sessionTimeoutMs);
            if (!prevNodeLatch.await(sessionTimeoutMs, TimeUnit.MILLISECONDS)) {
                throw new NodeExistsException(path);
            } else {
                return false;
            }
        }
        return true;
    } else {
        return false;
    }
}
 

private void createPersistentPathIfNecessary(String path) throws Exception {
	try {
		this.curatorFramework.create() //
				.creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
	} catch (NodeExistsException nex) {
		logger.debug("Path exists(path= {})!", path);
	}
}
 

private boolean addKey(String key, String value) throws Exception {
    try {
        LOGGER.debug("adding key {} with value {}", key, value);
        if (value == null) {
            curatorFramework.create().forPath(key);
            return true;
        }
        curatorFramework.create().forPath(key, value.getBytes());
        return true;
    } catch (NodeExistsException e) {
        LOGGER.debug("cannot add key {} because it already exists", key);
        return false;
    }
}
 

protected SingularityCreateResult create(String path, Optional<byte[]> data) {
  try {
    privateCreate(path, data);

    return SingularityCreateResult.CREATED;
  } catch (NodeExistsException nee) {
    return SingularityCreateResult.EXISTED;
  } catch (Throwable t) {
    throw new RuntimeException(t);
  }
}
 

protected SingularityCreateResult save(String path, Optional<byte[]> data) {
  try {
    privateCreate(path, data);

    return SingularityCreateResult.CREATED;
  } catch (NodeExistsException nee) {
    return set(path, data);
  } catch (Throwable t) {
    throw new RuntimeException(t);
  }
}
 

@Timed
public QueuedRequestId enqueueRequest(BaragonRequest request, InternalRequestStates state) throws NodeExistsException {
  final long start = System.currentTimeMillis();

  final String queuedRequestPath = String.format(REQUEST_ENQUEUE_FORMAT, request.getLoadBalancerService().getServiceId(), request.getLoadBalancerRequestId());
  final String requestPath = String.format(REQUEST_FORMAT, request.getLoadBalancerRequestId());
  final String requestStatePath = String.format(REQUEST_STATE_FORMAT, request.getLoadBalancerRequestId());

  try {
    if (!nodeExists(REQUESTS_FORMAT)) {
      createNode(REQUESTS_FORMAT);
    }
    if (!nodeExists(REQUEST_QUEUE_FORMAT)) {
      createNode(REQUEST_QUEUE_FORMAT);
    }

    byte[] requestBytes = objectMapper.writeValueAsBytes(request);
    byte[] stateBytes = objectMapper.writeValueAsBytes(state);

    Collection<CuratorTransactionResult> results = curatorFramework.inTransaction()
        .create().forPath(requestPath, requestBytes).and()
        .create().forPath(requestStatePath, stateBytes).and()
        .create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath(queuedRequestPath)
        .and().commit();

    log(OperationType.WRITE, Optional.of(3), Optional.of(requestBytes.length + stateBytes.length), start, String.format("Transaction Paths [%s + %s + %s]", requestPath, requestStatePath, queuedRequestPath));

    return QueuedRequestId.fromString(ZKPaths.getNodeFromPath(Iterables.find(results, CuratorTransactionResult.ofTypeAndPath(org.apache.curator.framework.api.transaction.OperationType.CREATE, queuedRequestPath))
        .getResultPath()));
  } catch (NodeExistsException nee) {
    throw nee;
  } catch (Exception e) {
    throw new RuntimeException(e);
  }
}
 

public BaragonResponse enqueueRequest(BaragonRequest request) throws RequestAlreadyEnqueuedException, InvalidRequestActionException, InvalidUpstreamsException {
  final Optional<BaragonResponse> maybePreexistingResponse = getResponse(request.getLoadBalancerService().getServiceId(), request.getLoadBalancerRequestId());

  if (maybePreexistingResponse.isPresent()) {
    Optional<BaragonRequest> maybePreexistingRequest = requestDatastore.getRequest(request.getLoadBalancerRequestId());
    if (maybePreexistingRequest.isPresent() && !maybePreexistingRequest.get().equals(request)) {
      throw new RequestAlreadyEnqueuedException(request.getLoadBalancerRequestId(), maybePreexistingResponse.get(), String.format("Request %s is already enqueued with different parameters", request.getLoadBalancerRequestId()));
    } else {
      return maybePreexistingResponse.get();
    }
  }

  if (request.isNoDuplicateUpstreams()) {
    validateNoDuplicateUpstreams(request);
  }

  if (request.isNoReload() && request.getAction().isPresent() && request.getAction().get().equals(RequestAction.RELOAD)) {
    throw new InvalidRequestActionException("You can not specify 'noReload' on a request with action 'RELOAD'");
  }

  if (!request.getReplaceUpstreams().isEmpty() && (!request.getAddUpstreams().isEmpty() || !request.getRemoveUpstreams().isEmpty())) {
    throw new InvalidUpstreamsException("If overrideUpstreams is specified, addUpstreams and removeUpstreams mustbe empty");
  }

  if (request.getAction().isPresent() && request.getAction().equals(Optional.of(RequestAction.REVERT))) {
    throw new InvalidRequestActionException("The REVERT action may only be used internally by Baragon, you may specify UPDATE, DELETE, RELOAD, or leave the action blank(UPDATE)");
  }

  try {
    final QueuedRequestId queuedRequestId = requestDatastore.enqueueRequest(request, InternalRequestStates.PENDING);

    requestDatastore.setRequestMessage(request.getLoadBalancerRequestId(), String.format("Queued as %s", queuedRequestId));
  } catch (NodeExistsException nee) {
    LOG.warn("Tried to write request {}, but already existed, returning current contents", request.getLoadBalancerRequestId());
  }

  return getResponse(request.getLoadBalancerService().getServiceId(), request.getLoadBalancerRequestId()).get();
}
 

private static void createPath(ZooKeeper zk, String path) throws KeeperException, InterruptedException {
    if ("/".equals(path)) {
        return;
    }
    if (zk.exists(path, false) == null) {
        createPath(zk, getParentPath(path));
        LOGGER.info("Creating path: {}", path);
        try {
            zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        } catch (NodeExistsException e) {
            // Race condition
        }
    }
}
 

private void restoreNode(ZooKeeper zk, BackupZNode zNode) throws KeeperException, InterruptedException {
    createPath(zk, getParentPath(zNode.path));
    try {
        zk.create(zNode.path, zNode.data, zNode.acls, CreateMode.PERSISTENT);
        LOGGER.info("Created node: {}", zNode.path);
    } catch (NodeExistsException e) {
        if (options.overwriteExisting) {
            // TODO: Compare with current data / acls
            zk.setACL(zNode.path, zNode.acls, -1);
            zk.setData(zNode.path, zNode.data, -1);
        } else {
            LOGGER.warn("Node already exists: {}", zNode.path);
        }
    }
}
 

@Override
public boolean try_to_be_leader(String path, String host, RunnableCallback callback) throws Exception {
    if (callback != null)
        this.master_callback.set(callback);
    try {
        cluster_state.tryToBeLeader(path, host.getBytes());
    } catch (NodeExistsException e) {
        cluster_state.node_existed(path, true);
        LOG.info("leader is alive");
        return false;
    }
    return true;
}
 

private static boolean isIgnoredException(final Throwable cause) {
    return cause instanceof ConnectionLossException || cause instanceof NoNodeException || cause instanceof NodeExistsException;
}
 

private static boolean isIgnoredException(final Throwable cause) {
	return cause instanceof NoNodeException || cause instanceof NodeExistsException;
}
 

private static boolean isIgnoredException(final Throwable cause) {
    return cause instanceof ConnectionLossException || cause instanceof NoNodeException || cause instanceof NodeExistsException;
}
 

@Test
public void assertHandleExceptionWithNoNodeExistsException() {
    RegExceptionHandler.handleException(new NodeExistsException());
}
 

@Test
public void assertHandleExceptionWithCausedNoNodeExistsException() {
    RegExceptionHandler.handleException(new RuntimeException(new NodeExistsException()));
}