code: def test_streaming_generator_bad_exception_not_failing(
      shutdown_only, capsys
  ):
      """This test verifies when a return value cannot be stored
          e.g., because it holds a lock) if it handles failures gracefully.
  
      Previously, when it happens, there was a check failure. This verifies
      the check failure doesn't happen anymore.
      """
      ray.init()
  
      class UnserializableException(Exception):
          def __init__(self):
              self.lock = threading.Lock()
  
      @ray.remote
      def f():
          raise UnserializableException
          yield 1  # noqa
  
      for ref in f.remote():
          with pytest.raises(ray.exceptions.RayTaskError):
              ray.get(ref)
      captured = capsys.readouterr()
      lines = captured.err.strip().split("\n")
  
      # Verify check failure doesn't happen because we handle the error
      # properly.
      for line in lines:
          assert "Check failed:" not in line
  
imports: import asyncio
  import pytest
  import numpy
  import sys
  import time
  import threading
  import gc
  from unittest.mock import patch, Mock
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.experimental.state.api import list_objects
  from ray._raylet import ObjectRefGenerator, ObjectRefStreamEndOfStreamError
  from ray.cloudpickle import dumps
  from ray.exceptions import WorkerCrashedError
label_desc: Use image node_id 5 for calling a global function with example usage: test_streaming_generator_bad_exception_not_failing(shutdown_only, capsys) without return types