LeaderSelector implements Closeable { public void start() { Preconditions.checkState(state.compareAndSet(State.LATENT, State.STARTED), "Cannot be started more than once"); Preconditions.checkState(!executorService.isShutdown(), "Already started"); Preconditions.checkState(!hasLeadership, "Already has leadership"); client.getConnectionStateListenable().addListener(listener); requeue(); } LeaderSelector(CuratorFramework client, String leaderPath, LeaderSelectorListener listener); @SuppressWarnings("UnusedParameters") @Deprecated  LeaderSelector(CuratorFramework client, String leaderPath, ThreadFactory threadFactory, Executor executor, LeaderSelectorListener listener);  LeaderSelector(CuratorFramework client, String leaderPath, ExecutorService executorService, LeaderSelectorListener listener);  LeaderSelector(CuratorFramework client, String leaderPath, CloseableExecutorService executorService, LeaderSelectorListener listener); void autoRequeue(); void setId(String id); String getId(); void start(); boolean requeue(); synchronized void close(); Collection<Participant> getParticipants(); Participant getLeader(); boolean hasLeadership(); synchronized void interruptLeadership();  }
@Test public void testInterruption() throws Exception { Timing2 timing = new Timing2(); LeaderSelector selector = null; CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1)); try { client.start(); CountDownLatch exitLatch = new CountDownLatch(1); BlockingQueue<Thread> threadExchange = new ArrayBlockingQueue<>(1); LeaderSelectorListener listener = new LeaderSelectorListenerAdapter() { @Override public void takeLeadership(CuratorFramework client) throws Exception { threadExchange.put(Thread.currentThread()); try { Thread.currentThread().join(); } finally { exitLatch.countDown(); } } }; selector = new LeaderSelector(client, PATH_NAME, listener); selector.failedMutexReleaseCount = new AtomicInteger(); selector.start(); Thread leaderThread = timing.takeFromQueue(threadExchange); leaderThread.interrupt(); Assert.assertTrue(timing.awaitLatch(exitLatch)); timing.sleepABit(); Assert.assertEquals(0, selector.failedMutexReleaseCount.get()); } finally { CloseableUtils.closeQuietly(selector); CloseableUtils.closeQuietly(client); } }
@Test public void testLeaderNodeDeleteOnInterrupt() throws Exception { Timing2 timing = new Timing2(); LeaderSelector selector = null; CuratorFramework client = null; try { client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1)); final CountDownLatch reconnectedLatch = new CountDownLatch(1); ConnectionStateListener connectionStateListener = new ConnectionStateListener() { @Override public void stateChanged(CuratorFramework client, ConnectionState newState) { if ( newState == ConnectionState.RECONNECTED ) { reconnectedLatch.countDown(); } } }; client.getConnectionStateListenable().addListener(connectionStateListener); client.start(); final BlockingQueue<Thread> queue = new ArrayBlockingQueue<Thread>(1); LeaderSelectorListener listener = new LeaderSelectorListener() { @Override public void takeLeadership(CuratorFramework client) throws Exception { queue.add(Thread.currentThread()); try { Thread.currentThread().join(); } catch ( InterruptedException e ) { Thread.currentThread().interrupt(); } } @Override public void stateChanged(CuratorFramework client, ConnectionState newState) { } }; selector = new LeaderSelector(client, "/leader", listener); selector.start(); Thread leaderThread = timing.takeFromQueue(queue); server.stop(); leaderThread.interrupt(); server.restart(); Assert.assertTrue(timing.awaitLatch(reconnectedLatch)); timing.sleepABit(); Assert.assertEquals(client.getChildren().forPath("/leader").size(), 0); } finally { CloseableUtils.closeQuietly(selector); CloseableUtils.closeQuietly(client); } }