def METHOD_NAME(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)