code: def test_singleton_in_flow_in_the_middle(protocols, return_type):
      class MySingletonFlowDoc(BaseDoc):
          text: str
          num: int
  
      class InputDoc(BaseDoc):
          input: str
  
      class OutputDoc(BaseDoc):
          output: int
  
      class MyFirstSingletonIntheMiddleExecutor(Executor):
          @requests
          def foo(
              self, docs: DocList[InputDoc], **kwargs
          ) -> DocList[MySingletonFlowDoc]:
              ret = DocList[MySingletonFlowDoc]()
              for doc in docs:
                  ret.append(
                      MySingletonFlowDoc(
                          text=doc.input, num=len(doc.input)
                      )
                  )
              return ret
  
      class MySingletonIntheMiddleExecutor(Executor):
          @requests
          def foo(
              self, doc: MySingletonFlowDoc, **kwargs
          ) -> MySingletonFlowDoc:
              return MySingletonFlowDoc(text=doc.text, num=doc.num * 2)
  
      class MyLastSingletonIntheMiddleExecutor(Executor):
          @requests
          def foo(
              self, docs: DocList[MySingletonFlowDoc], **kwargs
          ) -> DocList[OutputDoc]:
              ret = DocList[OutputDoc]()
              for doc in docs:
                  ret.append(OutputDoc(output=doc.num))
              return ret
  
      ports = [random_port() for _ in protocols]
  
      flow = (
          Flow(ports=ports, protocol=protocols)
          .add(uses=MyFirstSingletonIntheMiddleExecutor)
          .add(uses=MySingletonIntheMiddleExecutor)
          .add(uses=MyLastSingletonIntheMiddleExecutor)
      )
  
      with flow:
          for port, protocol in zip(ports, protocols):
              c = Client(port=port, protocol=protocol)
              docs = c.post(
                  on="/foo",
                  inputs=InputDoc(input="hello"),
                  return_type=DocList[OutputDoc]
                  if return_type == "batch"
                  else OutputDoc,
              )
              if return_type == "batch":
                  assert docs[0].output == 2 * len("hello")
              else:
                  assert docs.output == 2 * len("hello")
  
              c = Client(port=port, protocol=protocol)
              docs = c.post(
                  on="/foo",
                  inputs=DocList[InputDoc](
                      [InputDoc(input="hello"), InputDoc(input="hello")]
                  ),
                  return_type=DocList[OutputDoc]
                  if return_type == "batch"
                  else OutputDoc,
              )
              assert isinstance(
                  docs, DocList[OutputDoc]
              )  # I have sent 2
              assert len(docs) == 2
              for doc in docs:
                  assert doc.output == 2 * len("hello")
  
file_path: jina/tests/integration/docarray_v2/test_singleton.py
imports: import pytest
  from jina import Flow, Executor, Deployment, requests, Client
  from docarray import BaseDoc, DocList
  from docarray.documents import TextDoc, ImageDoc
  from jina.helper import random_port
  from jina.excepts import RuntimeFailToStart
  import asyncio
label_desc: Use image node_id 4 for calling a global function with example usage: test_singleton_in_flow_in_the_middle(protocols, return_type) and returns: ret, MySingletonFlowDoc, ret