@Test public void testRebalanceScheduling() { setUpLeadershipService(IntentPartitionManager.NUM_PARTITIONS); replay(leadershipService); partitionManager.activate(); leaderListener.event(event); assertTrue(partitionManager.rebalanceScheduled.get()); verify(leadershipService); }
@Activate public void activate() { localNodeId = clusterService.getLocalNode().id(); leadershipService.addListener(leaderListener); listenerRegistry = new ListenerRegistry<>(); eventDispatcher.addSink(IntentPartitionEvent.class, listenerRegistry); for (int i = 0; i < NUM_PARTITIONS; i++) { leadershipService.runForLeadership(getPartitionPath(i)); log.debug("Registered to run for {}", getPartitionPath(i)); } executor.scheduleAtFixedRate(() -> scheduleRebalance(0), 0, CHECK_PARTITION_BALANCE_PERIOD_SEC, TimeUnit.SECONDS); log.info("Started"); }
IntentPartitionManager implements IntentPartitionService { @Activate public void activate() { localNodeId = clusterService.getLocalNode().id(); leadershipService.addListener(leaderListener); listenerRegistry = new ListenerRegistry<>(); eventDispatcher.addSink(IntentPartitionEvent.class, listenerRegistry); for (int i = 0; i < NUM_PARTITIONS; i++) { leadershipService.runForLeadership(getPartitionPath(i)); log.debug("Registered to run for {}", getPartitionPath(i)); } executor.scheduleAtFixedRate(() -> scheduleRebalance(0), 0, CHECK_PARTITION_BALANCE_PERIOD_SEC, TimeUnit.SECONDS); log.info("Started"); } }
IntentPartitionManager implements IntentPartitionService { @Activate public void activate() { localNodeId = clusterService.getLocalNode().id(); leadershipService.addListener(leaderListener); listenerRegistry = new ListenerRegistry<>(); eventDispatcher.addSink(IntentPartitionEvent.class, listenerRegistry); for (int i = 0; i < NUM_PARTITIONS; i++) { leadershipService.runForLeadership(getPartitionPath(i)); log.debug("Registered to run for {}", getPartitionPath(i)); } executor.scheduleAtFixedRate(() -> scheduleRebalance(0), 0, CHECK_PARTITION_BALANCE_PERIOD_SEC, TimeUnit.SECONDS); log.info("Started"); }  }
IntentPartitionManager implements IntentPartitionService { @Activate public void activate() { localNodeId = clusterService.getLocalNode().id(); leadershipService.addListener(leaderListener); listenerRegistry = new ListenerRegistry<>(); eventDispatcher.addSink(IntentPartitionEvent.class, listenerRegistry); for (int i = 0; i < NUM_PARTITIONS; i++) { leadershipService.runForLeadership(getPartitionPath(i)); log.debug("Registered to run for {}", getPartitionPath(i)); } executor.scheduleAtFixedRate(() -> scheduleRebalance(0), 0, CHECK_PARTITION_BALANCE_PERIOD_SEC, TimeUnit.SECONDS); log.info("Started"); }  @Activate void activate(); @Deactivate void deactivate(); @Override boolean isMine(Key intentKey); @Override NodeId getLeader(Key intentKey); @Override void addListener(IntentPartitionEventListener listener); @Override void removeListener(IntentPartitionEventListener listener); }
IntentPartitionManager implements IntentPartitionService { @Activate public void activate() { localNodeId = clusterService.getLocalNode().id(); leadershipService.addListener(leaderListener); listenerRegistry = new ListenerRegistry<>(); eventDispatcher.addSink(IntentPartitionEvent.class, listenerRegistry); for (int i = 0; i < NUM_PARTITIONS; i++) { leadershipService.runForLeadership(getPartitionPath(i)); log.debug("Registered to run for {}", getPartitionPath(i)); } executor.scheduleAtFixedRate(() -> scheduleRebalance(0), 0, CHECK_PARTITION_BALANCE_PERIOD_SEC, TimeUnit.SECONDS); log.info("Started"); }  @Activate void activate(); @Deactivate void deactivate(); @Override boolean isMine(Key intentKey); @Override NodeId getLeader(Key intentKey); @Override void addListener(IntentPartitionEventListener listener); @Override void removeListener(IntentPartitionEventListener listener);  }