@Override
public boolean replace(final StateMap oldValue, final Map<String, String> newValue, final String componentId) throws IOException {
    verifyEnabled();

    try {
        setState(newValue, (int) oldValue.getVersion(), componentId, false);
        return true;
    } catch (final NoNodeException nne) {
        return false;
    } catch (final IOException ioe) {
        final Throwable cause = ioe.getCause();
        if (cause != null && cause instanceof KeeperException) {
            final KeeperException ke = (KeeperException) cause;
            if (Code.BADVERSION == ke.code()) {
                return false;
            }
        }

        throw ioe;
    }
}
 

public List<OffsetInfo> processTopic(String group, String topic) throws Exception {
	List<String> partitionIds = null;
	try {
		partitionIds = JavaConversions.seqAsJavaList(ZKUtils.getZKUtilsFromKafka()
				.getChildren(ZkUtils.BrokerTopicsPath() + "/" + topic + "/partitions"));
	} catch (Exception e) {
		if (e instanceof NoNodeException) {
			LOG.warn("Is topic >" + topic + "< exists!", e);
			return null;
		}
	}
	List<OffsetInfo> offsetInfos = new ArrayList<OffsetInfo>();
	OffsetInfo offsetInfo = null;
	if (partitionIds == null) {
		// TODO that topic exists in consumer node but not in topics node?!
		return null;
	}

	for (String partitionId : partitionIds) {
		offsetInfo = processPartition(group, topic, partitionId);
		if (offsetInfo != null) {
			offsetInfos.add(offsetInfo);
		}
	}
	return offsetInfos;
}
 

@BeforeMethod
public void beforeMethod() throws Exception {

    tsoPortForTest = TestUtils.getFreeLocalPort();

    int zkPortForTest = TestUtils.getFreeLocalPort();
    zkClusterForTest = TSO_HOST + ":" + zkPortForTest;
    LOG.info("Starting ZK Server in port {}", zkPortForTest);
    zkServer = TestUtils.provideTestingZKServer(zkPortForTest);
    LOG.info("ZK Server Started @ {}", zkServer.getConnectString());

    zkClient = TestUtils.provideConnectedZKClient(zkClusterForTest);

    Stat stat;
    try {
        zkClient.delete().forPath(CURRENT_TSO_PATH);
        stat = zkClient.checkExists().forPath(CURRENT_TSO_PATH);
        assertNull(stat, CURRENT_TSO_PATH + " should not exist");
    } catch (NoNodeException e) {
        LOG.info("{} ZNode did not exist", CURRENT_TSO_PATH);
    }

}
 

/**
 * Get current {@link AutoScalingConfig}.
 *
 * @param watcher optional {@link Watcher} to set on a znode to watch for config changes.
 * @return current configuration from <code>autoscaling.json</code>. NOTE:
 * this data is retrieved from ZK on each call.
 */
@SuppressWarnings({"unchecked"})
public AutoScalingConfig getAutoScalingConfig(Watcher watcher) throws KeeperException, InterruptedException {
  Stat stat = new Stat();

  Map<String, Object> map = new HashMap<>();
  try {
    byte[] bytes = zkClient.getData(SOLR_AUTOSCALING_CONF_PATH, watcher, stat, true);
    if (bytes != null && bytes.length > 0) {
      map = (Map<String, Object>) fromJSON(bytes);
    }
  } catch (KeeperException.NoNodeException e) {
    // ignore
  }
  map.put(AutoScalingParams.ZK_VERSION, stat.getVersion());
  return new AutoScalingConfig(map);
}
 

private DocCollection fetchCollectionState(String coll, Watcher watcher) throws KeeperException, InterruptedException {
  String collectionPath = getCollectionPath(coll);
  while (true) {
    try {
      Stat stat = new Stat();
      byte[] data = zkClient.getData(collectionPath, watcher, stat, true);
      ClusterState state = ClusterState.createFromJson(stat.getVersion(), data, Collections.emptySet());
      ClusterState.CollectionRef collectionRef = state.getCollectionStates().get(coll);
      return collectionRef == null ? null : collectionRef.get();
    } catch (KeeperException.NoNodeException e) {
      if (watcher != null) {
        // Leave an exists watch in place in case a state.json is created later.
        Stat exists = zkClient.exists(collectionPath, watcher, true);
        if (exists != null) {
          // Rare race condition, we tried to fetch the data and couldn't find it, then we found it exists.
          // Loop and try again.
          continue;
        }
      }
      return null;
    }
  }
}
 

public void removeEphemeralLiveNode() throws KeeperException, InterruptedException {
  if (zkRunOnly) {
    return;
  }
  String nodeName = getNodeName();
  String nodePath = ZkStateReader.LIVE_NODES_ZKNODE + "/" + nodeName;
  String nodeAddedPath = ZkStateReader.SOLR_AUTOSCALING_NODE_ADDED_PATH + "/" + nodeName;
  log.info("Remove node as live in ZooKeeper:{}", nodePath);
  List<Op> ops = new ArrayList<>(2);
  ops.add(Op.delete(nodePath, -1));
  ops.add(Op.delete(nodeAddedPath, -1));

  try {
    zkClient.multi(ops, true);
  } catch (NoNodeException e) {

  }
}
 

public void checkOverseerDesignate() {
  try {
    byte[] data = zkClient.getData(ZkStateReader.ROLES, null, new Stat(), true);
    if (data == null) return;
    @SuppressWarnings({"rawtypes"})
    Map roles = (Map) Utils.fromJSON(data);
    if (roles == null) return;
    @SuppressWarnings({"rawtypes"})
    List nodeList = (List) roles.get("overseer");
    if (nodeList == null) return;
    if (nodeList.contains(getNodeName())) {
      ZkNodeProps props = new ZkNodeProps(Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.ADDROLE.toString().toLowerCase(Locale.ROOT),
          "node", getNodeName(),
          "role", "overseer");
      log.info("Going to add role {} ", props);
      getOverseerCollectionQueue().offer(Utils.toJSON(props));
    }
  } catch (NoNodeException nne) {
    return;
  } catch (Exception e) {
    log.warn("could not read the overseer designate ", e);
  }
}
 

private String getLeaderUrl(final String collection, final String slice)
    throws KeeperException, InterruptedException {
  int iterCount = 60;
  while (iterCount-- > 0) {
    try {
      byte[] data = zkClient.getData(
          ZkStateReader.getShardLeadersPath(collection, slice), null, null,
          true);
      ZkCoreNodeProps leaderProps = new ZkCoreNodeProps(
          ZkNodeProps.load(data));
      return leaderProps.getCoreUrl();
    } catch (NoNodeException | SessionExpiredException e) {
      Thread.sleep(500);
    }
  }
  zkClient.printLayoutToStream(System.out);
  throw new RuntimeException("Could not get leader props for " + collection + " " + slice);
}
 

@Override
public byte[] getData(String path, Stat stat, Watcher watcher) throws KeeperException.NoNodeException {
  // we know we are doing a copy when we are getting data from the base config set and
  // the new config set (partially) exists
  String zkAddress = zkTestServer.getZkAddress();
  String chroot = zkAddress.substring(zkAddress.lastIndexOf("/"));
  if (path.startsWith(chroot + CONFIGS_ZKNODE + "/" + BASE_CONFIGSET_NAME)
      && !path.contains(ConfigSetProperties.DEFAULT_FILENAME)) {
    List<String> children = null;
    try {
      children = getChildren(chroot + CONFIGS_ZKNODE + "/" + CONFIGSET_NAME, null, null);
    } catch (KeeperException.NoNodeException e) {}
    if (children != null && children.size() > 0) {
      throw new RuntimeException("sample zookeeper error");
    }
  }
  return super.getData(path, stat, watcher);
}
 

private void updateGcTimestamp() throws Exception {
  List<String> children;
  try {
    children = curator.getChildren().forPath(ZookeeperPath.TRANSACTOR_TIMESTAMPS);
  } catch (NoNodeException nne) {
    children = Collections.emptyList();
  }

  long oldestTs = Long.MAX_VALUE;
  boolean nodeFound = false;

  for (String child : children) {
    Long ts = LongUtil.fromByteArray(
        curator.getData().forPath(ZookeeperPath.TRANSACTOR_TIMESTAMPS + "/" + child));
    nodeFound = true;
    if (ts < oldestTs) {
      oldestTs = ts;
    }
  }

  if (nodeFound) {
    updateAdvertisedGcTimestamp(oldestTs);
  } else {
    updateAdvertisedGcTimestamp(currentTs);
  }
}
 

/**
 * cleanupNamespaceNodes is only called when the NamespaceService is initialized. So, no need for synchronization.
 *
 * @throws Exception
 */
private static final void cleanupNamespaceNodes(ZooKeeper zkc, String root, String selfBrokerUrl) throws Exception {
    // we don't need a watch here since we are only cleaning up the stale ephemeral nodes from previous session
    try {
        for (String node : zkc.getChildren(root, false)) {
            String currentPath = root + "/" + node;
            // retrieve the content and try to decode with ServiceLookupData
            List<String> children = zkc.getChildren(currentPath, false);
            if (children.size() == 0) {
                // clean up a single namespace node
                cleanupSingleNamespaceNode(zkc, currentPath, selfBrokerUrl);
            } else {
                // this is an intermediate node, which means this is v2 namespace path
                cleanupNamespaceNodes(zkc, currentPath, selfBrokerUrl);
            }
        }
    } catch (NoNodeException nne) {
        LOG.info("No children for [{}]", nne.getPath());
    }
}
 

public <T> CompletableFuture<Optional<Entry<T, Stat>>> getEntryAsync(final String path, final Deserializer<T> deserializer) {
    CompletableFuture<Optional<Entry<T, Stat>>> future = new CompletableFuture<>();
    getDataAsync(path, this, deserializer)
        .thenAccept(future::complete)
        .exceptionally(ex -> {
            asyncInvalidate(path);
            if (ex.getCause() instanceof NoNodeException) {
                future.complete(Optional.empty());
            } else {
                future.completeExceptionally(ex.getCause());
            }

            return null;
        });
    return future;
}
 

public <T> CompletableFuture<Optional<T>> getDataAsync(final String path, final Deserializer<T> deserializer) {
    CompletableFuture<Optional<T>> future = new CompletableFuture<>();
    getDataAsync(path, this, deserializer).thenAccept(data -> {
        future.complete(data.map(e -> e.getKey()));
    }).exceptionally(ex -> {
        asyncInvalidate(path);
        if (ex.getCause() instanceof NoNodeException) {
            future.complete(Optional.empty());
        } else {
            future.completeExceptionally(ex.getCause());
        }

        return null;
    });
    return future;
}
 

/**
 * Creates the specified node and all parent nodes required for it to exist.  The creation of
 * parent znodes is not atomic with the leafe znode creation but the data is written atomically
 * when the leaf node is created.
 *
 * No watches are set and no errors are thrown if the node already exists.
 *
 * The nodes created are persistent and open access.
 *
 * @param zkw zk reference
 * @param znode path of node
 * @throws KeeperException if unexpected zookeeper exception
 */
public static void createWithParents(ZKWatcher zkw, String znode, byte[] data)
  throws KeeperException {
  try {
    if(znode == null) {
      return;
    }
    zkw.getRecoverableZooKeeper().create(znode, data, createACL(zkw, znode),
        CreateMode.PERSISTENT);
  } catch(KeeperException.NodeExistsException nee) {
    return;
  } catch(KeeperException.NoNodeException nne) {
    createWithParents(zkw, getParent(znode));
    createWithParents(zkw, znode, data);
  } catch(InterruptedException ie) {
    zkw.interruptedException(ie);
  }
}
 

@Test
public void testPrimaryCrash() throws Exception {
  createEdits(20);
  fs.close();
  InjectionHandler.set(new TestAvatarSyncLastTxidInjectionHandler());
  cluster.getPrimaryAvatar(0).avatar.shutdownAvatar();
  InjectionHandler.clear();
  ZookeeperTxId lastTxId = null;
  try {
    lastTxId = getLastTxid();
  } catch (NoNodeException e) {
    LOG.info("Expected exception", e);
    assertNull(lastTxId);
    return;
  }
  fail("Did not throw : " + NoNodeException.class);
}
 

@Test
public void testDoubleFailoverWithPrimaryCrash() throws Exception {
  // First failover
  createEdits(20);
  cluster.failOver();
  cluster.restartStandby();
  createEdits(20);
  fs.close();

  // Second failover.
  InjectionHandler.set(new TestAvatarSyncLastTxidInjectionHandler());
  cluster.killPrimary(0, true);
  InjectionHandler.clear();
  
  try {
    cluster.failOver();
  } catch (Exception e) {
    LOG.info("Expected exception : ", e);
    return;
  }
  fail("Did not throw : " + NoNodeException.class);
}
 

@Override
public boolean replace(final StateMap oldValue, final Map<String, String> newValue, final String componentId) throws IOException {
    verifyEnabled();

    try {
        setState(newValue, (int) oldValue.getVersion(), componentId, false);
        return true;
    } catch (final NoNodeException nne) {
        return false;
    } catch (final IOException ioe) {
        final Throwable cause = ioe.getCause();
        if (cause != null && cause instanceof KeeperException) {
            final KeeperException ke = (KeeperException) cause;
            if (Code.BADVERSION == ke.code()) {
                return false;
            }
        }

        throw ioe;
    }
}
 

protected List<String> getChildren(String root) {
  LOG.trace("Preparing to call getChildren() on {}", root);
  final long start = System.currentTimeMillis();
  int numChildren = 0;

  try {
    final List<String> children = curator.getChildren().forPath(root);
    numChildren = children.size();

    return children;
  } catch (NoNodeException nne) {
    return Collections.emptyList();
  } catch (Throwable t) {
    throw new RuntimeException(t);
  } finally {
    log(
      OperationType.GET_CHILDREN,
      Optional.of(numChildren),
      Optional.empty(),
      start,
      root
    );
  }
}
 

private List<SingularityHostState> getHostStates() {
  List<String> children = getChildren(ROOT_PATH);
  List<SingularityHostState> states = Lists.newArrayListWithCapacity(children.size());

  for (String child : children) {
    try {
      byte[] bytes = curator.getData().forPath(ZKPaths.makePath(ROOT_PATH, child));

      states.add(hostStateTranscoder.fromBytes(bytes));
    } catch (NoNodeException nne) {} catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  return states;
}
 

private void setState(final Map<String, String> stateValues, final int version, final String componentId) throws IOException {
    try {
        setState(stateValues, version, componentId, true);
    } catch (final NoNodeException nne) {
        // should never happen because we are passing 'true' for allowNodeCreation
        throw new IOException("Unable to create Node in ZooKeeper to set state for component with ID " + componentId, nne);
    }
}
 

@Test
public void testGuaranteedDeleteOnNonExistentNodeInForeground() throws Exception
{
    CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
    client.start();
    
    final AtomicBoolean pathAdded = new AtomicBoolean(false);
    
    ((CuratorFrameworkImpl)client).getFailedDeleteManager().debugListener = new FailedDeleteManagerListener()
    {
        
        @Override
        public void pathAddedForDelete(String path)
        {
            pathAdded.set(true);
        }
    };
    
    try
    {
        client.delete().guaranteed().forPath("/nonexistent");
        Assert.fail();
    }
    catch(NoNodeException e)
    {
        //Exception is expected, the delete should not be retried
        Assert.assertFalse(pathAdded.get());
    }
    finally
    {
        client.close();
    }        
}
 

/**
 * Expire the ZK session of the given service. This requires
 * (and asserts) that the given service be the current active.
 * @throws NoNodeException if no service holds the lock
 */
public void expireActiveLockHolder(int idx)
    throws NoNodeException {
  Stat stat = new Stat();
  byte[] data = zks.getZKDatabase().getData(
      DummyZKFC.LOCK_ZNODE, stat, null);
  
  assertArrayEquals(Ints.toByteArray(svcs[idx].index), data);
  long session = stat.getEphemeralOwner();
  LOG.info("Expiring svc " + idx + "'s zookeeper session " + session);
  zks.closeSession(session);
}
 

public static void waitForActiveLockData(TestContext ctx,
    ZooKeeperServer zks, String parentDir, byte[] activeData)
    throws Exception {
  long st = Time.now();
  long lastPrint = st;
  while (true) {
    if (ctx != null) {
      ctx.checkException();
    }
    try {
      Stat stat = new Stat();
      byte[] data = zks.getZKDatabase().getData(
        parentDir + "/" +
        ActiveStandbyElector.LOCK_FILENAME, stat, null);
      if (activeData != null &&
          Arrays.equals(activeData, data)) {
        return;
      }
      if (Time.now() > lastPrint + LOG_INTERVAL_MS) {
        LOG.info("Cur data: " + StringUtils.byteToHexString(data));
        lastPrint = Time.now();
      }
    } catch (NoNodeException nne) {
      if (activeData == null) {
        return;
      }
      if (Time.now() > lastPrint + LOG_INTERVAL_MS) {
        LOG.info("Cur data: no node");
        lastPrint = Time.now();
      }
    }
    Thread.sleep(50);
  }
}
 

/**
 * Expire the ZK session of the given service. This requires
 * (and asserts) that the given service be the current active.
 * @throws NoNodeException if no service holds the lock
 */
public void expireActiveLockHolder(int idx)
    throws NoNodeException {
  Stat stat = new Stat();
  byte[] data = zks.getZKDatabase().getData(
      DummyZKFC.LOCK_ZNODE, stat, null);
  
  assertArrayEquals(Ints.toByteArray(svcs[idx].index), data);
  long session = stat.getEphemeralOwner();
  LOG.info("Expiring svc " + idx + "'s zookeeper session " + session);
  zks.closeSession(session);
}
 

public static void waitForActiveLockData(TestContext ctx,
    ZooKeeperServer zks, String parentDir, byte[] activeData)
    throws Exception {
  long st = Time.now();
  long lastPrint = st;
  while (true) {
    if (ctx != null) {
      ctx.checkException();
    }
    try {
      Stat stat = new Stat();
      byte[] data = zks.getZKDatabase().getData(
        parentDir + "/" +
        ActiveStandbyElector.LOCK_FILENAME, stat, null);
      if (activeData != null &&
          Arrays.equals(activeData, data)) {
        return;
      }
      if (Time.now() > lastPrint + LOG_INTERVAL_MS) {
        LOG.info("Cur data: " + StringUtils.byteToHexString(data));
        lastPrint = Time.now();
      }
    } catch (NoNodeException nne) {
      if (activeData == null) {
        return;
      }
      if (Time.now() > lastPrint + LOG_INTERVAL_MS) {
        LOG.info("Cur data: no node");
        lastPrint = Time.now();
      }
    }
    Thread.sleep(50);
  }
}
 

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

/**
 * Returns config set name for collection.
 * TODO move to DocCollection (state.json).
 *
 * @param collection to return config set name for
 */
public String readConfigName(String collection) throws KeeperException {

  String configName = null;

  String path = COLLECTIONS_ZKNODE + "/" + collection;
  log.debug("Loading collection config from: [{}]", path);

  try {
    byte[] data = zkClient.getData(path, null, null, true);
    if (data == null) {
      log.warn("No config data found at path {}.", path);
      throw new KeeperException.NoNodeException("No config data found at path: " + path);
    }

    ZkNodeProps props = ZkNodeProps.load(data);
    configName = props.getStr(CONFIGNAME_PROP);

    if (configName == null) {
      log.warn("No config data found at path{}. ", path);
      throw new KeeperException.NoNodeException("No config data found at path: " + path);
    }
  } catch (InterruptedException e) {
    SolrZkClient.checkInterrupted(e);
    log.warn("Thread interrupted when loading config name for collection {}", collection);
    throw new SolrException(ErrorCode.SERVER_ERROR, "Thread interrupted when loading config name for collection " + collection, e);
  }

  return configName;
}
 

@SuppressWarnings({"unchecked"})
public synchronized void createClusterStateWatchersAndUpdate() throws KeeperException, InterruptedException {
  // We need to fetch the current cluster state and the set of live nodes

  log.debug("Updating cluster state from ZooKeeper... ");

  try {
    // on reconnect of SolrZkClient force refresh and re-add watches.
    loadClusterProperties();
    refreshLiveNodes(new LiveNodeWatcher());
    refreshCollections();
    refreshCollectionList(new CollectionsChildWatcher());
    refreshAliases(aliasesManager);

    if (securityNodeListener != null) {
      addSecurityNodeWatcher(pair -> {
        ConfigData cd = new ConfigData();
        cd.data = pair.first() == null || pair.first().length == 0 ? EMPTY_MAP : Utils.getDeepCopy((Map) fromJSON(pair.first()), 4, false);
        cd.version = pair.second() == null ? -1 : pair.second().getVersion();
        securityData = cd;
        securityNodeListener.run();
      });
      securityData = getSecurityProps(true);
    }

    collectionPropsObservers.forEach((k, v) -> {
      collectionPropsWatchers.computeIfAbsent(k, PropsWatcher::new).refreshAndWatch(true);
    });
  } catch (KeeperException.NoNodeException nne) {
    throw new SolrException(ErrorCode.SERVICE_UNAVAILABLE,
        "Cannot connect to cluster at " + zkClient.getZkServerAddress() + ": cluster not found/not ready");
  }
}
 

/**
 * Search for any lazy-loadable collections.
 */
private void refreshCollectionList(Watcher watcher) throws KeeperException, InterruptedException {
  synchronized (refreshCollectionListLock) {
    List<String> children = null;
    try {
      children = zkClient.getChildren(COLLECTIONS_ZKNODE, watcher, true);
    } catch (KeeperException.NoNodeException e) {
      log.warn("Error fetching collection names: [{}]", e.getMessage());
      // fall through
    }
    if (children == null || children.isEmpty()) {
      lazyCollectionStates.clear();
      return;
    }

    // Don't lock getUpdateLock() here, we don't need it and it would cause deadlock.
    // Don't mess with watchedCollections, they should self-manage.

    // First, drop any children that disappeared.
    this.lazyCollectionStates.keySet().retainAll(children);
    for (String coll : children) {
      // We will create an eager collection for any interesting collections, so don't add to lazy.
      if (!collectionWatches.containsKey(coll)) {
        // Double check contains just to avoid allocating an object.
        LazyCollectionRef existing = lazyCollectionStates.get(coll);
        if (existing == null) {
          lazyCollectionStates.putIfAbsent(coll, new LazyCollectionRef(coll));
        }
      }
    }
  }
}
 

@Override
public void applyMigration() {
  try {
    try {
      // Data format has changed and usage will repopulate when the poller runs
      curator.delete().deletingChildrenIfNeeded().forPath("/usage/slaves");
      curator.delete().deletingChildrenIfNeeded().forPath("/usage/tasks");
    } catch (NoNodeException nee) {}
  } catch (Exception e) {
    throw new RuntimeException(e);
  }
}