feat: wait for floating Executor tasks

docs/fundamentals/flow/add-executors.md

@@ -187,6 +187,120 @@ for further details
 ```
 
 
+(floating-executors)=
+### Floating Executors
+
+Normally, all Executors form a pipeline that handles and transforms a given request until it is finally returned to the Client.
+
+However, you can add Executors that do not feed their outputs back to the pipeline. Therefore, this output will not form the response for the Client.
+
+This way of adding Executors in a Flow can be used for asynchronous backround tasks that may take some time and that are not needed for the response of the service you are building. For instance,
+logging specific information in external services, storing partial results, etc.
+
+Those Executors are marked with the `floating` keyword when added to a `Flow`
+
+```python
+import time
+from jina import Flow, Executor, requests, DocumentArray
+
+
+class FastChangingExecutor(Executor):
+    @requests()
+    def foo(self, docs, **kwargs):
+        for doc in docs:
+            doc.text = 'Hello World'
+
+
+class SlowChangingExecutor(Executor):
+    @requests()
+    def foo(self, docs, **kwargs):
+        time.sleep(2)
+        print(f' Received {docs.texts}')
+        for doc in docs:
+            doc.text = 'Change the document but will not affect response'
+
+
+f = (
+    Flow()
+    .add(name='executor0', uses=FastChangingExecutor)
+    .add(
+        name='floating_executor',
+        uses=SlowChangingExecutor,
+        needs=['gateway'],
+        floating=True,
+    )
+)
+with f:
+    f.post(on='/endpoint', inputs=DocumentArray.empty(1))  # we need to send a first
+    start_time = time.time()
+    response = f.post(on='/endpoint', inputs=DocumentArray.empty(2))
+    end_time = time.time()
+    print(f' Response time took {end_time - start_time}s')
+    print(f' {response.texts}')
+```
+
+```text
+ Response time took 0.011997222900390625s
+ ['Hello World', 'Hello World']
+ Received ['Hello World', 'Hello World']
+```
+
+In this example you can see how the response is returned without waiting for the `floating` Executor to complete. However, the Flow is not closed until
+the request has been handled also by it.
+
+
+You can plot the Flow and observe how the Executor is floating disconnected from the **Gateway**.
+
+```{figure} flow_floating.svg
+:width: 70%
+
+```
+A floating Executor can never come before a non-floating Executor in the {ref}`topology <flow-complex-topologies>` of your Flow.
+
+This leads to the following behaviors:
+
+- **Implicit reordering**: When adding a non-floating Executor after a floating Executor without specifying its `needs` parameter, the non-floating Executor is chained after the previous non-floating one.
+```python
+from jina import Flow
+
+f = Flow().add().add(name='middle', floating=True).add()
+f.plot()
+```
+
+```{figure} flow_middle_1.svg
+:width: 70%
+
+```
+
+- **Chaining floating Executors**: If you want to chain more than one floating Executor, you need to add all of them with the `floating` flag, and explicitly specify the `needs` argument.
+
+```python
+from jina import Flow
+
+f = Flow().add().add(name='middle', floating=True).add(needs=['middle'], floating=True)
+f.plot()
+```
+
+```{figure} flow_chain_floating.svg
+:width: 70%
+
+```
+
+- **Overriding of `floating` flag**: If you try to add a floating Executor as part of `needs` parameter of a non-floating Executor, then the floating Executor is not considered floating anymore.
+
+```python
+from jina import Flow
+
+f = Flow().add().add(name='middle', floating=True).add(needs=['middle'])
+f.plot()
+```
+
+```{figure} flow_cancel_floating.svg
+:width: 70%
+
+```
+
+
 ## Set configs
 You can set and override {class}`~jina.Executor` configs when adding them into a {class}`~jina.Flow`.