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 object and task metadata. 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_task_and_object_metadata_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 object and task metadata. 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()

    def flush_shard(redis_client):
        # Flush the task table. Note that this also flushes the driver tasks
        # which may be undesirable.
        num_task_keys_deleted = 0
        for key in redis_client.scan_iter(match=TASK_PREFIX + b"*"):
            num_task_keys_deleted += redis_client.delete(key)
        print("Deleted {} task keys from Redis.".format(num_task_keys_deleted))

        # Flush the object information.
        num_object_keys_deleted = 0
        for key in redis_client.scan_iter(match=OBJECT_INFO_PREFIX + b"*"):
            num_object_keys_deleted += redis_client.delete(key)
        print("Deleted {} object info keys from Redis.".format(
            num_object_keys_deleted))

        # Flush the object locations.
        num_object_location_keys_deleted = 0
        for key in redis_client.scan_iter(match=OBJECT_LOCATION_PREFIX + b"*"):
            num_object_location_keys_deleted += redis_client.delete(key)
        print("Deleted {} object location keys from Redis.".format(
            num_object_location_keys_deleted))

    # Loop over the shards and flush all of them.
    for redis_client in ray.worker.global_state.redis_clients:
        flush_shard(redis_client)