KafkaBrokersTracker.java 文件源码

public void start() {
    if (!started.compareAndSet(false, true)) {
        throw new DefectException("Can only be started once!");
    }

    brokersById.addListener((MapChangeListener<String, KafkaBroker>) change -> version.incrementAndGet());
    zkClient.waitUntilExists("/brokers", TimeUnit.SECONDS, 10);

    executor.submit(() -> {
        zkClient.subscribeChildChanges("/brokers/ids", (parentPath, currentChilds) ->
                updateBrokers(ImmutableSet.copyOf(currentChilds)));
        updateBrokers(ImmutableSet.copyOf(zkClient.getChildren("/brokers/ids")));
    });
}