public LeaderService(WorkerService workerService,
                         PulsarClient pulsarClient,
                         FunctionAssignmentTailer functionAssignmentTailer,
                         SchedulerManager schedulerManager,
                         FunctionRuntimeManager functionRuntimeManager,
                         FunctionMetaDataManager functionMetaDataManager,
                         MembershipManager membershipManager,
                         ErrorNotifier errorNotifier) {
        this.workerConfig = workerService.getWorkerConfig();
        this.pulsarClient = pulsarClient;
        this.functionAssignmentTailer = functionAssignmentTailer;
        this.schedulerManager = schedulerManager;
        this.functionRuntimeManager = functionRuntimeManager;
        this.functionMetaDataManager = functionMetaDataManager;
        this.membershipManager = membershipManager;
        this.errorNotifier = errorNotifier;
        consumerName = String.format(
                "%s:%s:%d",
                workerConfig.getWorkerId(),
                workerConfig.getWorkerHostname(),
                workerConfig.getTlsEnabled() ? workerConfig.getWorkerPortTls() : workerConfig.getWorkerPort()
        );

    }
@Test
    public void testLeaderService() throws Exception {
        MessageId messageId = new MessageIdImpl(1, 2, -1);
        when(schedulerManager.getLastMessageProduced()).thenReturn(messageId);

        assertFalse(leaderService.isLeader());
        verify(mockClient, times(1)).newConsumer();

        listenerHolder.get().becameActive(mockConsumer, 0);
        assertTrue(leaderService.isLeader());

        verify(functionMetadataManager, times(1)).getIsInitialized();
        verify(metadataManagerInitFuture, times(1)).get();
        verify(functionRuntimeManager, times(1)).getIsInitialized();
        verify(runtimeManagerInitFuture, times(1)).get();

        verify(functionMetadataManager, times(1)).acquireExclusiveWrite(any());
        verify(schedulerManager, times(1)).acquireExclusiveWrite(any());

        verify(functionAssignmentTailer, times(1)).triggerReadToTheEndAndExit();
        verify(functionAssignmentTailer, times(1)).close();
        verify(schedulerManager, times((1))).initialize(any());

        listenerHolder.get().becameInactive(mockConsumer, 0);
        assertFalse(leaderService.isLeader());

        verify(functionAssignmentTailer, times(1)).startFromMessage(messageId);
        verify(schedulerManager, times(1)).close();
        verify(functionMetadataManager, times(1)).giveupLeadership();
    }