LeaseManager extends AbstractScheduledService implements LeaseManagement { @Override public void startService() throws LeaseManagementException { createLeaseManagementZNode(); createCurrentTSOZNode(); startAsync(); awaitRunning(); } LeaseManager(String tsoHostAndPort,
                 TSOChannelHandler tsoChannelHandler,
                 TSOStateManager stateManager,
                 long leasePeriodInMs,
                 String leasePath,
                 String currentTSOPath,
                 CuratorFramework zkClient,
                 Panicker panicker); @Override void startService(); @Override void stopService(); @Override boolean stillInLeasePeriod(); @Override String toString();  }
@Test(timeOut = 80_000) public void testLeaseManagerPanicsWhenUnexpectedInfoIsFoundInCurrentTSOZnode() throws Exception { final String TEST_TSO_LEASE_PATH = "/test_wronginfo_tsolease"; final String TEST_CURRENT_TSO_PATH = "/test_wronginfo_currenttso"; Panicker panicker = spy(new MockPanicker()); TSOStateManager stateManager1 = mock(TSOStateManager.class); when(stateManager1.initialize()).thenReturn(new TSOState(DUMMY_LOW_WATERMARK_1, DUMMY_EPOCH_1)); PausableLeaseManager leaseManager = new PausableLeaseManager(LEASE_MGR_ID_1, mock(TSOChannelHandler.class), stateManager1, TEST_LEASE_PERIOD_IN_MS, TEST_TSO_LEASE_PATH, TEST_CURRENT_TSO_PATH, zkClient, panicker); leaseManager.startService(); Thread.sleep(TEST_LEASE_PERIOD_IN_MS * 2); leaseManager.pausedInTryToRenewLeasePeriod(); zkClient.setData().forPath(TEST_CURRENT_TSO_PATH, "CorruptedData!!!".getBytes()); Thread.sleep(TEST_LEASE_PERIOD_IN_MS * 2); leaseManager.resume(); Thread.sleep(TEST_LEASE_PERIOD_IN_MS * 2); ArgumentCaptor<IllegalArgumentException> trowableIAE = ArgumentCaptor.forClass(IllegalArgumentException.class); verify(panicker, times(2)).panic(anyString(), trowableIAE.capture()); assertTrue(trowableIAE.getValue() != null); assertTrue(trowableIAE.getValue().getMessage().contains("Incorrect TSO Info found")); reset(panicker); zkClient.setData().forPath(TEST_CURRENT_TSO_PATH, "newTSO:12345#10000".getBytes()); leaseManager.pausedInTryToRenewLeasePeriod(); Thread.sleep(TEST_LEASE_PERIOD_IN_MS * 2); leaseManager.resume(); Thread.sleep(TEST_LEASE_PERIOD_IN_MS * 2); ArgumentCaptor<LeaseManagement.LeaseManagementException> trowableLME = ArgumentCaptor.forClass(LeaseManagement.LeaseManagementException.class); verify(panicker, times(2)).panic(anyString(), trowableLME.capture()); assertTrue(trowableLME.getValue() != null); assertTrue(trowableLME.getValue().getMessage().contains("Another TSO replica was found")); }