### Question:
ZkClusterMetadataStore implements ClusterMetadataStore { @Override public void watchClusterAssignmentData(Consumer<Void> watcher, Executor executor) { synchronized (this) { if (assignmentDataCache == null) { assignmentDataCache = new NodeCache(client, zkClusterAssignmentPath); try { assignmentDataCache.start(); } catch (Exception e) { throw new StorageRuntimeException("Failed to watch cluster assignment data", e); } } NodeCacheListener listener = assignmentDataConsumers.get(watcher); if (null == listener) { listener = () -> watcher.accept(null); assignmentDataConsumers.put(watcher, listener); assignmentDataCache.getListenable().addListener(listener, executor); } } } ZkClusterMetadataStore(CuratorFramework client, String zkServers, String zkRootPath); @Override void close(); @Override boolean initializeCluster(int numStorageContainers, Optional<String> segmentStorePath); @Override ClusterAssignmentData getClusterAssignmentData(); @Override void updateClusterAssignmentData(ClusterAssignmentData assigmentData); @Override void watchClusterAssignmentData(Consumer<Void> watcher, Executor executor); @Override void unwatchClusterAssignmentData(Consumer<Void> watcher); @Override ClusterMetadata getClusterMetadata(); @Override void updateClusterMetadata(ClusterMetadata metadata);  }### Answer:
@Test public void testWatchClusterAssignmentData() { ClusterAssignmentData assignmentData = ClusterAssignmentData.newBuilder() .putServers( "server-0", ServerAssignmentData.newBuilder() .addContainers(1L) .addContainers(2L) .build()) .build(); @Cleanup("shutdown") ExecutorService executor = Executors.newSingleThreadExecutor(); CompletableFuture<Void> watchFuture = new CompletableFuture<>(); store.watchClusterAssignmentData(data -> { FutureUtils.complete(watchFuture, null); }, executor); store.updateClusterAssignmentData(assignmentData); watchFuture.join(); assertEquals(assignmentData, store.getClusterAssignmentData()); }