@Test
public void testCanReadFromZookeeper() throws Exception {

    CuratorFramework curatorFramework = mock(CuratorFramework.class);
    ExistsBuilder existsBuilder = mock(ExistsBuilder.class);
    GetDataBuilder getDataBuilder = mock(GetDataBuilder.class);
    GetChildrenBuilder getChildrenBuilder = mock(GetChildrenBuilder.class);

    when(getDataBuilder.forPath(ConfigurationType.GLOBAL.getZookeeperRoot())).thenReturn(mockGlobalData());
    when(curatorFramework.checkExists()).thenReturn(existsBuilder);
    when(curatorFramework.getData()).thenReturn(getDataBuilder);
    when(curatorFramework.getChildren()).thenReturn(getChildrenBuilder);
    when(getChildrenBuilder.forPath(any())).thenReturn(Collections.emptyList());

    Configuration configuration = new Configuration(Paths.get("foo"));
    configuration.curatorFramework = curatorFramework;
    configuration.update();

    checkResult(configuration);
}
 

@Override
public List<PropertyItem> findProperties(String node) {
	LOGGER.debug("Find properties in node: [{}]", node);
	List<PropertyItem> properties = Lists.newArrayList();
	try {
		Stat stat = client.checkExists().forPath(node);
		if (stat != null) {
			GetChildrenBuilder childrenBuilder = client.getChildren();
			List<String> children = childrenBuilder.forPath(node);
			GetDataBuilder dataBuilder = client.getData();
			if (children != null) {
				for (String child : children) {
					String propPath = ZKPaths.makePath(node, child);
					PropertyItem item = new PropertyItem(child, new String(dataBuilder.forPath(propPath), Charsets.UTF_8));
					properties.add(item);
				}
			}
		}
	} catch (Exception e) {
		LOGGER.error(e.getMessage(), e);
	}
	return properties;
}
 

/**
 * 加载节点并监听节点变化
 */
void loadNode() {
    final String nodePath = ZKPaths.makePath(configProfile.getVersionedRootNode(), node);

    final GetChildrenBuilder childrenBuilder = client.getChildren();

    try {
        final List<String> children = childrenBuilder.watched().forPath(nodePath);
        if (children != null) {
            final Map<String, String> configs = new HashMap<>();
            for (String child : children) {
                final Tuple<String, String> keyValue = loadKey(ZKPaths.makePath(nodePath, child));
                if (keyValue != null) {
                    configs.put(keyValue.getFirst(), keyValue.getSecond());
                }
            }
            cleanAndPutAll(configs);
        }
    } catch (Exception e) {
        throw new RuntimeException(e);
    }

    if (getConfigLocalCache() != null) {
        getConfigLocalCache().saveLocalCache(this, getNode());
    }
}
 

/**
 * Tests that if we attempt to delete a node that doesnt actually exist
 * just silently returns.
 *
 * To simulate a race condition we do this using mocks.
 */
@Test
public void testDeleteNodeIfNoChildren_withNodeThatDoesntExist() throws Exception {
    final String basePath = "/testDeleteNodeIfNoChildren_withNodeThatDoesntExist";

    final CuratorFramework mockCurator = mock(CuratorFramework.class);

    // Exists builder should return true saying our basePath exists.
    final ExistsBuilder mockExistsBuilder = mock(ExistsBuilder.class);
    when(mockExistsBuilder.forPath(eq(basePath))).thenReturn(new Stat());
    when(mockCurator.checkExists()).thenReturn(mockExistsBuilder);

    // When we look for children, make sure it returns an empty list.
    final GetChildrenBuilder mockGetChildrenBuilder = mock(GetChildrenBuilder.class);
    when(mockGetChildrenBuilder.forPath(eq(basePath))).thenReturn(new ArrayList<>());
    when(mockCurator.getChildren()).thenReturn(mockGetChildrenBuilder);

    // When we go to delete the actual node, we toss a no-node exception.
    // This effectively simulates a race condition between checking if the node exists (our mock above says yes)
    // and it being removed before we call delete on it.
    final DeleteBuilder mockDeleteBuilder = mock(DeleteBuilder.class);
    when(mockDeleteBuilder.forPath(eq(basePath))).thenThrow(new KeeperException.NoNodeException());
    when(mockCurator.delete()).thenReturn(mockDeleteBuilder);

    // Now create our helper
    final CuratorHelper curatorHelper = new CuratorHelper(mockCurator, new HashMap<>());

    // Call our method
    curatorHelper.deleteNodeIfNoChildren(basePath);
}
 

/**
 * List all children of a path
 * @param path path of operation
 * @return a possibly empty list of children
 * @throws IOException
 */
public List<String> zkList(String path) throws IOException {
  checkServiceLive();
  String fullpath = createFullPath(path);
  try {
    if (LOG.isDebugEnabled()) {
      LOG.debug("ls {}", fullpath);
    }
    GetChildrenBuilder builder = curator.getChildren();
    List<String> children = builder.forPath(fullpath);
    return children;
  } catch (Exception e) {
    throw operationFailure(path, "ls()", e);
  }
}
 

/**
 * List all children of a path
 * @param path path of operation
 * @return a possibly empty list of children
 * @throws IOException
 */
public List<String> zkList(String path) throws IOException {
  checkServiceLive();
  String fullpath = createFullPath(path);
  try {
    if (LOG.isDebugEnabled()) {
      LOG.debug("ls {}", fullpath);
    }
    GetChildrenBuilder builder = curator.getChildren();
    List<String> children = builder.forPath(fullpath);
    return children;
  } catch (Exception e) {
    throw operationFailure(path, "ls()", e);
  }
}
 

@Override
protected List<String> query(final boolean setListener) throws NakadiRuntimeException {
    final GetChildrenBuilder builder = curatorFramework.getChildren();
    if (setListener) {
        builder.usingWatcher(this);
    }
    try {
        return builder.forPath(key);
    } catch (final Exception ex) {
        throw new NakadiRuntimeException(ex);
    }
}
 

@Test
public void defaultSpringApplicationNameWorks() {
	CuratorFramework curator = mock(CuratorFramework.class);
	when(curator.getChildren()).thenReturn(mock(GetChildrenBuilder.class));
	ZookeeperPropertySourceLocator locator = new ZookeeperPropertySourceLocator(
			curator, new ZookeeperConfigProperties());
	locator.locate(new MockEnvironment());
}
 

@Override
public List<String> listChildren(String node) {
	LOGGER.debug("Find children of node: [{}]", node);
	List<String> children = null;
	try {
		Stat stat = client.checkExists().forPath(node);
		if (stat != null) {
			GetChildrenBuilder childrenBuilder = client.getChildren();
			children = childrenBuilder.forPath(node);
		}
	} catch (Exception e) {
		LOGGER.error(e.getMessage(), e);
	}
	return children;
}
 

@Override
public CuratorFramework newClient(final String connectString, final int sessionTimeoutMs,
                                  final int connectionTimeoutMs, final RetryPolicy retryPolicy,
                                  final ACLProvider aclProvider,
                                  final List<AuthInfo> authorization) {
  final CuratorFramework curator = mock(CuratorFramework.class);

  final RetryLoop retryLoop = mock(RetryLoop.class);
  when(retryLoop.shouldContinue()).thenReturn(false);

  final CuratorZookeeperClient czkClient = mock(CuratorZookeeperClient.class);
  when(czkClient.newRetryLoop()).thenReturn(retryLoop);

  when(curator.getZookeeperClient()).thenReturn(czkClient);

  @SuppressWarnings("unchecked") final Listenable<ConnectionStateListener> mockListener =
      (Listenable<ConnectionStateListener>) mock(Listenable.class);

  when(curator.getConnectionStateListenable()).thenReturn(mockListener);

  final GetChildrenBuilder builder = mock(GetChildrenBuilder.class);
  when(curator.getChildren()).thenReturn(builder);

  try {
    when(builder.forPath(anyString())).thenThrow(
        new KeeperException.ConnectionLossException());
  } catch (Exception ignored) {
    // never throws
  }
  when(curator.newNamespaceAwareEnsurePath(anyString())).thenReturn(mock(EnsurePath.class));

  return curator;
}
 

/**
 * Recursively expand the path into the supplied string builder, increasing
 * the indentation by {@link #INDENT} as it proceeds (depth first) down
 * the tree
 * @param builder string build to append to
 * @param path path to examine
 * @param indent current indentation
 */
private void expand(StringBuilder builder,
    String path,
    int indent) {
  try {
    GetChildrenBuilder childrenBuilder = curator.getChildren();
    List<String> children = childrenBuilder.forPath(path);
    for (String child : children) {
      String childPath = path + "/" + child;
      String body;
      Stat stat = curator.checkExists().forPath(childPath);
      StringBuilder bodyBuilder = new StringBuilder(256);
      bodyBuilder.append("  [")
                        .append(stat.getDataLength())
                        .append("]");
      if (stat.getEphemeralOwner() > 0) {
        bodyBuilder.append("*");
      }
      if (verbose) {
        // verbose: extract ACLs
        builder.append(" -- ");
        List<ACL> acls =
            curator.getACL().forPath(childPath);
        for (ACL acl : acls) {
          builder.append(RegistrySecurity.aclToString(acl));
          builder.append(" ");
        }
      }
      body = bodyBuilder.toString();
      // print each child
      append(builder, indent, ' ');
      builder.append('/').append(child);
      builder.append(body);
      builder.append('\n');
      // recurse
      expand(builder, childPath, indent + INDENT);
    }
  } catch (Exception e) {
    builder.append(e.toString()).append("\n");
  }
}
 

/**
 * Recursively expand the path into the supplied string builder, increasing
 * the indentation by {@link #INDENT} as it proceeds (depth first) down
 * the tree
 * @param builder string build to append to
 * @param path path to examine
 * @param indent current indentation
 */
private void expand(StringBuilder builder,
    String path,
    int indent) {
  try {
    GetChildrenBuilder childrenBuilder = curator.getChildren();
    List<String> children = childrenBuilder.forPath(path);
    for (String child : children) {
      String childPath = path + "/" + child;
      String body;
      Stat stat = curator.checkExists().forPath(childPath);
      StringBuilder bodyBuilder = new StringBuilder(256);
      bodyBuilder.append("  [")
                        .append(stat.getDataLength())
                        .append("]");
      if (stat.getEphemeralOwner() > 0) {
        bodyBuilder.append("*");
      }
      if (verbose) {
        // verbose: extract ACLs
        builder.append(" -- ");
        List<ACL> acls =
            curator.getACL().forPath(childPath);
        for (ACL acl : acls) {
          builder.append(RegistrySecurity.aclToString(acl));
          builder.append(" ");
        }
      }
      body = bodyBuilder.toString();
      // print each child
      append(builder, indent, ' ');
      builder.append('/').append(child);
      builder.append(body);
      builder.append('\n');
      // recurse
      expand(builder, childPath, indent + INDENT);
    }
  } catch (Exception e) {
    builder.append(e.toString()).append("\n");
  }
}
 

@Override
public GetChildrenBuilder getChildren() {
    return new MockGetChildrenBuilder();
}
 

private void updateBootstrapServers(final boolean createWatcher) {
    final List<Broker> brokers = new ArrayList<>();
    try {
        final CuratorFramework curator = zkFactory.get();
        final GetChildrenBuilder childrenBuilder = curator.getChildren();
        if (createWatcher) {
            LOG.info("Creating watcher on brokers change");
            childrenBuilder.usingWatcher((Watcher) event -> {
                if (event.getType() != Watcher.Event.EventType.NodeChildrenChanged) {
                    return;
                }
                this.scheduledExecutor.schedule(() -> updateBootstrapServersSafe(true), 0, TimeUnit.MILLISECONDS);
            });
        }
        for (final String brokerId : childrenBuilder.forPath(BROKERS_IDS_PATH)) {
            final byte[] brokerData = curator.getData().forPath(BROKERS_IDS_PATH + "/" + brokerId);
            brokers.add(Broker.fromByteJson(brokerData));
        }
    } catch (final Exception e) {
        LOG.error("Failed to fetch list of brokers from ZooKeeper", e);
        return;
    }

    if (brokers.isEmpty()) {
        return;
    }
    final String bootstrapServers = brokers.stream()
            .sorted()
            .map(Broker::toString)
            .collect(Collectors.joining(","));
    final String currentBootstrapServers =
            (String) kafkaProperties.get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG);

    if (bootstrapServers.equals(currentBootstrapServers)) {
        return;
    }

    kafkaProperties.setProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    this.ipAddressChangeListeners.forEach(listener -> {
        try {
            listener.run();
        } catch (final RuntimeException re) {
            LOG.error("Failed to process listener {}", re.getMessage(), re);
        }
    });
    LOG.info("Kafka client bootstrap servers changed: {}", bootstrapServers);
}