@Override
    public SettableFuture schedule(SourceScheduler scheduler)
    {
        // Return a new future even if newDriverGroupReady has not finished.
        // Returning the same SettableFuture instance could lead to ListenableFuture retaining too many listener objects.

        // TODO: After initial scheduling, tasks would only be available after they finished at least one bucket.
        //  This is not necessarily the case if the initial scheduling covered all buckets and
        //  the available slots is not fully utilized (concurrentLifespansPerTask is large or infinite).
        //  In this case if a task failed, the recovered driver groups have to wait for tasks to be available again after finishing at least one bucket,
        //  even though by definition of concurrentLifespansPerTask they are already available.

        checkState(initialScheduled, "schedule should only be called after initial scheduling finished");
        checkState(failedTasks.size() < nodeByTaskId.size(), "All tasks have failed");

        synchronized (this) {
            newDriverGroupReady = SettableFuture.create();
            while (!availableTasks.isEmpty() && (!noPreferenceDriverGroups.isEmpty() || !nodeToPreferredDriverGroups.isEmpty())) {
                int taskId = availableTasks.dequeueInt();
                if (failedTasks.contains(taskId)) {
                    continue;
                }

                OptionalInt nextDriverGroupId = getNextDriverGroup(nodeByTaskId.get(taskId));
                if (!nextDriverGroupId.isPresent()) {
                    continue;
                }
                scheduler.startLifespan(Lifespan.driverGroup(nextDriverGroupId.getAsInt()), partitionHandles.get(nextDriverGroupId.getAsInt()));
                taskByDriverGroup[nextDriverGroupId.getAsInt()] = taskId;
                runningDriverGroupIdsByTask[taskId].add(nextDriverGroupId.getAsInt());
            }
        }
        return newDriverGroupReady;
    }
@Test
    public void testSchedule()
    {
        LifespanScheduler lifespanScheduler = getLifespanScheduler();
        TestingSourceScheduler sourceScheduler = new TestingSourceScheduler();
        lifespanScheduler.scheduleInitial(sourceScheduler);
        lifespanScheduler.onLifespanExecutionFinished(sourceScheduler.getLastStartedLifespans());
        assertEquals(sourceScheduler.getLastStartedLifespans().size(), 2);
        sourceScheduler.getLastStartedLifespans().clear();

        while (!lifespanScheduler.allLifespanExecutionFinished()) {
            lifespanScheduler.schedule(sourceScheduler);
            lifespanScheduler.onLifespanExecutionFinished(sourceScheduler.getLastStartedLifespans());
            assertEquals(sourceScheduler.getLastStartedLifespans().size(), 2);
            sourceScheduler.getLastStartedLifespans().clear();
        }
    }