### Question:
InMemClusterMetadataStore implements ClusterMetadataStore { @Override public synchronized void watchClusterAssignmentData(Consumer<Void> watcher, Executor executor) { WatcherAndExecutor wae = watchers.get(watcher); if (null == wae) { wae = new WatcherAndExecutor(watcher, executor); watchers.put(watcher, wae); } } InMemClusterMetadataStore(int numStorageContainers); @Override synchronized boolean initializeCluster(int numStorageContainers,
                                               Optional<String> segmentStorePath); @Override synchronized ClusterAssignmentData getClusterAssignmentData(); @Override synchronized void updateClusterAssignmentData(ClusterAssignmentData assignmentData); @Override synchronized void watchClusterAssignmentData(Consumer<Void> watcher, Executor executor); @Override synchronized void unwatchClusterAssignmentData(Consumer<Void> watcher); @Override synchronized ClusterMetadata getClusterMetadata(); @Override synchronized void updateClusterMetadata(ClusterMetadata clusterMetadata); @Override void close();  }### 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()); }