### Question:
TopologyAwareParallelScheduler { public TaskTracker<K> submit(Process<K> process, Scope context) { log.log(Level.INFO, "parallel.scheduler.schedulingtask", process); val result = new StagedScheduleEnqueuer(process, context); workerPool.submitKernelAllocated(result); log.log(Level.INFO, "parallel.scheduler.scheduledtask", process); return result; } TopologyAwareParallelScheduler(WorkerPool workerPool); TaskTracker<K> submit(Process<K> process, Scope context);  }### Answer:
@Test @SuppressWarnings("PMD.JUnitTestsShouldIncludeAssert") void ensureDependentTaskRetrievesPredecessorValueForSimpleTaskGraph() throws ExecutionException, InterruptedException { graph.connect( new Task("a") { @Override public TaskValue run(Scope scope) { assertEquals(scope.get("test"), "hello world!", "message should be correct"); return null; } }, new Task("b") { @Override public TaskValue run(Scope scope) { scope.set("test", "hello world!"); return null; } }, DirectedGraph.outgoing("a dependsOn b")); var task = scheduler.submit(scheduleFrom(graph), scope); task.get(); }
@Test void ensureSimpleTasksWork() throws ExecutionException, InterruptedException { TaskGraph<String> g = new TaskGraph<>(); List<String> results = new ArrayList<>(); var scheduler = new TopologyAwareParallelScheduler( new ExecutorWorkerPool( Executors.newFixedThreadPool(1), Executors.newFixedThreadPool(2))); g.connect( new Task("a") { @Override public Task.TaskValue run(Scope scope) { results.add(name); return null; } }, new Task("b") { @Override public Task.TaskValue run(Scope scope) { results.add(name); return null; } }, DirectedGraph.incoming("a dependsOn b")); var process = scheduleFrom(g); var result = scheduler.submit(process, Scope.root()); result.get(); assertEquals(results.get(0), "b", "be must be first"); assertEquals(results.get(1), "a", "must be second"); }