Repository: jina-ai/serve
Pull request: fix: fix bug inheritance, requests nested dict
**Goals:**
Makes `requests` Executor class method a nested dict including the class name to support inheritance


Problem statement:
bug: wrong inherited Executor called
At Jina now, we are using the inheritance of executors since it allows us to constrain certain behavior on a group of executors.
For instance, we have a BaseIndexer that makes sure all derived indexers like Qdrant/Elastic have the same interface. And if we change the interface, all Indexers are updated with the new behavior.

Now we are experiencing the following issue. 
This is our code (simplified):
```python
from docarray import Document
from jina import Flow

from jina import Executor, requests

class BaseExecutor(Executor):
    @requests()
    def x(self, *args, **kwargs):
        pass


class EnhancedExecutor(BaseExecutor):
    @requests()
    def y(
        self,  *args, **kwargs
    ):
        print('enhanced was called')

with Flow().add(uses=BaseExecutor) as f:
    f.index(inputs=Document())
```

Output:
```
enhanced was called
```

As you can see the `EnhancedExecutor` is called even if the flow is just using the `BaseExecutor`.

UAC
- [ ] Please support inheritance for Executors.

Hints from issue discussion:
Can u try without having one of them called `BaseExecutor`? Since this is a special class in our Jina inheritance tree I want to make sure that it is just a very bad coincidence. Does it also happen with `ExecutorBase`and `ExecutorInherited`?
also, can u run here?

```python
exec = Executor.load_executor(ExecutorInherited)
print(type(exec))
```
This is even more funny, there is something odd with our `requests` decorator:

```python
from docarray import Document
from jina import Flow

from jina import Executor, requests


class A(Executor):
    @requests()
    def x(self, *args, **kwargs):
        pass


class EnhancedExecutor(A):
    @requests()
    def y(
            self, *args, **kwargs
    ):
        print(f'enhanced was called => {type(self)} => {isinstance(self, EnhancedExecutor)}')


with Flow().add(uses=A) as f:
    f.index(inputs=Document())
```

You see that the call actually gets an instance of `A` and not an instance of `EnhancedExecutor`

```shell
enhanced was called => <class '__main__.A'> => False
```
Yes, this is also what we experienced. Just the decorator is incorrectly attached or something

you can see:

```python
print(f' A => {A.requests}')
```
```shell
 A => {'/default': <function EnhancedExecutor.y at 0x7f51dd51c5f0>}
```
The issue is do to this:

```python
class A:
    c_attr = {}


class B(A):
    pass


B.c_attr['hey'] = 'ho'

print(f' A {A.c_attr}')
```

```shell
 A {'hey': 'ho'}
```

`requests` is a class attribute and it is shared

Commit message: test: add regression test

Patch to review:
diff --git a/jina/serve/executors/__init__.py b/jina/serve/executors/__init__.py
index 51adfbd795c0b..b1a8d94802f17 100644
--- a/jina/serve/executors/__init__.py
+++ b/jina/serve/executors/__init__.py
@@ -18,7 +18,7 @@
 from jina.serve.helper import store_init_kwargs, wrap_func
 from jina.serve.instrumentation import MetricsTimer
 
-if TYPE_CHECKING: # pragma: no cover
+if TYPE_CHECKING:  # pragma: no cover
     from opentelemetry.context.context import Context
 
 __dry_run_endpoint__ = '_jina_dry_run_'
@@ -55,7 +55,7 @@ def register_class(cls):
             arg_spec = inspect.getfullargspec(cls.__init__)
 
             if not arg_spec.varkw and not __args_executor_init__.issubset(
-                arg_spec.args
+                    arg_spec.args
             ):
                 raise TypeError(
                     f'{cls.__init__} does not follow the full signature of `Executor.__init__`, '
@@ -115,12 +115,12 @@ def __init__(awesomeness=5):
     """
 
     def __init__(
-        self,
-        metas: Optional[Dict] = None,
-        requests: Optional[Dict] = None,
-        runtime_args: Optional[Dict] = None,
-        workspace: Optional[str] = None,
-        **kwargs,
+            self,
+            metas: Optional[Dict] = None,
+            requests: Optional[Dict] = None,
+            runtime_args: Optional[Dict] = None,
+            workspace: Optional[str] = None,
+            **kwargs,
     ):
         """`metas` and `requests` are always auto-filled with values from YAML config.
 
@@ -158,12 +158,12 @@ def _add_runtime_args(self, _runtime_args: Optional[Dict]):
 
     def _init_monitoring(self):
         if (
-            hasattr(self.runtime_args, 'metrics_registry')
-            and self.runtime_args.metrics_registry
+                hasattr(self.runtime_args, 'metrics_registry')
+                and self.runtime_args.metrics_registry
         ):
             with ImportExtensions(
-                required=True,
-                help_text='You need to install the `prometheus_client` to use the montitoring functionality of jina',
+                    required=True,
+                    help_text='You need to install the `prometheus_client` to use the montitoring functionality of jina',
             ):
                 from prometheus_client import Summary
 
@@ -214,9 +214,20 @@ def _init_instrumentation(self, _runtime_args: Optional[Dict] = None):
             self.meter_provider = None
             self.meter = None
 
+    @property
+    def requests(self):
+        """
+        Get the request dictionary corresponding to this specific class
+
+        :return: Returns the requests corresponding to the specific Executor instance class
+        """
+        if not hasattr(self, 'requests_by_class'):
+            self.requests_by_class = {}
+        if self.__class__.__name__ not in self.requests_by_class:
+            self.requests_by_class[self.__class__.__name__] = {}
+        return self.requests_by_class[self.__class__.__name__]
+
     def _add_requests(self, _requests: Optional[Dict]):
-        if not hasattr(self, 'requests'):
-            self.requests = {}
         if _requests:
             func_names = {f.__name__: e for e, f in self.requests.items()}
             for endpoint, func in _requests.items():
@@ -278,7 +289,7 @@ def _add_metas(self, _metas: Optional[Dict]):
                 if not hasattr(target, k):
                     if isinstance(v, str):
                         if not (
-                            env_var_regex.findall(v) or internal_var_regex.findall(v)
+                                env_var_regex.findall(v) or internal_var_regex.findall(v)
                         ):
                             setattr(target, k, v)
                         else:
@@ -323,16 +334,17 @@ async def __acall__(self, req_endpoint: str, **kwargs):
         # noqa: DAR102
         # noqa: DAR201
         """
+
         if req_endpoint in self.requests:
             return await self.__acall_endpoint__(req_endpoint, **kwargs)
         elif __default_endpoint__ in self.requests:
             return await self.__acall_endpoint__(__default_endpoint__, **kwargs)
 
     async def __acall_endpoint__(
-        self, req_endpoint, tracing_context: Optional['Context'], **kwargs
+            self, req_endpoint, tracing_context: Optional['Context'], **kwargs
     ):
         async def exec_func(
-            summary, histogram, histogram_metric_labels, tracing_context
+                summary, histogram, histogram_metric_labels, tracing_context
         ):
             with MetricsTimer(summary, histogram, histogram_metric_labels):
                 if iscoroutinefunction(func):
@@ -390,10 +402,10 @@ def workspace(self) -> Optional[str]:
         :return: returns the workspace of the current shard of this Executor.
         """
         workspace = (
-            getattr(self.runtime_args, 'workspace', None)
-            or getattr(self.metas, 'workspace')
-            or self._init_workspace
-            or __cache_path__
+                getattr(self.runtime_args, 'workspace', None)
+                or getattr(self.metas, 'workspace')
+                or self._init_workspace
+                or __cache_path__
         )
         if workspace:
             complete_workspace = os.path.join(workspace, self.metas.name)
@@ -416,13 +428,13 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
     @classmethod
     def from_hub(
-        cls: Type[T],
-        uri: str,
-        context: Optional[Dict[str, Any]] = None,
-        uses_with: Optional[Dict] = None,
-        uses_metas: Optional[Dict] = None,
-        uses_requests: Optional[Dict] = None,
-        **kwargs,
+            cls: Type[T],
+            uri: str,
+            context: Optional[Dict[str, Any]] = None,
+            uses_with: Optional[Dict] = None,
+            uses_metas: Optional[Dict] = None,
+            uses_requests: Optional[Dict] = None,
+            **kwargs,
     ) -> T:
         """Construct an Executor from Hub.
 
@@ -474,12 +486,12 @@ def from_hub(
 
     @classmethod
     def serve(
-        cls,
-        uses_with: Optional[Dict] = None,
-        uses_metas: Optional[Dict] = None,
-        uses_requests: Optional[Dict] = None,
-        stop_event: Optional[Union[threading.Event, multiprocessing.Event]] = None,
-        **kwargs,
+            cls,
+            uses_with: Optional[Dict] = None,
+            uses_metas: Optional[Dict] = None,
+            uses_requests: Optional[Dict] = None,
+            stop_event: Optional[Union[threading.Event, multiprocessing.Event]] = None,
+            **kwargs,
     ):
         """Serve this Executor in a temporary Flow. Useful in testing an Executor in remote settings.
 
@@ -512,16 +524,16 @@ class StandaloneExecutorType(BetterEnum):
 
     @staticmethod
     def to_kubernetes_yaml(
-        uses: str,
-        output_base_path: str,
-        k8s_namespace: Optional[str] = None,
-        executor_type: Optional[
-            StandaloneExecutorType
-        ] = StandaloneExecutorType.EXTERNAL,
-        uses_with: Optional[Dict] = None,
-        uses_metas: Optional[Dict] = None,
-        uses_requests: Optional[Dict] = None,
-        **kwargs,
+            uses: str,
+            output_base_path: str,
+            k8s_namespace: Optional[str] = None,
+            executor_type: Optional[
+                StandaloneExecutorType
+            ] = StandaloneExecutorType.EXTERNAL,
+            uses_with: Optional[Dict] = None,
+            uses_metas: Optional[Dict] = None,
+            uses_requests: Optional[Dict] = None,
+            **kwargs,
     ):
         """
         Converts the Executor into a set of yaml deployments to deploy in Kubernetes.
@@ -549,23 +561,23 @@ def to_kubernetes_yaml(
             output_base_path=output_base_path,
             k8s_namespace=k8s_namespace,
             include_gateway=executor_type
-            == BaseExecutor.StandaloneExecutorType.EXTERNAL,
+                            == BaseExecutor.StandaloneExecutorType.EXTERNAL,
         )
 
     to_k8s_yaml = to_kubernetes_yaml
 
     @staticmethod
     def to_docker_compose_yaml(
-        uses: str,
-        output_path: Optional[str] = None,
-        network_name: Optional[str] = None,
-        executor_type: Optional[
-            StandaloneExecutorType
-        ] = StandaloneExecutorType.EXTERNAL,
-        uses_with: Optional[Dict] = None,
-        uses_metas: Optional[Dict] = None,
-        uses_requests: Optional[Dict] = None,
-        **kwargs,
+            uses: str,
+            output_path: Optional[str] = None,
+            network_name: Optional[str] = None,
+            executor_type: Optional[
+                StandaloneExecutorType
+            ] = StandaloneExecutorType.EXTERNAL,
+            uses_with: Optional[Dict] = None,
+            uses_metas: Optional[Dict] = None,
+            uses_requests: Optional[Dict] = None,
+            **kwargs,
     ):
         """
         Converts the Executor into a yaml file to run with `docker-compose up`
@@ -590,11 +602,11 @@ def to_docker_compose_yaml(
             output_path=output_path,
             network_name=network_name,
             include_gateway=executor_type
-            == BaseExecutor.StandaloneExecutorType.EXTERNAL,
+                            == BaseExecutor.StandaloneExecutorType.EXTERNAL,
         )
 
     def monitor(
-        self, name: Optional[str] = None, documentation: Optional[str] = None
+            self, name: Optional[str] = None, documentation: Optional[str] = None
     ) -> Optional[MetricsTimer]:
         """
         Get a given prometheus metric, if it does not exist yet, it will create it and store it in a buffer.
diff --git a/jina/serve/executors/decorators.py b/jina/serve/executors/decorators.py
index 6d6f9bdc529f3..b51e2cc195eaa 100644
--- a/jina/serve/executors/decorators.py
+++ b/jina/serve/executors/decorators.py
@@ -10,7 +10,7 @@
 from jina.helper import iscoroutinefunction
 from jina.importer import ImportExtensions
 
-if TYPE_CHECKING: # pragma: no cover
+if TYPE_CHECKING:  # pragma: no cover
     from jina import DocumentArray
 
 
@@ -44,11 +44,10 @@ def arg_wrapper(self, *args, **kwargs):
                 )
 
             if self.__class__ == cls:
-                file_lock = nullcontext()
                 with ImportExtensions(
-                    required=False,
-                    help_text=f'FileLock is needed to guarantee non-concurrent initialization of replicas in the same '
-                    f'machine.',
+                        required=False,
+                        help_text=f'FileLock is needed to guarantee non-concurrent initialization of replicas in the '
+                                  f'same machine.',
                 ):
                     import filelock
 
@@ -70,18 +69,18 @@ def arg_wrapper(self, *args, **kwargs):
 
 
 def requests(
-    func: Callable[
-        [
-            'DocumentArray',
-            Dict,
-            'DocumentArray',
-            List['DocumentArray'],
-            List['DocumentArray'],
-        ],
-        Optional[Union['DocumentArray', Dict]],
-    ] = None,
-    *,
-    on: Optional[Union[str, Sequence[str]]] = None,
+        func: Optional[Callable[
+            [
+                'DocumentArray',
+                Dict,
+                'DocumentArray',
+                List['DocumentArray'],
+                List['DocumentArray'],
+            ],
+            Optional[Union['DocumentArray', Dict]],
+        ]] = None,
+        *,
+        on: Optional[Union[str, Sequence[str]]] = None,
 ):
     """
     `@requests` defines the endpoints of an Executor. It has a keyword `on=` to define the endpoint.
@@ -133,10 +132,9 @@ def foo(self, docs, **kwargs):
 
     class FunctionMapper:
         def __init__(self, fn):
-
             arg_spec = inspect.getfullargspec(fn)
             if not arg_spec.varkw and not __args_executor_func__.issubset(
-                arg_spec.args
+                    arg_spec.args
             ):
                 raise TypeError(
                     f'{fn} accepts only {arg_spec.args} which is fewer than expected, '
@@ -144,10 +142,9 @@ def __init__(self, fn):
                 )
 
             if iscoroutinefunction(fn):
-
                 @functools.wraps(fn)
                 async def arg_wrapper(
-                    executor_instance, *args, **kwargs
+                        executor_instance, *args, **kwargs
                 ):  # we need to get the summary from the executor, so we need to access the self
                     return await fn(executor_instance, *args, **kwargs)
 
@@ -156,7 +153,7 @@ async def arg_wrapper(
 
                 @functools.wraps(fn)
                 def arg_wrapper(
-                    executor_instance, *args, **kwargs
+                        executor_instance, *args, **kwargs
                 ):  # we need to get the summary from the executor, so we need to access the self
                     return fn(executor_instance, *args, **kwargs)
 
@@ -164,14 +161,17 @@ def arg_wrapper(
 
         def __set_name__(self, owner, name):
             self.fn.class_name = owner.__name__
-            if not hasattr(owner, 'requests'):
-                owner.requests = {}
+            if not hasattr(owner, 'requests_by_class'):
+                owner.requests_by_class = {}
+
+            if owner.__name__ not in owner.requests_by_class:
+                owner.requests_by_class[owner.__name__] = {}
 
             if isinstance(on, (list, tuple)):
                 for o in on:
-                    owner.requests[o] = self.fn
+                    owner.requests_by_class[owner.__name__][o] = self.fn
             else:
-                owner.requests[on or __default_endpoint__] = self.fn
+                owner.requests_by_class[owner.__name__][on or __default_endpoint__] = self.fn
 
             setattr(owner, name, self.fn)
 
@@ -182,9 +182,9 @@ def __set_name__(self, owner, name):
 
 
 def monitor(
-    *,
-    name: Optional[str] = None,
-    documentation: Optional[str] = None,
+        *,
+        name: Optional[str] = None,
+        documentation: Optional[str] = None,
 ):
     """
     Decorator and context manager that allows monitoring of an Executor.
@@ -250,7 +250,6 @@ def foo(self, docs, *args, **kwargs):
     """
 
     def _decorator(func: Callable):
-
         name_ = name if name else f'{func.__name__}_seconds'
         documentation_ = (
             documentation
diff --git a/jina/serve/runtimes/worker/request_handling.py b/jina/serve/runtimes/worker/request_handling.py
index 910f46512f55d..3433d19f30b97 100644
--- a/jina/serve/runtimes/worker/request_handling.py
+++ b/jina/serve/runtimes/worker/request_handling.py
@@ -23,14 +23,14 @@ class WorkerRequestHandler:
     _KEY_RESULT = '__results__'
 
     def __init__(
-        self,
-        args: 'argparse.Namespace',
-        logger: 'JinaLogger',
-        metrics_registry: Optional['CollectorRegistry'] = None,
-        tracer_provider: Optional['trace.TracerProvider'] = None,
-        meter_provider: Optional['metrics.MeterProvider'] = None,
-        deployment_name: str = '',
-        **kwargs,
+            self,
+            args: 'argparse.Namespace',
+            logger: 'JinaLogger',
+            metrics_registry: Optional['CollectorRegistry'] = None,
+            tracer_provider: Optional['trace.TracerProvider'] = None,
+            meter_provider: Optional['metrics.MeterProvider'] = None,
+            deployment_name: str = '',
+            **kwargs,
     ):
         """Initialize private parameters and execute private loading functions.
 
@@ -60,16 +60,16 @@ def __init__(
         self.deployment_name = deployment_name
 
     def _init_monitoring(
-        self,
-        metrics_registry: Optional['CollectorRegistry'] = None,
-        meter: Optional['metrics.Meter'] = None,
+            self,
+            metrics_registry: Optional['CollectorRegistry'] = None,
+            meter: Optional['metrics.Meter'] = None,
     ):
 
         if metrics_registry:
 
             with ImportExtensions(
-                required=True,
-                help_text='You need to install the `prometheus_client` to use the montitoring functionality of jina',
+                    required=True,
+                    help_text='You need to install the `prometheus_client` to use the montitoring functionality of jina',
             ):
                 from prometheus_client import Counter, Summary
 
@@ -125,10 +125,10 @@ def _init_monitoring(
             self._sent_response_size_histogram = None
 
     def _load_executor(
-        self,
-        metrics_registry: Optional['CollectorRegistry'] = None,
-        tracer_provider: Optional['trace.TracerProvider'] = None,
-        meter_provider: Optional['metrics.MeterProvider'] = None,
+            self,
+            metrics_registry: Optional['CollectorRegistry'] = None,
+            tracer_provider: Optional['trace.TracerProvider'] = None,
+            meter_provider: Optional['metrics.MeterProvider'] = None,
     ):
         """
         Load the executor to this runtime, specified by ``uses`` CLI argument.
@@ -259,7 +259,7 @@ def _set_result(self, requests, return_data, docs):
         return docs
 
     async def handle(
-        self, requests: List['DataRequest'], tracing_context: Optional['Context'] = None
+            self, requests: List['DataRequest'], tracing_context: Optional['Context'] = None
     ) -> DataRequest:
         """Initialize private parameters and execute private loading functions.
 
@@ -268,9 +268,8 @@ async def handle(
         :returns: the processed message
         """
         # skip executor if endpoints mismatch
-        if (
-            requests[0].header.exec_endpoint not in self._executor.requests
-            and __default_endpoint__ not in self._executor.requests
+        if (requests[0].header.exec_endpoint not in self._executor.requests
+                and __default_endpoint__ not in self._executor.requests
         ):
             self.logger.debug(
                 f'skip executor: mismatch request, exec_endpoint: {requests[0].header.exec_endpoint}, requests: {self._executor.requests}'
@@ -310,7 +309,7 @@ async def handle(
 
     @staticmethod
     def replace_docs(
-        request: List['DataRequest'], docs: 'DocumentArray', ndarrray_type: str = None
+            request: List['DataRequest'], docs: 'DocumentArray', ndarrray_type: str = None
     ) -> None:
         """Replaces the docs in a message with new Documents.
 
@@ -352,7 +351,7 @@ def close(self):
 
     @staticmethod
     def _get_docs_matrix_from_request(
-        requests: List['DataRequest'],
+            requests: List['DataRequest'],
     ) -> Tuple[Optional[List['DocumentArray']], Optional[Dict[str, 'DocumentArray']]]:
         """
         Returns a docs matrix from a list of DataRequest objects.
@@ -376,7 +375,7 @@ def _get_docs_matrix_from_request(
 
     @staticmethod
     def get_parameters_dict_from_request(
-        requests: List['DataRequest'],
+            requests: List['DataRequest'],
     ) -> 'Dict':
         """
         Returns a parameters dict from a list of DataRequest objects.
@@ -396,8 +395,8 @@ def get_parameters_dict_from_request(
 
     @staticmethod
     def get_docs_from_request(
-        requests: List['DataRequest'],
-        field: str,
+            requests: List['DataRequest'],
+            field: str,
     ) -> 'DocumentArray':
         """
         Gets a field from the message
diff --git a/tests/integration/issues/github_5735/__init__.py b/tests/integration/issues/github_5735/__init__.py
new file mode 100644
index 0000000000000..e69de29bb2d1d
diff --git a/tests/integration/issues/github_5735/test_inherited_executor.py b/tests/integration/issues/github_5735/test_inherited_executor.py
new file mode 100644
index 0000000000000..120e1593a0a28
--- /dev/null
+++ b/tests/integration/issues/github_5735/test_inherited_executor.py
@@ -0,0 +1,23 @@
+from jina import Executor, requests, Flow, DocumentArray
+
+
+def test_call_good_base_executor():
+    class A(Executor):
+        @requests
+        def x(self, docs, **kwargs):
+            for doc in docs:
+                doc.tags['exec'] = 'A'
+
+    class EnhancedExecutor(A):
+        @requests
+        def y(self, docs, **kwargs):
+            for doc in docs:
+                doc.tags['exec'] = 'Enhanced'
+
+    assert EnhancedExecutor.__name__ in A.requests_by_class
+    assert A.__name__ in EnhancedExecutor.requests_by_class
+    f = Flow().add(uses=A)
+    with f:
+        ret = f.post(on='/', inputs=DocumentArray.empty(1))[0]
+
+    assert ret.tags['exec'] == 'A'
diff --git a/tests/unit/serve/executors/test_executor.py b/tests/unit/serve/executors/test_executor.py
index d0979ed9f8a0f..06249d2484d86 100644
--- a/tests/unit/serve/executors/test_executor.py
+++ b/tests/unit/serve/executors/test_executor.py
@@ -2,10 +2,8 @@
 import multiprocessing
 import os
 import time
-from copy import deepcopy
 from multiprocessing import Process
 from threading import Event
-from unittest import mock
 
 import pytest
 import yaml
@@ -19,8 +17,7 @@
 from jina.serve.networking import GrpcConnectionPool
 from jina.serve.runtimes.asyncio import AsyncNewLoopRuntime
 from jina.serve.runtimes.worker import WorkerRuntime
-
-PORT = 12350
+from jina.helper import random_port
 
 
 class WorkspaceExec(Executor):
@@ -35,8 +32,14 @@ def foo(self, docs, **kwargs):
         docs.texts = ['foo' for _ in docs]
 
 
+@pytest.fixture()
+def exposed_port():
+    port = random_port()
+    yield port
+
+
 @pytest.fixture(autouse=False)
-def served_exec():
+def served_exec(exposed_port):
     import threading
     import time
 
@@ -47,7 +50,7 @@ def serve_exec(**kwargs):
     t = threading.Thread(
         name='serve-exec',
         target=serve_exec,
-        kwargs={'port_expose': PORT, 'stop_event': e},
+        kwargs={'port_expose': exposed_port, 'stop_event': e},
     )
     t.start()
     time.sleep(3)  # allow Flow to start
@@ -78,7 +81,7 @@ def test_executor_import_with_external_dependencies(capsys):
 
 def test_executor_with_pymodule_path():
     with pytest.raises(FileNotFoundError):
-        ex = Executor.load_config(
+        _ = Executor.load_config(
             '''
         jtype: BaseExecutor
         py_modules:
@@ -101,7 +104,7 @@ def test_executor_with_pymodule_path():
 
 def test_flow_uses_with_pymodule_path():
     with Flow.load_config(
-        '''
+            '''
     jtype: Flow
     executors:
         - uses: unit.serve.executors.dummy_executor.MyExecutor
@@ -112,13 +115,13 @@ def test_flow_uses_with_pymodule_path():
         pass
 
     with Flow().add(
-        uses='unit.serve.executors.dummy_executor.MyExecutor', uses_with={'bar': 123}
+            uses='unit.serve.executors.dummy_executor.MyExecutor', uses_with={'bar': 123}
     ):
         pass
 
     with pytest.raises(RuntimeFailToStart):
         with Flow.load_config(
-            '''
+                '''
             jtype: Flow
             executors:
                 - uses: jina.no_valide.executor
@@ -235,7 +238,7 @@ def test_executor_workspace(test_metas_workspace_replica_pods, shard_id):
 
 @pytest.mark.parametrize('shard_id', [None, -1], indirect=True)
 def test_executor_workspace_parent_replica_nopea(
-    test_metas_workspace_replica_pods, shard_id
+        test_metas_workspace_replica_pods, shard_id
 ):
     executor = Executor(
         metas={'name': test_metas_workspace_replica_pods['name']},
@@ -251,7 +254,7 @@ def test_executor_workspace_parent_replica_nopea(
 
 @pytest.mark.parametrize('shard_id', [0, 1, 2], indirect=True)
 def test_executor_workspace_parent_noreplica_pod(
-    test_metas_workspace_replica_pods, shard_id
+        test_metas_workspace_replica_pods, shard_id
 ):
     executor = Executor(
         metas={'name': test_metas_workspace_replica_pods['name']},
@@ -268,7 +271,7 @@ def test_executor_workspace_parent_noreplica_pod(
 
 @pytest.mark.parametrize('shard_id', [None, -1], indirect=True)
 def test_executor_workspace_parent_noreplica_nopea(
-    test_metas_workspace_replica_pods, shard_id
+        test_metas_workspace_replica_pods, shard_id
 ):
     executor = Executor(
         metas={'name': test_metas_workspace_replica_pods['name']},
@@ -385,8 +388,8 @@ async def foo(self, docs: DocumentArray, **kwargs):
     assert da1.texts == ['hello'] * N
 
 
-def test_serve(served_exec):
-    docs = Client(port=PORT).post(on='/foo', inputs=DocumentArray.empty(5))
+def test_serve(served_exec, exposed_port):
+    docs = Client(port=exposed_port).post(on='/foo', inputs=DocumentArray.empty(5))
 
     assert docs.texts == ['foo' for _ in docs]
 
@@ -403,7 +406,7 @@ def test_set_workspace(tmpdir):
         os.path.join(tmpdir, 'WorkspaceExec')
     )
     assert (
-        WorkspaceExec(workspace=str(tmpdir)).workspace == complete_workspace_no_replicas
+            WorkspaceExec(workspace=str(tmpdir)).workspace == complete_workspace_no_replicas
     )
 
 
@@ -448,10 +451,10 @@ def test_to_k8s_yaml(tmpdir, exec_type):
         with open(os.path.join(tmpdir, 'gateway', 'gateway.yml')) as f:
             gatewayyaml = list(yaml.safe_load_all(f))[-1]
             assert (
-                gatewayyaml['spec']['template']['spec']['containers'][0]['ports'][0][
-                    'containerPort'
-                ]
-                == 2020
+                    gatewayyaml['spec']['template']['spec']['containers'][0]['ports'][0][
+                        'containerPort'
+                    ]
+                    == 2020
             )
             gateway_args = gatewayyaml['spec']['template']['spec']['containers'][0][
                 'args'


Code review comments:
diff_hunk: @@ -214,9 +214,20 @@ def _init_instrumentation(self, _runtime_args: Optional[Dict] = None):
             self.meter_provider = None
             self.meter = None
 
+    @property
+    def requests(self):
+        """
+        Get the request dictionary corresponding to this specific class
+
+        :return: Returns the requests corresponding to the specific Executor instance class
+        """
+        if not hasattr(self, 'requests_by_class'):
+            self.requests_by_class = {}
path: jina/serve/executors/__init__.py
text: @user1:
should we make this attribute private ?