ZookeeperMasterMonitor implements MasterMonitor { @Override public Observable<MasterDescription> getLeaderObservable() { return leaderSubject.asObservable(); } @Inject  ZookeeperMasterMonitor(ZookeeperPaths zkPaths, CuratorService curatorService, TitusRuntime titusRuntime);  ZookeeperMasterMonitor(ZookeeperPaths zkPaths,
                                  CuratorFramework curator,
                                  MasterDescription initValue,
                                  TitusRuntime titusRuntime,
                                  Scheduler scheduler); @PostConstruct void start(); @PreDestroy void shutdown(); @Override Observable<MasterDescription> getLeaderObservable(); @Override MasterDescription getLatestLeader(); @Override MasterInstance getCurrentMasterInstance(); @Override Completable updateOwnMasterInstance(MasterInstance self); @Override Observable<List<MasterInstance>> observeMasters();  }
@Test(timeout = 30_000) public void testMonitorWorksForMultipleLeaderUpdates() throws Exception { ExtTestSubscriber<MasterDescription> leaderSubscriber = new ExtTestSubscriber<>(); masterMonitor.getLeaderObservable().filter(Objects::nonNull).subscribe(leaderSubscriber); for (int i = 0; i < 5; i++) { curator.setData() .forPath(zkPaths.getLeaderAnnouncementPath(), ObjectMappers.defaultMapper().writeValueAsBytes(newMasterDescription(i))); for (int j = 0; j < 3; j++) { MasterDescription newLeader = leaderSubscriber.takeNext(5, TimeUnit.SECONDS); if (newLeader != null && newLeader.getApiPort() == i) { return; } } fail("Did not received TitusMaster update for iteration " + i); } }