thread_pool_.reset();
  mutex_lock l(mu_);
  return status_;
}

void QueueRunner::UpdateStatus(const Status& status) {
  {
    mutex_lock l(mu_);
    if (!status_.ok() || status.ok() || IsQueueClosed(status)) {
      return;
    }
    status_ = status;
  }
  if (coord_) {
    coord_->ReportStatus(status);
  }
  mutex_lock l(cb_mu_);
  for (auto& cb : callbacks_) {
    cb(status);
  }

  void Run(Session* sess, const string& enqueue_op);

  // Updates the internal status; it only keeps OK or the first unexpected error
  // status.
  void UpdateStatus(const Status& status);

  bool IsQueueClosed(Status status) const {
    return queue_closed_exception_types_.count(
               static_cast<int>(status.code())) > 0;
  }

  bool IsRunning() const override { return !stopped_; }