For both this and the local scheduler, does this work?  It seems like you still need to call error() (or better, abort()) on the task set manager, so that eventually taskSetFailed() gets called on the DAGScheduler and all of the state gets cleaned up