ZookeeperLeaderElector implements LeaderElector { @Override public boolean join() { synchronized (leaderElectionProcessRef) { if (leaderElectionProcessRef.get() == null) { leaderElectionProcessRef.set(new LeaderElectionProcess()); return true; } } return false; } @Inject  ZookeeperLeaderElector(LeaderActivator leaderActivator,
                                  CuratorService curatorService,
                                  ZookeeperPaths zookeeperPaths,
                                  MasterDescription masterDescription,
                                  TitusRuntime titusRuntime); @PreDestroy void shutdown(); @Override boolean join(); @Override boolean leaveIfNotLeader(); @Override Observable<MasterState> awaitElection();  }
@Test public void testConnectionLossWillLeadToStartupFailure() { ZookeeperConfiguration config = mock(ZookeeperConfiguration.class); when(config.getZkConnectionString()).thenReturn("127.0.0.1:44444"); ZookeeperClusterResolver clusterResolver = new DefaultZookeeperClusterResolver(config); try { CuratorServiceImpl cs = closeable.autoCloseable(new CuratorServiceImpl(config, clusterResolver, new DefaultRegistry()), CuratorServiceImpl::shutdown); cs.start(); ZookeeperLeaderElector elector = new ZookeeperLeaderElector(leaderActivator, cs, zkPaths, masterDescription, titusRuntime); elector.join(); fail("The elector should fail fast"); } catch (IllegalStateException e) { assertEquals("The cause should be from ZK connection failure", CuratorConnectionLossException.class, e.getCause().getClass()); assertTrue("The error message is unexpected: " + e.getMessage(), e.getCause().getMessage().contains("ConnectionLoss")); } }
@Test public void testLeaderCanHandleExistingPath() throws Exception { CuratorUtils.createPathIfNotExist(curator, zkPaths.getLeaderElectionPath(), true); CuratorUtils.createPathIfNotExist(curator, zkPaths.getLeaderAnnouncementPath(), true); ZookeeperLeaderElector elector = newZookeeperLeaderElector(); assertThat(elector.join()).isTrue(); awaitLeaderActivation(); }