### Question:
AbstractZkLedgerManager implements LedgerManager, Watcher { @Override public void unregisterLedgerMetadataListener(long ledgerId, LedgerMetadataListener listener) { Set<LedgerMetadataListener> listenerSet = listeners.get(ledgerId); if (listenerSet != null) { synchronized (listenerSet) { if (listenerSet.remove(listener)) { if (LOG.isDebugEnabled()) { LOG.debug("Unregistered ledger metadata listener {} on ledger {}.", listener, ledgerId); } } if (listenerSet.isEmpty()) { listeners.remove(ledgerId, listenerSet); } } } } protected  AbstractZkLedgerManager(AbstractConfiguration conf, ZooKeeper zk); abstract String getLedgerPath(long ledgerId); @Override void process(WatchedEvent event); @Override CompletableFuture<Versioned<LedgerMetadata>> createLedgerMetadata(long ledgerId,
                                                                             LedgerMetadata inputMetadata); @Override CompletableFuture<Void> removeLedgerMetadata(final long ledgerId, final Version version); @Override void registerLedgerMetadataListener(long ledgerId, LedgerMetadataListener listener); @Override void unregisterLedgerMetadataListener(long ledgerId, LedgerMetadataListener listener); @Override CompletableFuture<Versioned<LedgerMetadata>> readLedgerMetadata(long ledgerId); @Override CompletableFuture<Versioned<LedgerMetadata>> writeLedgerMetadata(long ledgerId, LedgerMetadata metadata,
                                                                            Version currentVersion); static boolean isSpecialZnode(String znode); static boolean isLeadgerIdGeneratorZnode(String znode); boolean isLedgerParentNode(String znode); @Override void close();  }### Answer:
@Test public void testUnregisterLedgerMetadataListener() throws Exception { long ledgerId = System.currentTimeMillis(); String ledgerStr = String.valueOf(ledgerId); LinkedBlockingQueue<LedgerMetadata> changes = new LinkedBlockingQueue<>(); LedgerMetadataListener listener = (ledgerId1, metadata) -> changes.add(metadata.getValue()); Stat stat = mock(Stat.class); when(stat.getVersion()).thenReturn(1234); when(stat.getCtime()).thenReturn(metadata.getCtime()); mockZkGetData( ledgerStr, true, KeeperException.Code.OK.intValue(), serDe.serialize(metadata), stat); ledgerManager.registerLedgerMetadataListener(ledgerId, listener); assertTrue(ledgerManager.listeners.containsKey(ledgerId)); LedgerMetadata change1 = changes.take(); assertEquals(metadata, change1); verify(mockZk, times(1)) .getData(anyString(), any(Watcher.class), any(DataCallback.class), any()); assertTrue(watchers.containsKey(ledgerStr)); Set<Watcher> watcherSet1 = watchers.get(ledgerStr); assertEquals(1, watcherSet1.size()); Watcher registeredWatcher1 = watcherSet1.stream().findFirst().get(); when(stat.getVersion()).thenReturn(1235); mockZkGetData( ledgerStr, true, KeeperException.Code.OK.intValue(), serDe.serialize(metadata), stat); ledgerManager.unregisterLedgerMetadataListener(ledgerId, listener); assertFalse(ledgerManager.listeners.containsKey(ledgerId)); notifyWatchedEvent( EventType.NodeDataChanged, KeeperState.SyncConnected, ledgerStr); assertNull(changes.poll()); verify(mockZk, times(1)) .getData(anyString(), any(Watcher.class), any(DataCallback.class), any()); }