Multi-path Flow breaks when one path skips the request and parallel>1

[nan-wang]

Describe the bug
When using parallel>1 and the target_peapod is not the one with multiple parallel, it keeps hanging.

from jina import Document, Executor, Flow, requests
class UpExecutor(Executor):
    @requests
    def foo(self, **kwargs):
        print(f'hello, up')
class DownExecutor(Executor):
    @requests
    def foo(self, **kwargs):
        print(f'hello, down')
f = (Flow()
     .add(uses=UpExecutor, parallel=3, name='up')
     .add(uses=DownExecutor, needs='gateway', name='down')
     .join(needs=['up', 'down']))
with f:
    f.post(on='/foo', target_peapod='down', inputs=Document())

Describe how you solve it

---

Environment

- jina 2.0.6
- jina-proto 0.0.84
- jina-vcs-tag (unset)
- libzmq 4.3.4
- pyzmq 1.20.3
- protobuf 3.17.0
- proto-backend cpp
- grpcio 1.37.1
- pyyaml 5.4.1
- python 3.8.5
- platform Darwin
- platform-release 20.3.0
- platform-version Darwin Kernel Version 20.3.0: Thu Jan 21 00:07:06 PST 2021;
  root:xnu-7195.81.3~1/RELEASE_X86_64
- architecture x86_64
- processor i386
- uid 154505279077031
- session-id 55bd87ac-e185-11eb-b4a4-8c8590467aa7
- uptime 2021-07-10T21:47:14.809177
- ci-vendor (unset)
* JINA_ARRAY_QUANT (unset)
* JINA_CONTROL_PORT (unset)
* JINA_DEFAULT_HOST (unset)
* JINA_DISABLE_UVLOOP (unset)
* JINA_FULL_CLI (unset)
* JINA_HUBBLE_REGISTRY (unset)
* JINA_HUB_CACHE_DIR (unset)
* JINA_HUB_ROOT (unset)
* JINA_LOG_CONFIG (unset)
* JINA_LOG_ID (unset)
* JINA_LOG_LEVEL DEBUG
* JINA_LOG_NO_COLOR (unset)
* JINA_LOG_WORKSPACE (unset)
* JINA_OPTIMIZER_TRIAL_WORKSPACE(unset)
* JINA_POD_NAME (unset)
* JINA_RANDOM_PORT_MAX (unset)
* JINA_RANDOM_PORT_MIN (unset)
* JINA_VCS_VERSION (unset)

Screenshots