code: def flush_task_and_object_metadata_unsafe():
      
      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)
docs: 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.