This removes some critical state from the Redis shards.

    In a multitenant environment, this will flush metadata for all jobs, which
    may be undesirable.

    This removes all of the metadata for finished tasks. This can be used to
    try to address out-of-memory errors caused by the accumulation of metadata
    in Redis. However, after running this command, fault tolerance will most
    likely not work.

def flush_finished_tasks_unsafe():
    """This removes some critical state from the Redis shards.

    In a multitenant environment, this will flush metadata for all jobs, which
    may be undesirable.

    This removes all of the metadata for finished tasks. This can be used to
    try to address out-of-memory errors caused by the accumulation of metadata
    in Redis. However, after running this command, fault tolerance will most
    likely not work.
    """
    ray.worker.global_worker.check_connected()

    for shard_index in range(len(ray.global_state.redis_clients)):
        _flush_finished_tasks_unsafe_shard(shard_index)