code: def test_read_text_remote_args(ray_start_cluster, tmp_path):
      cluster = ray_start_cluster
      cluster.add_node(
          resources={"foo": 100},
          num_cpus=1,
          _system_config={"max_direct_call_object_size": 0},
      )
      cluster.add_node(resources={"bar": 100}, num_cpus=1)
  
      ray.init(cluster.address)
  
      @ray.remote
      def get_node_id():
          return ray.get_runtime_context().get_node_id()
  
      bar_node_id = ray.get(
          get_node_id.options(resources={"bar": 1}).remote()
      )
  
      path = os.path.join(tmp_path, "test_text")
      os.mkdir(path)
      with open(os.path.join(path, "file1.txt"), "w") as f:
          f.write("hello\n")
          f.write("world")
      with open(os.path.join(path, "file2.txt"), "w") as f:
          f.write("goodbye")
  
      ds = ray.data.read_text(
          path, parallelism=2, ray_remote_args={"resources": {"bar": 1}}
      )
  
      blocks = ds.get_internal_block_refs()
      ray.wait(blocks, num_returns=len(blocks), fetch_local=False)
      location_data = ray.experimental.get_object_locations(blocks)
      locations = []
      for block in blocks:
          locations.extend(location_data[block]["node_ids"])
      assert set(locations) == {bar_node_id}, locations
      assert sorted(_to_lines(ds.take())) == [
          "goodbye",
          "hello",
          "world",
      ]
  
file_path: ray/python/ray/data/tests/test_text.py
imports: import os
  import pandas
  import pytest
  import ray
  from ray.data.datasource import BaseFileMetadataProvider, FastFileMetadataProvider, Partitioning, PartitionStyle, PathPartitionFilter
  from ray.data.tests.conftest import *
  from ray.data.tests.mock_http_server import *
  from ray.data.tests.test_partitioning import PathPartitionEncoder
  from ray.data.tests.util import Counter
  from ray.tests.conftest import *
label_desc: Use image node_id 8 for calling a global function with example usage: test_read_text_remote_args(ray_start_cluster, tmp_path) and returns: ray