}
    }

    private void monitoredNodeReady(Node node) {
        lockCoordinator.assertHasStateLock();
        node.updateAllDependenciesComplete();
        maybeNodeReady(node);
    }

    @Override
    public void finishedExecuting(Node node, @Nullable Throwable failure) {
        lockCoordinator.assertHasStateLock();
        try {
            runningNodes.remove(node);

            if (failure != null) {

        }

        public void cancelExecution() {
            coordinationService.assertHasStateLock();
            for (PlanDetails details : queues) {
                details.source.cancelExecution();
            }
        }

        public void abortAllAndFail(Throwable t) {
            coordinationService.assertHasStateLock();
            for (PlanDetails details : queues) {

     */
    <T> T withStateLock(Supplier<T> action);

    /**
     * Notify other threads about changes to resource locks.
     */
    void notifyStateChange();

    void assertHasStateLock();

    /**
     * Adds a listener that is notified when a lock is released. Called while the state lock is held.
     */
    void addLockReleaseListener(Action<ResourceLock> listener);

            if (!releaseHandlers.isEmpty()) {
                throw new IllegalStateException("Some lock release listeners have not been removed.");
            }
        }
    }

    @Override
    public void assertHasStateLock() {
        synchronized (lock) {
            if (getCurrent() == null) {
                throw new IllegalStateException();
            }
        }
    }

    @Override