@Test public void testSchedule() throws Exception { List<Function.FunctionMetaData> functionMetaDataList = new LinkedList<>(); long version = 5; Function.FunctionMetaData function1 = Function.FunctionMetaData.newBuilder() .setFunctionDetails(Function.FunctionDetails.newBuilder().setName("func-1") .setNamespace("namespace-1").setTenant("tenant-1").setParallelism(1)).setVersion(version) .build(); functionMetaDataList.add(function1); doReturn(functionMetaDataList).when(functionMetaDataManager).getAllFunctionMetaData(); ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy", new ClearTextSecretsProvider(), new CollectorRegistry(), null, null); doReturn(factory).when(functionRuntimeManager).getRuntimeFactory(); Function.Assignment assignment1 = Function.Assignment.newBuilder() .setWorkerId("worker-1") .setInstance(Function.Instance.newBuilder() .setFunctionMetaData(function1).setInstanceId(0).build()) .build(); Map<String, Map<String, Function.Assignment>> currentAssignments = new HashMap<>(); Map<String, Function.Assignment> assignmentEntry1 = new HashMap<>(); assignmentEntry1.put(FunctionCommon.getFullyQualifiedInstanceId(assignment1.getInstance()), assignment1); currentAssignments.put("worker-1", assignmentEntry1); doReturn(currentAssignments).when(functionRuntimeManager).getCurrentAssignments(); List<WorkerInfo> workerInfoList = new LinkedList<>(); workerInfoList.add(WorkerInfo.of("worker-1", "workerHostname-1", 5000)); doReturn(workerInfoList).when(membershipManager).getCurrentMembership(); doReturn(false).when(leaderService).isLeader(); callSchedule(); verify(producer, times(0)).sendAsync(any()); doReturn(true).when(leaderService).isLeader(); callSchedule(); List<Invocation> invocations = getMethodInvocationDetails(schedulerManager, SchedulerManager.class.getDeclaredMethod("invokeScheduler")); Assert.assertEquals(invocations.size(), 1); verify(errorNotifier, times(0)).triggerError(any()); }
public Future<?> schedule() { return scheduleInternal(() -> { workerStatsManager.scheduleTotalExecTimeStart(); invokeScheduler(); workerStatsManager.scheduleTotalExecTimeEnd(); }, "Encountered error when invoking scheduler"); }
SchedulerManager implements AutoCloseable { public Future<?> schedule() { return scheduleInternal(() -> { workerStatsManager.scheduleTotalExecTimeStart(); invokeScheduler(); workerStatsManager.scheduleTotalExecTimeEnd(); }, "Encountered error when invoking scheduler"); } }
SchedulerManager implements AutoCloseable { public Future<?> schedule() { return scheduleInternal(() -> { workerStatsManager.scheduleTotalExecTimeStart(); invokeScheduler(); workerStatsManager.scheduleTotalExecTimeEnd(); }, "Encountered error when invoking scheduler"); } SchedulerManager(WorkerConfig workerConfig,
                            PulsarClient pulsarClient,
                            PulsarAdmin admin,
                            WorkerStatsManager workerStatsManager,
                            ErrorNotifier errorNotifier); }
SchedulerManager implements AutoCloseable { public Future<?> schedule() { return scheduleInternal(() -> { workerStatsManager.scheduleTotalExecTimeStart(); invokeScheduler(); workerStatsManager.scheduleTotalExecTimeEnd(); }, "Encountered error when invoking scheduler"); } SchedulerManager(WorkerConfig workerConfig,
                            PulsarClient pulsarClient,
                            PulsarAdmin admin,
                            WorkerStatsManager workerStatsManager,
                            ErrorNotifier errorNotifier); synchronized void initialize(); Future<?> schedule(); Future<?> rebalanceIfNotInprogress(); @Override synchronized void close(); }
SchedulerManager implements AutoCloseable { public Future<?> schedule() { return scheduleInternal(() -> { workerStatsManager.scheduleTotalExecTimeStart(); invokeScheduler(); workerStatsManager.scheduleTotalExecTimeEnd(); }, "Encountered error when invoking scheduler"); } SchedulerManager(WorkerConfig workerConfig,
                            PulsarClient pulsarClient,
                            PulsarAdmin admin,
                            WorkerStatsManager workerStatsManager,
                            ErrorNotifier errorNotifier); synchronized void initialize(); Future<?> schedule(); Future<?> rebalanceIfNotInprogress(); @Override synchronized void close(); static final String HEARTBEAT_TENANT; static final String HEARTBEAT_NAMESPACE; }