code: def test_pending_request(
      port_generator, failure_in_executor, protocol
  ):
      port0 = port_generator()
      port1 = port_generator()
      runtime_name = "gateway/rep-0"
  
      class SlowExecutor(Executor):
          @requests
          def foo(self, docs, **kwargs):
              time.sleep(5)
              if failure_in_executor:
                  raise Exception
  
      with Flow(
          monitoring=True, port_monitoring=port0, protocol=protocol
      ).add(uses=SlowExecutor, port_monitoring=port1) as f:
  
          def _send_request():
              f.search(
                  inputs=DocumentArray.empty(size=1),
                  continue_on_error=True,
              )
  
          p_send = multiprocessing.Process(
              target=_send_request, daemon=True
          )
          _assert_pending_value("0.0", runtime_name, port0)
  
          p_send.start()
          time.sleep(3)
  
          _assert_pending_value("1.0", runtime_name, port0)
  
          p_send.join()
          assert p_send.exitcode == 0
          _assert_pending_value("0.0", runtime_name, port0)
  
file_path: jina/tests/integration/monitoring/test_manual_flow.py
imports: import multiprocessing
  import time
  import pytest
  import requests
  from docarray import Document, DocumentArray
  from jina import Executor, Flow, requests
  from jina.parsers import set_gateway_parser
  from jina.serve.runtimes.asyncio import AsyncNewLoopRuntime
  from jina.serve.runtimes.worker.request_handling import WorkerRequestHandler
  from jina.serve.runtimes.gateway.request_handling import GatewayRequestHandler
  from jina.serve.runtimes.servers import BaseServer
  from tests.helper import _generate_pod_args
label_desc: Use image node_id 8 for calling a global function with example usage: test_pending_request(port_generator, failure_in_executor, protocol) without return types