@Test public void testShutdownFailed() throws Exception { int reacquireDelayMillis = 1500; ServiceTriggers triggers1 = new ServiceTriggers(); ServiceTriggers triggers2 = new ServiceTriggers(); ServiceTimer timer1 = new ServiceTimer(); ServiceTimer timer2 = new ServiceTimer(); List<Event> events = Collections.synchronizedList(new ArrayList<>()); Service leader = newLeaderService(reacquireDelayMillis, TimeUnit.MILLISECONDS, supply( triggers1.listenTo(timer1.listenTo(trackEvents("1", events, new NopService() { @Override protected void shutDown() throws Exception { throw new Exception("ShutDown failed"); } }))), triggers2.listenTo(timer2.listenTo(trackEvents("2", events, new NopService()))))); leader.startAsync(); assertTrue(triggers1.getRunning().firedWithin(1, TimeUnit.MINUTES)); killSession(_curator); assertTrue(triggers1.getFailed().firedWithin(1, TimeUnit.MINUTES)); assertTrue(triggers2.getRunning().firedWithin(1, TimeUnit.MINUTES)); assertEquals(ImmutableList.of( new Event("1", Service.State.STARTING), new Event("1", Service.State.RUNNING), new Event("1", Service.State.STOPPING), new Event("1", Service.State.FAILED), new Event("2", Service.State.STARTING), new Event("2", Service.State.RUNNING) ), events); long actualDelayMillis = timer2.getStartedAt() - timer1.getStoppedAt(); assertTrue("Re-acquire delay was not observed: " + actualDelayMillis, actualDelayMillis >= reacquireDelayMillis); }
private Service listenTo(Service delegate) { delegate.addListener(new Listener() { @Override public void starting() { } @Override public void running() { } @Override public void stopping(State from) { } @Override public void terminated(State from) { closeLeaderLatch(); } @Override public void failed(State from, Throwable failure) { closeLeaderLatch(); } }, MoreExecutors.directExecutor()); return delegate; }
LeaderService extends AbstractExecutionThreadService { private Service listenTo(Service delegate) { delegate.addListener(new Listener() { @Override public void starting() { } @Override public void running() { } @Override public void stopping(State from) { } @Override public void terminated(State from) { closeLeaderLatch(); } @Override public void failed(State from, Throwable failure) { closeLeaderLatch(); } }, MoreExecutors.directExecutor()); return delegate; } }
LeaderService extends AbstractExecutionThreadService { private Service listenTo(Service delegate) { delegate.addListener(new Listener() { @Override public void starting() { } @Override public void running() { } @Override public void stopping(State from) { } @Override public void terminated(State from) { closeLeaderLatch(); } @Override public void failed(State from, Throwable failure) { closeLeaderLatch(); } }, MoreExecutors.directExecutor()); return delegate; } LeaderService(CuratorFramework curator, String leaderPath, String instanceId, String serviceName,
                         long reacquireDelay, TimeUnit reacquireDelayUnit, Supplier<Service> serviceFactory); }
LeaderService extends AbstractExecutionThreadService { private Service listenTo(Service delegate) { delegate.addListener(new Listener() { @Override public void starting() { } @Override public void running() { } @Override public void stopping(State from) { } @Override public void terminated(State from) { closeLeaderLatch(); } @Override public void failed(State from, Throwable failure) { closeLeaderLatch(); } }, MoreExecutors.directExecutor()); return delegate; } LeaderService(CuratorFramework curator, String leaderPath, String instanceId, String serviceName,
                         long reacquireDelay, TimeUnit reacquireDelayUnit, Supplier<Service> serviceFactory); String getId(); Collection<Participant> getParticipants(); Participant getLeader(); boolean hasLeadership(); Optional<Service> getCurrentDelegateService(); }
LeaderService extends AbstractExecutionThreadService { private Service listenTo(Service delegate) { delegate.addListener(new Listener() { @Override public void starting() { } @Override public void running() { } @Override public void stopping(State from) { } @Override public void terminated(State from) { closeLeaderLatch(); } @Override public void failed(State from, Throwable failure) { closeLeaderLatch(); } }, MoreExecutors.directExecutor()); return delegate; } LeaderService(CuratorFramework curator, String leaderPath, String instanceId, String serviceName,
                         long reacquireDelay, TimeUnit reacquireDelayUnit, Supplier<Service> serviceFactory); String getId(); Collection<Participant> getParticipants(); Participant getLeader(); boolean hasLeadership(); Optional<Service> getCurrentDelegateService();  }