code: def _on_rpc_done(self, future):
          
          _LOGGER.info("RPC termination has signaled manager shutdown.")
          future = _maybe_wrap_exception(future)
          thread = threading.Thread(
              name=_RPC_ERROR_THREAD_NAME, target=self.close, kwargs={"reason": future}
          )
          thread.daemon = True
          thread.start()
docs: Triggered whenever the underlying RPC terminates without recovery.
  
          This is typically triggered from one of two threads: the background
          consumer thread (when calling ``recv()`` produces a non-recoverable
          error) or the grpc management thread (when cancelling the RPC).
  
          This method is *non-blocking*. It will start another thread to deal
          with shutting everything down. This is to prevent blocking in the
          background consumer and preventing it from being ``joined()``.