def METHOD_NAME(cls):
    logger.info("Stopping cluster %s.", cls._cluster_info)
    if not cls._cluster_info:  # pragma: no cover
        return
    pg_name = cls._cluster_info["pg_name"]
    pg = cls._cluster_info["pg_group"]
    for index, bundle_spec in enumerate(pg.bundle_specs):
        # Main pool took a process.
        # If supervisor is created in the same node with worker, it will take a process too.
        n_process = int(bundle_spec["CPU"]) + 2
        for process_index in reversed(range(n_process)):
            address = process_placement_to_address(
                pg_name, index, process_index=process_index
            )
            try:
                ray_actor = ray.get_actor(address)
                if "COV_CORE_SOURCE" in os.environ:  # pragma: no cover
                    # must clean up first, or coverage info lost.
                    # must save the local reference until this is fixed:
                    # https://github.com/ray-project/ray/issues/7815
                    ray.get(ray_actor.cleanup.remote())
                ray.kill(ray_actor, no_restart=True)
                while True:
                    try:
                        ray.get(ray_actor.wait.remote(30))
                        logger.warning(
                            "Waiting actor %s to be killed.", ray_actor
                        )  # pragma: no cover
                    except ray.exceptions.RayActorError:
                        break
            except:  # noqa: E722  # nosec  # pylint: disable=bare-except
                pass
    ray.util.remove_placement_group(pg)
    cls._cluster_info = dict()
    logger.info("Stopped cluster %s.", pg_name)