### Question:
ZkClusterMetadataStore implements ClusterMetadataStore { @Override public void unwatchClusterAssignmentData(Consumer<Void> watcher) { synchronized (this) { NodeCacheListener listener = assignmentDataConsumers.remove(watcher); if (null != listener && null != assignmentDataCache) { assignmentDataCache.getListenable().removeListener(listener); } if (assignmentDataConsumers.isEmpty() && null != assignmentDataCache) { try { assignmentDataCache.close(); } catch (IOException e) { log.warn("Failed to close assignment data cache when there is no watcher", e); } } } } 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 testUnwatchClusterAssignmentData() throws Exception { 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<>(); CountDownLatch latch = new CountDownLatch(2); Consumer<Void> dataConsumer = ignored -> { log.info("Notify cluster assignment data changed"); latch.countDown(); FutureUtils.complete(watchFuture, null); }; assertEquals(0, store.getNumWatchers()); store.watchClusterAssignmentData(dataConsumer, executor); assertEquals(1, store.getNumWatchers()); store.updateClusterAssignmentData(assignmentData); watchFuture.join(); assertEquals(1, latch.getCount()); assertEquals(assignmentData, store.getClusterAssignmentData()); store.unwatchClusterAssignmentData(dataConsumer); assertEquals(0, store.getNumWatchers()); }