code: def test_complex_flow(disable_reduce):
      f = (
          Flow()
          .add(name="first", uses=SimpleAddExecutor, needs=["gateway"])
          .add(
              name="forth",
              uses=SimpleAddExecutor,
              needs=["first"],
              shards=2,
          )
          .add(
              name="second_shards_needs",
              uses=SimpleAddExecutor,
              needs=["gateway"],
              shards=2,
          )
          .add(
              name="third",
              uses=SimpleAddExecutor,
              shards=3,
              needs=["second_shards_needs"],
          )
          .add(
              name="merger",
              uses=MergeDocsExecutor,
              needs=["forth", "third"],
              disable_reduce=disable_reduce,
          )
      )
  
      with f:
          docs = f.post(on="/index", inputs=[Document(text="1")])
      assert len(docs) == 6 if disable_reduce else 5
  
imports: from docarray import Document, DocumentArray
  from jina import Executor, Flow, requests
  import pytest
label_desc: Use image node_id 4 for calling a global function with example usage: test_complex_flow(disable_reduce) without return types