@Test public void testElectedLeadersAsPerExpectedShufflingOrder() throws Exception { List<Lease> leases = getLeases(5, false ,false , true ); when(leaseRefresher.listLeases()).thenReturn(leases); Set<String> expectedLeaders = getExpectedLeaders(leases); for (String leader : expectedLeaders) { assertTrue(leaderDecider.isLeader(leader)); } for (Lease lease : leases) { if (!expectedLeaders.contains(lease.leaseOwner())) { assertFalse(leaderDecider.isLeader(lease.leaseOwner())); } } }
@Override public synchronized Boolean isLeader(String workerId) { if (executeConditionCheckWithReadLock(() -> CollectionUtils.isNullOrEmpty(leaders))) { electLeaders(); leaderElectionThreadPool.scheduleWithFixedDelay(this::electLeaders, ELECTION_INITIAL_DELAY_MILLIS, ELECTION_SCHEDULING_INTERVAL_MILLIS, TimeUnit.MILLISECONDS); } return executeConditionCheckWithReadLock(() -> isWorkerLeaderForShardSync(workerId)); }
DeterministicShuffleShardSyncLeaderDecider implements LeaderDecider { @Override public synchronized Boolean isLeader(String workerId) { if (executeConditionCheckWithReadLock(() -> CollectionUtils.isNullOrEmpty(leaders))) { electLeaders(); leaderElectionThreadPool.scheduleWithFixedDelay(this::electLeaders, ELECTION_INITIAL_DELAY_MILLIS, ELECTION_SCHEDULING_INTERVAL_MILLIS, TimeUnit.MILLISECONDS); } return executeConditionCheckWithReadLock(() -> isWorkerLeaderForShardSync(workerId)); } }
DeterministicShuffleShardSyncLeaderDecider implements LeaderDecider { @Override public synchronized Boolean isLeader(String workerId) { if (executeConditionCheckWithReadLock(() -> CollectionUtils.isNullOrEmpty(leaders))) { electLeaders(); leaderElectionThreadPool.scheduleWithFixedDelay(this::electLeaders, ELECTION_INITIAL_DELAY_MILLIS, ELECTION_SCHEDULING_INTERVAL_MILLIS, TimeUnit.MILLISECONDS); } return executeConditionCheckWithReadLock(() -> isWorkerLeaderForShardSync(workerId)); } DeterministicShuffleShardSyncLeaderDecider(LeaseRefresher leaseRefresher, ScheduledExecutorService leaderElectionThreadPool,
            int numPeriodicShardSyncWorkers); }
DeterministicShuffleShardSyncLeaderDecider implements LeaderDecider { @Override public synchronized Boolean isLeader(String workerId) { if (executeConditionCheckWithReadLock(() -> CollectionUtils.isNullOrEmpty(leaders))) { electLeaders(); leaderElectionThreadPool.scheduleWithFixedDelay(this::electLeaders, ELECTION_INITIAL_DELAY_MILLIS, ELECTION_SCHEDULING_INTERVAL_MILLIS, TimeUnit.MILLISECONDS); } return executeConditionCheckWithReadLock(() -> isWorkerLeaderForShardSync(workerId)); } DeterministicShuffleShardSyncLeaderDecider(LeaseRefresher leaseRefresher, ScheduledExecutorService leaderElectionThreadPool,
            int numPeriodicShardSyncWorkers); @Override synchronized Boolean isLeader(String workerId); @Override synchronized void shutdown(); }
DeterministicShuffleShardSyncLeaderDecider implements LeaderDecider { @Override public synchronized Boolean isLeader(String workerId) { if (executeConditionCheckWithReadLock(() -> CollectionUtils.isNullOrEmpty(leaders))) { electLeaders(); leaderElectionThreadPool.scheduleWithFixedDelay(this::electLeaders, ELECTION_INITIAL_DELAY_MILLIS, ELECTION_SCHEDULING_INTERVAL_MILLIS, TimeUnit.MILLISECONDS); } return executeConditionCheckWithReadLock(() -> isWorkerLeaderForShardSync(workerId)); } DeterministicShuffleShardSyncLeaderDecider(LeaseRefresher leaseRefresher, ScheduledExecutorService leaderElectionThreadPool,
            int numPeriodicShardSyncWorkers); @Override synchronized Boolean isLeader(String workerId); @Override synchronized void shutdown();  }