code: def test_union_local_async(ray_start_regular_shared):
      def gen_fast():
          for i in range(10):
              time.sleep(0.05)
              print("PRODUCE FAST", i)
              yield i
  
      def gen_slow():
          for i in range(10):
              time.sleep(0.3)
              print("PRODUCE SLOW", i)
              yield i
  
      it1 = from_iterators([gen_fast]).for_each(lambda x: ("fast", x))
      it2 = from_iterators([gen_slow]).for_each(lambda x: ("slow", x))
      it = it1.gather_async().union(it2.gather_async())
      assert (
          repr(it) == "LocalIterator[LocalUnion[LocalIterator["
          "ParallelIterator[from_iterators[shards=1].for_each()]"
          ".gather_async()], LocalIterator[ParallelIterator["
          "from_iterators[shards=1].for_each()].gather_async()]]]"
      )
      results = list(it)
      assert all(x[0] == "slow" for x in results[-3:]), results
  
imports: import sys
  import time
  import collections
  from collections import Counter
  import pytest
  import ray
  from ray.util.iter import from_items, from_iterators, from_range, from_actors, ParallelIteratorWorker, LocalIterator
  from ray._private.test_utils import Semaphore
label_desc: Use image node_id 34 for calling a global function with example usage: test_union_local_async(ray_start_regular_shared) without return types