@Activate
    public void activate() {
        localNodeId = clusterService.getLocalNode().id();
        leadershipService.addListener(leaderListener);

        listenerRegistry = new ListenerRegistry<>();
        eventDispatcher.addSink(WorkPartitionEvent.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");
    }
@Test
    public void testRebalanceScheduling() {
        // We have all the partitions so we'll need to relinquish some
        setUpLeadershipService(WorkPartitionManager.NUM_PARTITIONS);

        replay(leadershipService);

        partitionManager.activate();
        // Send in the event
        leaderListener.event(event);

        assertTrue(partitionManager.rebalanceScheduled.get());

        verify(leadershipService);
    }