@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 testActivate() {
        reset(leadershipService);

        leadershipService.addListener(anyObject(LeadershipEventListener.class));

        for (int i = 0; i < WorkPartitionManager.NUM_PARTITIONS; i++) {
            expect(leadershipService.runForLeadership(ELECTION_PREFIX + i))
                .andReturn(null)
                .times(1);
        }

        replay(leadershipService);

        partitionManager.activate();

        verify(leadershipService);
    }