content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *   http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing,
   * software distributed under the License is distributed on an
   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
   * KIND, either express or implied.  See the License for the
   * specific language governing permissions and limitations
   * under the License.
   */
  package org.apache.bookkeeper.meta;
  
  import static org.apache.bookkeeper.common.concurrent.FutureUtils.result;
  import static org.apache.bookkeeper.meta.AbstractZkLedgerManager.ZK_CONNECT_BACKOFF_MS;
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertFalse;
  import static org.junit.Assert.assertNull;
  import static org.junit.Assert.assertSame;
  import static org.junit.Assert.assertTrue;
  import static org.junit.Assert.fail;
  import static org.mockito.ArgumentMatchers.any;
  import static org.mockito.ArgumentMatchers.anyInt;
  import static org.mockito.ArgumentMatchers.anyLong;
  import static org.mockito.ArgumentMatchers.anyString;
  import static org.mockito.ArgumentMatchers.eq;
  import static org.mockito.Mockito.CALLS_REAL_METHODS;
  import static org.mockito.Mockito.doAnswer;
  import static org.mockito.Mockito.mock;
  import static org.mockito.Mockito.mockStatic;
  import static org.mockito.Mockito.times;
  import static org.mockito.Mockito.verify;
  import static org.mockito.Mockito.when;
  import static org.mockito.Mockito.withSettings;
  
  import com.google.common.collect.Lists;
  import java.time.Duration;
  import java.util.List;
  import java.util.Optional;
  import java.util.Set;
  import java.util.concurrent.Executors;
  import java.util.concurrent.LinkedBlockingQueue;
  import java.util.concurrent.ScheduledExecutorService;
  import java.util.concurrent.TimeUnit;
  import org.apache.bookkeeper.client.BKException;
  import org.apache.bookkeeper.client.BKException.Code;
  import org.apache.bookkeeper.client.LedgerMetadataBuilder;
  import org.apache.bookkeeper.client.api.DigestType;
  import org.apache.bookkeeper.client.api.LedgerMetadata;
  import org.apache.bookkeeper.common.testing.executors.MockExecutorController;
  import org.apache.bookkeeper.conf.ClientConfiguration;
  import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
  import org.apache.bookkeeper.net.BookieId;
  import org.apache.bookkeeper.net.BookieSocketAddress;
  import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.LedgerMetadataListener;
  import org.apache.bookkeeper.util.ZkUtils;
  import org.apache.bookkeeper.versioning.LongVersion;
  import org.apache.bookkeeper.versioning.Version;
  import org.apache.bookkeeper.versioning.Versioned;
  import org.apache.bookkeeper.zookeeper.MockZooKeeperTestCase;
  import org.apache.zookeeper.AsyncCallback.DataCallback;
  import org.apache.zookeeper.AsyncCallback.StatCallback;
  import org.apache.zookeeper.AsyncCallback.VoidCallback;
  import org.apache.zookeeper.CreateMode;
  import org.apache.zookeeper.KeeperException;
  import org.apache.zookeeper.Watcher;
  import org.apache.zookeeper.Watcher.Event.EventType;
  import org.apache.zookeeper.Watcher.Event.KeeperState;
  import org.apache.zookeeper.data.Stat;
  import org.junit.After;
  import org.junit.Before;
  import org.junit.Test;
  import org.junit.runner.RunWith;
  import org.mockito.MockedStatic;
  import org.mockito.junit.MockitoJUnitRunner;
  
  /**
   * Unit test of {@link AbstractZkLedgerManager}.
   */
  @RunWith(MockitoJUnitRunner.Silent.class)
  public class AbstractZkLedgerManagerTest extends MockZooKeeperTestCase {
  
      private ClientConfiguration conf;
      private AbstractZkLedgerManager ledgerManager;
      private ScheduledExecutorService scheduler;
      private MockExecutorController schedulerController;
      private LedgerMetadata metadata;
      private LedgerMetadataSerDe serDe;
      private MockedStatic<Executors> executorsMockedStatic;
  
      @Before
      public void setup() throws Exception {
          executorsMockedStatic = mockStatic(Executors.class, CALLS_REAL_METHODS);
  
          super.setup();
  
          this.scheduler = mock(ScheduledExecutorService.class);
          this.schedulerController = new MockExecutorController()
              .controlSubmit(scheduler)
              .controlSchedule(scheduler)
              .controlExecute(scheduler)
              .controlScheduleAtFixedRate(scheduler, 10);
  
          executorsMockedStatic.when(() -> Executors.newSingleThreadScheduledExecutor(any())).thenReturn(scheduler);
  
          this.conf = new ClientConfiguration();
          this.ledgerManager = mock(
              AbstractZkLedgerManager.class,
              withSettings()
                  .useConstructor(conf, mockZk)
                  .defaultAnswer(CALLS_REAL_METHODS));
          List<BookieId> ensemble = Lists.newArrayList(new BookieSocketAddress("192.0.2.1", 3181).toBookieId(),
                  new BookieSocketAddress("192.0.2.2", 3181).toBookieId(),
                  new BookieSocketAddress("192.0.2.3", 3181).toBookieId(),
                  new BookieSocketAddress("192.0.2.4", 3181).toBookieId(),
                  new BookieSocketAddress("192.0.2.5", 3181).toBookieId());
          this.metadata = LedgerMetadataBuilder.create()
              .withId(123L)
              .withDigestType(DigestType.CRC32C).withPassword(new byte[0])
              .withEnsembleSize(5)
              .withWriteQuorumSize(3)
              .withAckQuorumSize(3)
              .newEnsembleEntry(0L, ensemble)
              .withCreationTime(12345L).build();
  
          doAnswer(invocationOnMock -> {
              long ledgerId = invocationOnMock.getArgument(0);
              return String.valueOf(ledgerId);
          }).when(ledgerManager).getLedgerPath(anyLong());
          doAnswer(invocationOnMock -> {
              String ledgerStr = invocationOnMock.getArgument(0);
              return Long.parseLong(ledgerStr);
          }).when(ledgerManager).getLedgerId(anyString());
  
          // verify constructor
          assertEquals(ZKMetadataDriverBase.resolveZkLedgersRootPath(conf), ledgerManager.ledgerRootPath);
          assertSame(mockZk, ledgerManager.zk);
          assertSame(conf, ledgerManager.conf);
          assertSame(scheduler, ledgerManager.scheduler);
  
          this.serDe = new LedgerMetadataSerDe();
      }
  
      @After
      public void teardown() throws Exception {
          super.teardown();
  
          executorsMockedStatic.close();
  
          if (null != ledgerManager) {
              ledgerManager.close();
  
              // zookeeper is passed in, it should not be closed.
              verify(mockZk, times(0)).close();
              verify(scheduler, times(1)).shutdown();
          }
      }
  
      @Test
      public void testCreateLedgerMetadataSuccess() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
          mockZkUtilsAsyncCreateFullPathOptimistic(
              ledgerStr, CreateMode.PERSISTENT,
              KeeperException.Code.OK.intValue(), ledgerStr
          );
  
          Versioned<LedgerMetadata> result = ledgerManager.createLedgerMetadata(ledgerId, metadata).get();
  
          assertEquals(new LongVersion(0), result.getVersion());
      }
  
      @Test
      public void testCreateLedgerMetadataNodeExists() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
          mockZkUtilsAsyncCreateFullPathOptimistic(
              ledgerStr, CreateMode.PERSISTENT,
              KeeperException.Code.NODEEXISTS.intValue(), null);
          Stat stat = mock(Stat.class);
          when(stat.getVersion()).thenReturn(1234);
          when(stat.getCtime()).thenReturn(metadata.getCtime());
          /*
           * this is needed because in AbstractZkLedgerManager.readLedgerMetadata
           * if MetadataFormatVersion is >2, then for createLedgerMetadata if we
           * get NODEEXISTS exception then it will try to read to make sure ledger
           * creation is robust to ZK connection loss. Please check Issue #1967.
           */
          mockZkGetData(
                  ledgerStr, false,
                  KeeperException.Code.OK.intValue(), serDe.serialize(metadata), stat);
          try {
              result(ledgerManager.createLedgerMetadata(ledgerId, metadata));
              fail("Should fail to create ledger metadata if the ledger already exists");
          } catch (Exception e) {
              assertTrue(e instanceof BKException);
              BKException bke = (BKException) e;
              assertEquals(Code.LedgerExistException, bke.getCode());
          }
      }
  
      @Test
      public void testCreateLedgerMetadataException() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
          mockZkUtilsAsyncCreateFullPathOptimistic(
              ledgerStr, CreateMode.PERSISTENT,
              KeeperException.Code.CONNECTIONLOSS.intValue(), null);
  
          try {
              result(ledgerManager.createLedgerMetadata(ledgerId, metadata));
              fail("Should fail to create ledger metadata when encountering zookeeper exception");
          } catch (Exception e) {
              assertTrue(e instanceof BKException);
              BKException bke = (BKException) e;
              assertEquals(Code.ZKException, bke.getCode());
              assertTrue(bke.getCause() instanceof KeeperException);
          }
      }
  
      @Test
      public void testRemoveLedgerMetadataSuccess() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
          LongVersion version = new LongVersion(1234L);
  
          mockZkDelete(
              ledgerStr, (int) version.getLongVersion(),
              KeeperException.Code.OK.intValue());
  
          ledgerManager.removeLedgerMetadata(ledgerId, version).get();
  
          verify(mockZk, times(1))
              .delete(eq(ledgerStr), eq(1234), any(VoidCallback.class), eq(null));
      }
  
      @Test
      public void testRemoveLedgerMetadataVersionAny() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
  
          mockZkDelete(
              ledgerStr, -1,
              KeeperException.Code.OK.intValue());
  
          ledgerManager.removeLedgerMetadata(ledgerId, Version.ANY).get();
  
          verify(mockZk, times(1))
              .delete(eq(ledgerStr), eq(-1), any(VoidCallback.class), eq(null));
      }
  
      @Test
      public void testRemoveLedgerMetadataVersionNew() throws Exception {
          testRemoveLedgerMetadataInvalidVersion(Version.NEW);
      }
  
      @Test
      public void testRemoveLedgerMetadataUnknownVersionType() throws Exception {
          Version version = mock(Version.class);
          testRemoveLedgerMetadataInvalidVersion(version);
      }
  
      private void testRemoveLedgerMetadataInvalidVersion(Version version) throws Exception {
          long ledgerId = System.currentTimeMillis();
  
          try {
              result(ledgerManager.removeLedgerMetadata(ledgerId, version));
              fail("Should fail to remove metadata if version is " + Version.NEW);
          } catch (BKException bke) {
              assertEquals(Code.MetadataVersionException, bke.getCode());
          }
      }
  
      @Test
      public void testRemoveLedgerMetadataNoNode() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
          LongVersion version = new LongVersion(1234L);
  
          mockZkDelete(
              ledgerStr, (int) version.getLongVersion(),
              KeeperException.Code.NONODE.intValue());
  
          try {
              result(ledgerManager.removeLedgerMetadata(ledgerId, version));
          } catch (BKException bke) {
              fail("Should succeed");
          }
  
          verify(mockZk, times(1))
              .delete(eq(ledgerStr), eq(1234), any(VoidCallback.class), eq(null));
      }
  
      @Test
      public void testRemoveLedgerMetadataException() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
          LongVersion version = new LongVersion(1234L);
  
          mockZkDelete(
              ledgerStr, (int) version.getLongVersion(),
              KeeperException.Code.CONNECTIONLOSS.intValue());
  
          try {
              result(ledgerManager.removeLedgerMetadata(ledgerId, version));
              fail("Should fail to remove metadata upon ZKException");
          } catch (BKException bke) {
              assertEquals(Code.ZKException, bke.getCode());
          }
  
          verify(mockZk, times(1))
              .delete(eq(ledgerStr), eq(1234), any(VoidCallback.class), eq(null));
      }
  
      @Test
      public void testRemoveLedgerMetadataHierarchical() throws Exception {
          HierarchicalLedgerManager hlm = new HierarchicalLedgerManager(conf, mockZk);
          testRemoveLedgerMetadataHierarchicalLedgerManager(hlm);
      }
  
      @Test
      public void testRemoveLedgerMetadataLongHierarchical() throws Exception {
          LongHierarchicalLedgerManager hlm = new LongHierarchicalLedgerManager(conf, mockZk);
          testRemoveLedgerMetadataHierarchicalLedgerManager(hlm);
      }
  
      private void testRemoveLedgerMetadataHierarchicalLedgerManager(AbstractZkLedgerManager lm) throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = lm.getLedgerPath(ledgerId);
          LongVersion version = new LongVersion(1234L);
  
          mockZkUtilsAsyncDeleteFullPathOptimistic(
              ledgerStr, (int) version.getLongVersion(),
              KeeperException.Code.OK.intValue());
  
          lm.removeLedgerMetadata(ledgerId, version).get();
  
          ZkUtils.asyncDeleteFullPathOptimistic(
              eq(mockZk), eq(ledgerStr), eq(1234), any(VoidCallback.class), eq(ledgerStr));
  
      }
  
      @Test
      public void testReadLedgerMetadataSuccess() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
  
          Stat stat = mock(Stat.class);
          when(stat.getVersion()).thenReturn(1234);
          when(stat.getCtime()).thenReturn(metadata.getCtime());
          mockZkGetData(
              ledgerStr, false,
              KeeperException.Code.OK.intValue(), serDe.serialize(metadata), stat);
  
          Versioned<LedgerMetadata> readMetadata = result(ledgerManager.readLedgerMetadata(ledgerId));
          assertEquals(metadata, readMetadata.getValue());
          assertEquals(new LongVersion(1234), readMetadata.getVersion());
  
          verify(mockZk, times(1))
              .getData(eq(ledgerStr), eq(null), any(DataCallback.class), any());
      }
  
      @Test
      public void testReadLedgerMetadataNoNode() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
  
          mockZkGetData(
              ledgerStr, false,
              KeeperException.Code.NONODE.intValue(), null, null);
  
          try {
              result(ledgerManager.readLedgerMetadata(ledgerId));
              fail("Should fail on reading ledger metadata if a ledger doesn't exist");
          } catch (BKException bke) {
              assertEquals(Code.NoSuchLedgerExistsOnMetadataServerException, bke.getCode());
          }
  
          verify(mockZk, times(1))
              .getData(eq(ledgerStr), eq(null), any(DataCallback.class), any());
      }
  
      @Test
      public void testReadLedgerMetadataException() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
  
          mockZkGetData(
              ledgerStr, false,
              KeeperException.Code.CONNECTIONLOSS.intValue(), null, null);
  
          try {
              result(ledgerManager.readLedgerMetadata(ledgerId));
              fail("Should fail on reading ledger metadata if a ledger doesn't exist");
          } catch (BKException bke) {
              assertEquals(Code.ZKException, bke.getCode());
          }
  
          verify(mockZk, times(1))
              .getData(eq(ledgerStr), eq(null), any(DataCallback.class), any());
      }
  
      @Test
      public void testReadLedgerMetadataStatMissing() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
  
          mockZkGetData(
              ledgerStr, false,
              KeeperException.Code.OK.intValue(), serDe.serialize(metadata), null);
  
          try {
              result(ledgerManager.readLedgerMetadata(ledgerId));
              fail("Should fail on reading ledger metadata if a ledger doesn't exist");
          } catch (BKException bke) {
              assertEquals(Code.ZKException, bke.getCode());
          }
  
          verify(mockZk, times(1))
              .getData(eq(ledgerStr), eq(null), any(DataCallback.class), any());
      }
  
      @Test
      public void testReadLedgerMetadataDataCorrupted() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
  
          Stat stat = mock(Stat.class);
          when(stat.getVersion()).thenReturn(1234);
          when(stat.getCtime()).thenReturn(metadata.getCtime());
          mockZkGetData(
              ledgerStr, false,
              KeeperException.Code.OK.intValue(), new byte[0], stat);
  
          try {
              result(ledgerManager.readLedgerMetadata(ledgerId));
              fail("Should fail on reading ledger metadata if a ledger doesn't exist");
          } catch (BKException bke) {
              assertEquals(Code.ZKException, bke.getCode());
          }
  
          verify(mockZk, times(1))
              .getData(eq(ledgerStr), eq(null), any(DataCallback.class), any());
      }
  
      @Test
      public void testWriteLedgerMetadataSuccess() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
  
          Stat stat = mock(Stat.class);
          when(stat.getVersion()).thenReturn(1235);
          when(stat.getCtime()).thenReturn(metadata.getCtime());
          mockZkSetData(
              ledgerStr, serDe.serialize(metadata), 1234,
              KeeperException.Code.OK.intValue(), stat);
  
          Version v = ledgerManager.writeLedgerMetadata(ledgerId, metadata, new LongVersion(1234L)).get().getVersion();
  
          assertEquals(new LongVersion(1235L), v);
  
          verify(mockZk, times(1))
              .setData(eq(ledgerStr), any(byte[].class), eq(1234), any(StatCallback.class), any());
      }
  
      @Test
      public void testWriteLedgerMetadataBadVersion() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
  
          mockZkSetData(
              ledgerStr, serDe.serialize(metadata), 1234,
              KeeperException.Code.BADVERSION.intValue(), null);
  
          try {
              result(ledgerManager.writeLedgerMetadata(ledgerId, metadata, new LongVersion(1234L)));
              fail("Should fail on writing ledger metadata if encountering bad version");
          } catch (BKException bke) {
              assertEquals(Code.MetadataVersionException, bke.getCode());
          }
  
          verify(mockZk, times(1))
              .setData(eq(ledgerStr), any(byte[].class), eq(1234), any(StatCallback.class), any());
      }
  
      @Test
      public void testWriteLedgerMetadataException() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
  
          mockZkSetData(
              ledgerStr, serDe.serialize(metadata), 1234,
              KeeperException.Code.CONNECTIONLOSS.intValue(), null);
  
          try {
              result(ledgerManager.writeLedgerMetadata(ledgerId, metadata, new LongVersion(1234L)));
              fail("Should fail on writing ledger metadata if encountering zookeeper exceptions");
          } catch (BKException bke) {
              assertEquals(Code.ZKException, bke.getCode());
          }
  
  
          verify(mockZk, times(1))
              .setData(eq(ledgerStr), any(byte[].class), eq(1234), any(StatCallback.class), any());
      }
  
      @Test
      public void testWriteLedgerMetadataInvalidVersion() throws Exception {
          Version[] versions = new Version[] {
              Version.NEW,
              Version.ANY,
              mock(Version.class)
          };
          for (Version version : versions) {
              testWriteLedgerMetadataInvalidVersion(version);
          }
      }
  
      private void testWriteLedgerMetadataInvalidVersion(Version invalidVersion) throws Exception {
          long ledgerId = System.currentTimeMillis();
  
          try {
              result(ledgerManager.writeLedgerMetadata(ledgerId, metadata, invalidVersion));
              fail("Should fail on writing ledger metadata if an invalid version is provided.");
          } catch (BKException bke) {
              assertEquals(Code.MetadataVersionException, bke.getCode());
          }
  
          verify(mockZk, times(0))
              .setData(anyString(), any(byte[].class), anyInt(), any(StatCallback.class), any());
      }
  
      @Test
      public void testLedgerMetadataListener() 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);
  
          // the listener will be notified with first get
          LedgerMetadata change1 = changes.take();
          assertEquals(metadata, change1);
          verify(mockZk, times(1))
              .getData(anyString(), any(Watcher.class), any(DataCallback.class), any());
  
          // the watcher is registered for receiving watched event
          assertTrue(watchers.containsKey(ledgerStr));
          Set<Watcher> watcherSet1 = watchers.get(ledgerStr);
          assertEquals(1, watcherSet1.size());
          Watcher registeredWatcher1 = watcherSet1.stream().findFirst().get();
  
          // mock get data to return an updated metadata
          when(stat.getVersion()).thenReturn(1235);
          mockZkGetData(
              ledgerStr, true,
              KeeperException.Code.OK.intValue(), serDe.serialize(metadata), stat);
  
          // notify the watcher event
          notifyWatchedEvent(
              EventType.NodeDataChanged, KeeperState.SyncConnected, ledgerStr);
  
          // the listener should receive an updated metadata
          LedgerMetadata change2 = changes.take();
          assertEquals(metadata, change2);
          verify(mockZk, times(2))
              .getData(anyString(), any(Watcher.class), any(DataCallback.class), any());
  
          // after the listener receive an updated metadata, a new watcher should be registered
          // for subsequent changes again.
          assertTrue(watchers.containsKey(ledgerStr));
          Set<Watcher> watcherSet2 = watchers.get(ledgerStr);
          assertEquals(1, watcherSet2.size());
          Watcher registeredWatcher2 = watcherSet2.stream().findFirst().get();
  
          // zookeeper watchers are same, since there is only one giant watcher per ledger manager.
          assertSame(registeredWatcher1, registeredWatcher2);
  
          // verify scheduler
          verify(scheduler, times(2)).submit(any(Runnable.class));
          verify(scheduler, times(0))
              .schedule(any(Runnable.class), anyLong(), any(TimeUnit.class));
      }
  
      @Test
      public void testLedgerMetadataListenerOnLedgerDeleted() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
  
          LinkedBlockingQueue<Optional<LedgerMetadata>> changes = new LinkedBlockingQueue<>();
          LedgerMetadataListener listener =
              (ledgerId1, metadata) -> changes.add(Optional.ofNullable(metadata != null ? metadata.getValue() : null));
  
          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));
  
          // the listener will be notified with first get
          LedgerMetadata change1 = changes.take().get();
          assertEquals(metadata, change1);
          verify(mockZk, times(1))
              .getData(anyString(), any(Watcher.class), any(DataCallback.class), any());
  
          // the watcher is registered for receiving watched event
          assertTrue(watchers.containsKey(ledgerStr));
  
          // mock get data to simulate an ledger is deleted
          mockZkGetData(
              ledgerStr, true,
              KeeperException.Code.NONODE.intValue(), null, null);
  
          // notify the watcher event
          notifyWatchedEvent(
              EventType.NodeDataChanged, KeeperState.SyncConnected, ledgerStr);
  
          // the listener should be removed from listener set and not receive an updated metadata anymore
          Optional<LedgerMetadata> change2 = changes.take();
          assertFalse(change2.isPresent());
          assertFalse(ledgerManager.listeners.containsKey(ledgerId));
  
          // verify scheduler: the listener is only triggered once
          verify(scheduler, times(1)).submit(any(Runnable.class));
          verify(scheduler, times(0)).schedule(
              any(Runnable.class), anyLong(), any(TimeUnit.class));
  
          // no watcher is registered
          assertFalse(watchers.containsKey(ledgerStr));
      }
  
      @Test
      public void testLedgerMetadataListenerOnLedgerDeletedEvent() throws Exception {
          long ledgerId = System.currentTimeMillis();
          String ledgerStr = String.valueOf(ledgerId);
  
          LinkedBlockingQueue<Optional<LedgerMetadata>> changes = new LinkedBlockingQueue<>();
          LedgerMetadataListener listener =
              (ledgerId1, metadata) -> changes.add(
                      Optional.ofNullable(metadata != null ? metadata.getValue() : null));
  
          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));
  
          // the listener will be notified with first get
          LedgerMetadata change1 = changes.take().get();
          assertEquals(metadata, change1);
          verify(mockZk, times(1))
              .getData(anyString(), any(Watcher.class), any(DataCallback.class), any());
  
          // the watcher is registered for receiving watched event
          assertTrue(watchers.containsKey(ledgerStr));
  
          // notify the watcher event
          notifyWatchedEvent(
              EventType.NodeDeleted, KeeperState.SyncConnected, ledgerStr);
  
          // the listener should be removed from listener set and a null change is notified.
          Optional<LedgerMetadata> change2 = changes.take();
          assertFalse(change2.isPresent());
          // no more `getData` is called.
          verify(mockZk, times(1))
              .getData(anyString(), any(Watcher.class), any(DataCallback.class), any());
          // listener is automatically unregistered after a ledger is deleted.
          assertFalse(ledgerManager.listeners.containsKey(ledgerId));
      }
  
      @Test
      public void testLedgerMetadataListenerOnRetries() 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());
  
          // fail the first get, so the ledger manager will retry get data again.
          mockZkGetData(
              ledgerStr, true,
              KeeperException.Code.SESSIONEXPIRED.intValue(), null, null);
  
          ledgerManager.registerLedgerMetadataListener(ledgerId, listener);
          assertTrue(ledgerManager.listeners.containsKey(ledgerId));
  
          // the listener will not be notified with any updates
          assertNull(changes.poll());
          // an retry task is scheduled
          verify(scheduler, times(1))
              .schedule(any(Runnable.class), anyLong(), any(TimeUnit.class));
          // zookeeper is called once
          verify(mockZk, times(1))
              .getData(anyString(), any(Watcher.class), any(DataCallback.class), any());
          // watcher is not registered since getData call is failed
          assertFalse(watchers.containsKey(ledgerStr));
  
          // mock get data to return a valid response
          mockZkGetData(
              ledgerStr, true,
              KeeperException.Code.OK.intValue(), serDe.serialize(metadata), stat);
  
          schedulerController.advance(Duration.ofMillis(ZK_CONNECT_BACKOFF_MS));
  
          // the listener will be notified with first get
          LedgerMetadata change = changes.take();
          assertEquals(metadata, change);
          verify(mockZk, times(2))
              .getData(anyString(), any(Watcher.class), any(DataCallback.class), any());
  
          // watcher is registered after successfully `getData`
          assertTrue(watchers.containsKey(ledgerStr));
      }
  
      @Test
      public void testLedgerMetadataListenerOnSessionExpired() 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);
  
          // the listener will be notified with first get
          LedgerMetadata change1 = changes.take();
          assertEquals(metadata, change1);
          verify(mockZk, times(1))
              .getData(anyString(), any(Watcher.class), any(DataCallback.class), any());
  
          // the watcher is registered for receiving watched event
          assertTrue(watchers.containsKey(ledgerStr));
          Set<Watcher> watcherSet1 = watchers.get(ledgerStr);
          assertEquals(1, watcherSet1.size());
          Watcher registeredWatcher1 = watcherSet1.stream().findFirst().get();
  
          // simulate session expired
          notifyWatchedEvent(
              EventType.None, KeeperState.Expired, ledgerStr);
  
          // ledger manager will retry to read metadata again
          LedgerMetadata change2 = changes.take();
          assertEquals(metadata, change2);
          verify(mockZk, times(2))
              .getData(anyString(), any(Watcher.class), any(DataCallback.class), any());
  
          // the watcher is registered for receiving watched event
          assertTrue(watchers.containsKey(ledgerStr));
          Set<Watcher> watcherSet2 = watchers.get(ledgerStr);
          assertEquals(1, watcherSet2.size());
          Watcher registeredWatcher2 = watcherSet2.stream().findFirst().get();
  
          assertSame(registeredWatcher1, registeredWatcher2);
      }
  
      @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));
  
          // the listener will be notified with first get
          LedgerMetadata change1 = changes.take();
          assertEquals(metadata, change1);
          verify(mockZk, times(1))
              .getData(anyString(), any(Watcher.class), any(DataCallback.class), any());
  
          // the watcher is registered for receiving watched event
          assertTrue(watchers.containsKey(ledgerStr));
          Set<Watcher> watcherSet1 = watchers.get(ledgerStr);
          assertEquals(1, watcherSet1.size());
          Watcher registeredWatcher1 = watcherSet1.stream().findFirst().get();
  
          // mock get data to return an updated metadata
          when(stat.getVersion()).thenReturn(1235);
          mockZkGetData(
              ledgerStr, true,
              KeeperException.Code.OK.intValue(), serDe.serialize(metadata), stat);
  
          mockZkRemoveWatcher();
  
          // unregister the listener
          ledgerManager.unregisterLedgerMetadataListener(ledgerId, listener);
          assertFalse(ledgerManager.listeners.containsKey(ledgerId));
          assertFalse(watchers.containsKey(ledgerStr));
          verify(mockZk, times(1)).removeWatches(eq(ledgerManager.getLedgerPath(ledgerId)),
                  any(Watcher.class), any(Watcher.WatcherType.class), any(Boolean.class),
                  any(VoidCallback.class), any());
  
  
          // notify the watcher event
          notifyWatchedEvent(
              EventType.NodeDataChanged, KeeperState.SyncConnected, ledgerStr);
  
          // since listener is already unregistered so no more `getData` is issued.
          assertNull(changes.poll());
          verify(mockZk, times(1))
              .getData(anyString(), any(Watcher.class), any(DataCallback.class), any());
      }
  
  }
  