### Question:
FixedCountScheduler implements StageScheduler { @Override public ScheduleResult schedule() { OptionalInt totalPartitions = OptionalInt.of(partitionToNode.size()); List<RemoteTask> newTasks = IntStream.range(0, partitionToNode.size()) .mapToObj(partition -> taskScheduler.scheduleTask(partitionToNode.get(partition), partition, totalPartitions)) .filter(Optional::isPresent) .map(Optional::get) .collect(toImmutableList()); return new ScheduleResult(true, newTasks, 0); } FixedCountScheduler(SqlStageExecution stage, List<InternalNode> partitionToNode); @VisibleForTesting  FixedCountScheduler(TaskScheduler taskScheduler, List<InternalNode> partitionToNode); @Override ScheduleResult schedule();  }### Answer:
@Test public void testSingleNode() { FixedCountScheduler nodeScheduler = new FixedCountScheduler( (node, partition, totalPartitions) -> Optional.of(taskFactory.createTableScanTask( new TaskId("test", 1, 1), node, ImmutableList.of(), new PartitionedSplitCountTracker(delta -> {}))), generateRandomNodes(1)); ScheduleResult result = nodeScheduler.schedule(); assertTrue(result.isFinished()); assertTrue(result.getBlocked().isDone()); assertEquals(result.getNewTasks().size(), 1); assertTrue(result.getNewTasks().iterator().next().getNodeId().equals("other 0")); }
@Test public void testMultipleNodes() { FixedCountScheduler nodeScheduler = new FixedCountScheduler( (node, partition, totalPartitions) -> Optional.of(taskFactory.createTableScanTask( new TaskId("test", 1, 1), node, ImmutableList.of(), new PartitionedSplitCountTracker(delta -> {}))), generateRandomNodes(5)); ScheduleResult result = nodeScheduler.schedule(); assertTrue(result.isFinished()); assertTrue(result.getBlocked().isDone()); assertEquals(result.getNewTasks().size(), 5); assertEquals(result.getNewTasks().stream().map(RemoteTask::getNodeId).collect(toImmutableSet()).size(), 5); }