@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(functionAssignmentTailer, times(1)).triggerReadToTheEndAndExit(); verify(functionAssignmentTailer, times(1)).close(); verify(schedulerManager, times((1))).initialize(); listenerHolder.get().becameInactive(mockConsumer, 0); assertFalse(leaderService.isLeader()); verify(functionAssignmentTailer, times(1)).startFromMessage(messageId); verify(schedulerManager, times(1)).close(); }
public LeaderService(WorkerService workerService, PulsarClient pulsarClient, FunctionAssignmentTailer functionAssignmentTailer, SchedulerManager schedulerManager, FunctionRuntimeManager functionRuntimeManager, FunctionMetaDataManager functionMetaDataManager, ErrorNotifier errorNotifier) { this.workerConfig = workerService.getWorkerConfig(); this.pulsarClient = pulsarClient; this.functionAssignmentTailer = functionAssignmentTailer; this.schedulerManager = schedulerManager; this.functionRuntimeManager = functionRuntimeManager; this.functionMetaDataManager = functionMetaDataManager; this.errorNotifier = errorNotifier; consumerName = String.format( "%s:%s:%d", workerConfig.getWorkerId(), workerConfig.getWorkerHostname(), workerConfig.getWorkerPort() ); }
LeaderService implements AutoCloseable, ConsumerEventListener { public LeaderService(WorkerService workerService, PulsarClient pulsarClient, FunctionAssignmentTailer functionAssignmentTailer, SchedulerManager schedulerManager, FunctionRuntimeManager functionRuntimeManager, FunctionMetaDataManager functionMetaDataManager, ErrorNotifier errorNotifier) { this.workerConfig = workerService.getWorkerConfig(); this.pulsarClient = pulsarClient; this.functionAssignmentTailer = functionAssignmentTailer; this.schedulerManager = schedulerManager; this.functionRuntimeManager = functionRuntimeManager; this.functionMetaDataManager = functionMetaDataManager; this.errorNotifier = errorNotifier; consumerName = String.format( "%s:%s:%d", workerConfig.getWorkerId(), workerConfig.getWorkerHostname(), workerConfig.getWorkerPort() ); } }
LeaderService implements AutoCloseable, ConsumerEventListener { public LeaderService(WorkerService workerService, PulsarClient pulsarClient, FunctionAssignmentTailer functionAssignmentTailer, SchedulerManager schedulerManager, FunctionRuntimeManager functionRuntimeManager, FunctionMetaDataManager functionMetaDataManager, ErrorNotifier errorNotifier) { this.workerConfig = workerService.getWorkerConfig(); this.pulsarClient = pulsarClient; this.functionAssignmentTailer = functionAssignmentTailer; this.schedulerManager = schedulerManager; this.functionRuntimeManager = functionRuntimeManager; this.functionMetaDataManager = functionMetaDataManager; this.errorNotifier = errorNotifier; consumerName = String.format( "%s:%s:%d", workerConfig.getWorkerId(), workerConfig.getWorkerHostname(), workerConfig.getWorkerPort() ); } LeaderService(WorkerService workerService,
                         PulsarClient pulsarClient,
                         FunctionAssignmentTailer functionAssignmentTailer,
                         SchedulerManager schedulerManager,
                         FunctionRuntimeManager functionRuntimeManager,
                         FunctionMetaDataManager functionMetaDataManager,
                         ErrorNotifier errorNotifier); }
LeaderService implements AutoCloseable, ConsumerEventListener { public LeaderService(WorkerService workerService, PulsarClient pulsarClient, FunctionAssignmentTailer functionAssignmentTailer, SchedulerManager schedulerManager, FunctionRuntimeManager functionRuntimeManager, FunctionMetaDataManager functionMetaDataManager, ErrorNotifier errorNotifier) { this.workerConfig = workerService.getWorkerConfig(); this.pulsarClient = pulsarClient; this.functionAssignmentTailer = functionAssignmentTailer; this.schedulerManager = schedulerManager; this.functionRuntimeManager = functionRuntimeManager; this.functionMetaDataManager = functionMetaDataManager; this.errorNotifier = errorNotifier; consumerName = String.format( "%s:%s:%d", workerConfig.getWorkerId(), workerConfig.getWorkerHostname(), workerConfig.getWorkerPort() ); } LeaderService(WorkerService workerService,
                         PulsarClient pulsarClient,
                         FunctionAssignmentTailer functionAssignmentTailer,
                         SchedulerManager schedulerManager,
                         FunctionRuntimeManager functionRuntimeManager,
                         FunctionMetaDataManager functionMetaDataManager,
                         ErrorNotifier errorNotifier); void start(); @Override void becameActive(Consumer<?> consumer, int partitionId); @Override synchronized void becameInactive(Consumer<?> consumer, int partitionId); synchronized boolean isLeader(); @Override void close(); }
LeaderService implements AutoCloseable, ConsumerEventListener { public LeaderService(WorkerService workerService, PulsarClient pulsarClient, FunctionAssignmentTailer functionAssignmentTailer, SchedulerManager schedulerManager, FunctionRuntimeManager functionRuntimeManager, FunctionMetaDataManager functionMetaDataManager, ErrorNotifier errorNotifier) { this.workerConfig = workerService.getWorkerConfig(); this.pulsarClient = pulsarClient; this.functionAssignmentTailer = functionAssignmentTailer; this.schedulerManager = schedulerManager; this.functionRuntimeManager = functionRuntimeManager; this.functionMetaDataManager = functionMetaDataManager; this.errorNotifier = errorNotifier; consumerName = String.format( "%s:%s:%d", workerConfig.getWorkerId(), workerConfig.getWorkerHostname(), workerConfig.getWorkerPort() ); } LeaderService(WorkerService workerService,
                         PulsarClient pulsarClient,
                         FunctionAssignmentTailer functionAssignmentTailer,
                         SchedulerManager schedulerManager,
                         FunctionRuntimeManager functionRuntimeManager,
                         FunctionMetaDataManager functionMetaDataManager,
                         ErrorNotifier errorNotifier); void start(); @Override void becameActive(Consumer<?> consumer, int partitionId); @Override synchronized void becameInactive(Consumer<?> consumer, int partitionId); synchronized boolean isLeader(); @Override void close();  }