code: def test_zip_with_source_actor(ray_start_regular_shared):
      it = from_items([1, 2, 3, 4], num_shards=2)
      counts = collections.defaultdict(int)
      for actor, value in it.gather_async().zip_with_source_actor():
          counts[actor] += 1
      assert len(counts) == 2
      for a, count in counts.items():
          assert count == 2
  
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 4 for calling a global function with example usage: test_zip_with_source_actor(ray_start_regular_shared) without return types