Repository: jina-ai/serve
Pull request: feat: add hot reload option to Executor for development
<!---Decomposing the complex issue into subtasks can help you build it step-by-step. Thanks for your pull request! :rocket: --->
<!---We know that dev life is hectic, but **please provide a (brief) description** of what your PR does, and how it does it. **Otherwise, your PR cannot be reviewed!** --->
<!---This policy was agreed upon in a past company retro, and makes everyone's life a little easier. Thanks for your collaboration!--->

**Goals:**
Add hot reload to the Executor
closes: https://github.com/jina-ai/jina/issues/5400

Problem statement:
Hot reload for API endpoints
**Describe the feature**
When editing code while running in a local development environment, something like an inotifywait watcher either kills the existing executor or sends a message to the executor to hotpatch the server. This makes development much easier, and the latter method does not require reloading ML models which can be a slow process.

**Your proposal**
Look at how other command development servers like FastAPI implement hot updates and conceive of some way to hotpatch or reload modules related to the jina RPC server with low overhead.

**Environment**
Linux x64 latest.

**Screenshots**
N/A

Hints from issue discussion:
Hello @AmericanPresidentJimmyCarter , thanks for this feature request.

Let me clarify what it can be expected and some doubts I have:

Would you expect something like this experience (APPROXIMATE):

```python
from jina import Flow

with Flow().add(uses=some_executor, reload=True) as f:
    f.block()
```

This would mean that somehow a `watcher` is observing this `Executor` (perhaps the `file`)? and trying to see if it is updated, and if it is updated, reload the Executor in memory so that the developer can test the feature without needing to restart the Flow. Is that correct?

Assuming this is correct, I am not sure how it is possible to avoid reloading ML models into memory.

For instance:

if I have an Executor in a file my_executor.py:

```python
from jina import Executor, requests

class MyExec(Execurtor):
    def __init__(self, **kwargs):
         ....
         self.model = LOAD_MODEL_FROM_FILE

    @requests
    def foo(docs, **kwargs):
       self.model(docs)
```

and then I change this File, to something else, I would just reload the Executor class from the file, but I would not be able to keep the `MyExec` from reloading `self.model`.
> This would mean that somehow a watcher is observing this Executor (perhaps the file)? and trying to see if it is updated, and if it is updated, reload the Executor in memory so that the developer can test the feature without needing to restart the Flow. Is that correct?

This is correct.

> and then I change this File, to something else, I would just reload the Executor class from the file, but I would not be able to keep the MyExec from reloading self.model.

It depends on the implementation. In the case that you reload the entire process, this would be true, and that is the easiest way to implement this. However, other implementations will only reload modules that have been changed by replacing pointers, while completely ignoring those that do not. One such python implementation is jurigged:

https://github.com/breuleux/jurigged

And it is common for js webapps to hotpatch via JSON updates. There is a probably a smart way to do this locally that would not involve reloading any python modules that have not changed.
Reloadium is another such hot reloader: https://github.com/reloadware/reloadium
I will take a look, thanks for the recommendations.

Any pointer to the specific `Python` magic these projects use to provide such functionality?
> other

Hey @AmericanPresidentJimmyCarter ,

If I were to reload modules, then it means that u may separate the `executor.py` from some other `dependency.py` where you do the changes?

Then we could have a watcher observing every file inside `py_modules` https://docs.jina.ai/fundamentals/executor/yaml-spec/#py-modules and see which are updated and do `reload` module with `importlib` right?

There are some caveats however on using this

https://docs.python.org/3/library/importlib.html#importlib.reload

Commit message: docs: add documentation about reload feature

Patch to review:
diff --git a/docs/fundamentals/executor/executor-files.md b/docs/fundamentals/executor/executor-files.md
index 833527d5e9b62..7b07f2cca3ebe 100644
--- a/docs/fundamentals/executor/executor-files.md
+++ b/docs/fundamentals/executor/executor-files.md
@@ -18,8 +18,6 @@ py_modules:
 
 ## Multiple Python files + YAML
 
-
-
 When you are working with multiple Python files, you should organize them as a **Python package** and put them in a special folder inside
 your repository (as you would normally do with Python packages). Specifically, you should do the following:
 
diff --git a/docs/fundamentals/executor/index.md b/docs/fundamentals/executor/index.md
index 4a23bf1821095..f6fa531f08c8b 100644
--- a/docs/fundamentals/executor/index.md
+++ b/docs/fundamentals/executor/index.md
@@ -42,6 +42,7 @@ instrumenting-executor
 executor-run
 executor-serve
 executor-files
+executor-reload
 containerize-executor
 yaml-spec
 ```
\ No newline at end of file
diff --git a/docs/fundamentals/flow/executor-args.md b/docs/fundamentals/flow/executor-args.md
index 6ddf2858f92a9..9fcfbafb32aee 100644
--- a/docs/fundamentals/flow/executor-args.md
+++ b/docs/fundamentals/flow/executor-args.md
@@ -29,6 +29,7 @@
 | `timeout_ready` | The timeout in milliseconds of a Pod waits for the runtime to be ready, -1 for waiting forever | `number` | `600000` |
 | `env` | The map of environment variables that are available inside runtime | `object` | `None` |
 | `floating` | If set, the current Pod/Deployment can not be further chained, and the next `.add()` will chain after the last Pod/Deployment not this current one. | `boolean` | `False` |
+| `reload` | If set, the Executor will reload the modules as they change | `boolean` | `False` |
 | `monitoring` | If set, spawn an http server with a prometheus endpoint to expose metrics | `boolean` | `False` |
 | `port_monitoring` | The port on which the prometheus server is exposed, default is a random port between [49152, 65535] | `string` | `random in [49152, 65535]` |
 | `retries` | Number of retries per gRPC call. If <0 it defaults to max(3, num_replicas) | `number` | `-1` |
diff --git a/extra-requirements.txt b/extra-requirements.txt
index 602c9612fd3fb..dea308b8ad8f9 100644
--- a/extra-requirements.txt
+++ b/extra-requirements.txt
@@ -85,3 +85,4 @@ portforward>=0.2.4:         cicd
 tensorflow>=2.0:            cicd
 opentelemetry-test-utils>=0.33b0:  test
 prometheus-api-client>=0.5.1:  test
+watchfiles>=0.18.0:         devel
\ No newline at end of file
diff --git a/jina/orchestrate/flow/base.py b/jina/orchestrate/flow/base.py
index ec6a921d10446..acb3741d34cdd 100644
--- a/jina/orchestrate/flow/base.py
+++ b/jina/orchestrate/flow/base.py
@@ -856,6 +856,7 @@ def add(
         py_modules: Optional[List[str]] = None,
         quiet: Optional[bool] = False,
         quiet_error: Optional[bool] = False,
+        reload: Optional[bool] = False,
         replicas: Optional[int] = 1,
         retries: Optional[int] = -1,
         runtime_cls: Optional[str] = 'WorkerRuntime',
@@ -945,6 +946,7 @@ def add(
           `Executor cookbook <https://docs.jina.ai/fundamentals/executor/executor-files/>`__
         :param quiet: If set, then no log will be emitted from this object.
         :param quiet_error: If set, then exception stack information will not be added to the log
+        :param reload: If set, the Executor will reload the modules as they change
         :param replicas: The number of replicas in the deployment
         :param retries: Number of retries per gRPC call. If <0 it defaults to max(3, num_replicas)
         :param runtime_cls: The runtime class to run inside the Pod
@@ -1089,6 +1091,7 @@ def add(
           `Executor cookbook <https://docs.jina.ai/fundamentals/executor/executor-files/>`__
         :param quiet: If set, then no log will be emitted from this object.
         :param quiet_error: If set, then exception stack information will not be added to the log
+        :param reload: If set, the Executor will reload the modules as they change
         :param replicas: The number of replicas in the deployment
         :param retries: Number of retries per gRPC call. If <0 it defaults to max(3, num_replicas)
         :param runtime_cls: The runtime class to run inside the Pod
diff --git a/jina/parsers/orchestrate/pod.py b/jina/parsers/orchestrate/pod.py
index 4486399dc9a6d..d151276cc5e42 100644
--- a/jina/parsers/orchestrate/pod.py
+++ b/jina/parsers/orchestrate/pod.py
@@ -123,6 +123,12 @@ def mixin_pod_runtime_args_parser(arg_group, pod_type='worker'):
             action=CastToIntAction,
             help=port_description,
         )
+        arg_group.add_argument(
+            '--reload',
+            action='store_true',
+            default=False,
+            help='If set, the Executor will reload the modules as they change'
+        )
     else:
         port_description = (
             'The port for input data to bind the gateway server to, by default, random ports between range [49152, 65535] will be assigned. '
diff --git a/jina/resources/extra-requirements.txt b/jina/resources/extra-requirements.txt
index 602c9612fd3fb..dea308b8ad8f9 100644
--- a/jina/resources/extra-requirements.txt
+++ b/jina/resources/extra-requirements.txt
@@ -85,3 +85,4 @@ portforward>=0.2.4:         cicd
 tensorflow>=2.0:            cicd
 opentelemetry-test-utils>=0.33b0:  test
 prometheus-api-client>=0.5.1:  test
+watchfiles>=0.18.0:         devel
\ No newline at end of file
diff --git a/jina/serve/executors/__init__.py b/jina/serve/executors/__init__.py
index c09a25170c21c..a8d5128df9a84 100644
--- a/jina/serve/executors/__init__.py
+++ b/jina/serve/executors/__init__.py
@@ -1,4 +1,3 @@
-import contextlib
 import copy
 import inspect
 import multiprocessing
@@ -13,7 +12,7 @@
 
 from jina import __args_executor_init__, __cache_path__, __default_endpoint__
 from jina.enums import BetterEnum
-from jina.helper import ArgNamespace, T, iscoroutinefunction, typename
+from jina.helper import ArgNamespace, T, iscoroutinefunction, typename, get_or_reuse_loop
 from jina.importer import ImportExtensions
 from jina.jaml import JAML, JAMLCompatible, env_var_regex, internal_var_regex
 from jina.logging.logger import JinaLogger
@@ -24,7 +23,6 @@
 from jina.serve.executors.metas import get_executor_taboo
 from jina.serve.helper import store_init_kwargs, wrap_func
 from jina.serve.instrumentation import MetricsTimer
-from jina.helper import get_or_reuse_loop
 
 if TYPE_CHECKING:  # pragma: no cover
     from opentelemetry.context.context import Context
diff --git a/jina/serve/runtimes/worker/__init__.py b/jina/serve/runtimes/worker/__init__.py
index 554cb04a5e520..7daa5c403e0f2 100644
--- a/jina/serve/runtimes/worker/__init__.py
+++ b/jina/serve/runtimes/worker/__init__.py
@@ -1,4 +1,6 @@
 import argparse
+import asyncio
+import os
 from abc import ABC
 from typing import TYPE_CHECKING, List, Optional
 
@@ -24,14 +26,15 @@ class WorkerRuntime(AsyncNewLoopRuntime, ABC):
     """Runtime procedure leveraging :class:`Grpclet` for sending DataRequests"""
 
     def __init__(
-        self,
-        args: argparse.Namespace,
-        **kwargs,
+            self,
+            args: argparse.Namespace,
+            **kwargs,
     ):
         """Initialize grpc and data request handling.
         :param args: args from CLI
         :param kwargs: keyword args
         """
+        self._hot_reload_task = None
         self._health_servicer = health.aio.HealthServicer()
         super().__init__(args, **kwargs)
 
@@ -41,8 +44,8 @@ async def async_setup(self):
         """
         if self.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
 
@@ -135,24 +138,56 @@ async def _async_setup_grpc_server(self):
             reflection.SERVICE_NAME,
         )
         # Mark all services as healthy.
-        health_pb2_grpc.add_HealthServicer_to_server(self._health_servicer, self._grpc_server)
+        health_pb2_grpc.add_HealthServicer_to_server(
+            self._health_servicer, self._grpc_server
+        )
 
         for service in service_names:
-            await self._health_servicer.set(service, health_pb2.HealthCheckResponse.SERVING)
+            await self._health_servicer.set(
+                service, health_pb2.HealthCheckResponse.SERVING
+            )
         reflection.enable_server_reflection(service_names, self._grpc_server)
         bind_addr = f'{self.args.host}:{self.args.port}'
         self.logger.debug(f'start listening on {bind_addr}')
         self._grpc_server.add_insecure_port(bind_addr)
         await self._grpc_server.start()
 
+    async def _hot_reload(self):
+        import inspect
+        executor_file = inspect.getfile(self._request_handler._executor.__class__)
+        watched_files = set([executor_file] + (self.args.py_modules or []))
+        executor_base_path = os.path.dirname(os.path.abspath(executor_file))
+        extra_paths = [os.path.join(path, name) for path, subdirs, files in os.walk(executor_base_path) for name in files]
+        extra_python_paths = list(filter(lambda x: x.endswith('.py'), extra_paths))
+        for extra_python_file in extra_python_paths:
+            watched_files.add(extra_python_file)
+
+        with ImportExtensions(
+                required=True,
+                logger=self.logger,
+                help_text='''hot reload requires watchfiles dependency to be installed. You can do `pip install 
+                watchfiles''',
+        ):
+            from watchfiles import awatch
+
+        async for changes in awatch(*watched_files):
+            changed_files = [changed_file for _, changed_file in changes]
+            self.logger.info(
+                f'detected changes in: {changed_files}. Refreshing the Executor'
+            )
+            self._request_handler._refresh_executor(changed_files)
+
     async def async_run_forever(self):
         """Block until the GRPC server is terminated"""
+        if self.args.reload:
+            self._hot_reload_task = asyncio.create_task(self._hot_reload())
         await self._grpc_server.wait_for_termination()
 
     async def async_cancel(self):
         """Stop the GRPC server"""
         self.logger.debug('cancel WorkerRuntime')
-
+        if self._hot_reload_task is not None:
+            self._hot_reload_task.cancel()
         # 0.5 gives the runtime some time to complete outstanding responses
         # this should be handled better, 1.0 is a rather random number
         await self._grpc_server.stop(1.0)
@@ -190,7 +225,7 @@ async def endpoint_discovery(self, empty, context) -> jina_pb2.EndpointsProto:
         return endpoints_proto
 
     def _extract_tracing_context(
-        self, metadata: grpc.aio.Metadata
+            self, metadata: grpc.aio.Metadata
     ) -> Optional['Context']:
         if self.tracer:
             from opentelemetry.propagate import extract
@@ -210,7 +245,7 @@ async def process_data(self, requests: List[DataRequest], context) -> DataReques
         """
 
         with MetricsTimer(
-            self._summary, self._receiving_request_seconds, self._metric_attributes
+                self._summary, self._receiving_request_seconds, self._metric_attributes
         ):
             try:
                 if self.logger.debug_enabled:
@@ -248,8 +283,8 @@ async def process_data(self, requests: List[DataRequest], context) -> DataReques
                     )
 
                 if (
-                    self.args.exit_on_exceptions
-                    and type(ex).__name__ in self.args.exit_on_exceptions
+                        self.args.exit_on_exceptions
+                        and type(ex).__name__ in self.args.exit_on_exceptions
                 ):
                     self.logger.info('Exiting because of "--exit-on-exceptions".')
                     raise RuntimeTerminated
diff --git a/jina/serve/runtimes/worker/request_handling.py b/jina/serve/runtimes/worker/request_handling.py
index 3433d19f30b97..8a207ef84d25a 100644
--- a/jina/serve/runtimes/worker/request_handling.py
+++ b/jina/serve/runtimes/worker/request_handling.py
@@ -1,6 +1,7 @@
 from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
 
 from docarray import DocumentArray
+
 from jina import __default_endpoint__
 from jina.excepts import BadConfigSource
 from jina.importer import ImportExtensions
@@ -23,14 +24,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 +61,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 +126,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.
@@ -167,6 +168,34 @@ def _load_executor(
             self.logger.critical(f'can not load the executor from {self.args.uses}')
             raise
 
+    def _refresh_executor(self, changed_files):
+        import importlib
+        import inspect
+        import copy
+        import sys
+
+        try:
+            sys_mod_files_modules = {getattr(module, '__file__', ''): module for module in sys.modules.values()}
+
+            for file in changed_files:
+                importlib.reload(sys_mod_files_modules[file])
+        except Exception as exc:
+            self.logger.error(f' Exception when refreshing Executor when changes detected in {changed_files}')
+            raise exc
+
+        importlib.reload(inspect.getmodule(self._executor.__class__))
+        requests = copy.copy(self._executor.requests)
+        old_cls = self._executor.__class__
+        new_cls = getattr(importlib.import_module(old_cls.__module__), old_cls.__name__)
+        new_executor = new_cls.__new__(new_cls)
+        new_executor.__dict__ = self._executor.__dict__
+        for k, v in requests.items():
+            requests[k] = getattr(new_executor.__class__, requests[k].__name__)
+        self._executor = new_executor
+        self._executor.requests.clear()
+        requests = {k: v.__name__ for k, v in requests.items()}
+        self._executor._add_requests(requests)
+
     @staticmethod
     def _parse_params(parameters: Dict, executor_name: str):
         parsed_params = parameters
@@ -259,7 +288,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,8 +297,9 @@ 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}'
@@ -309,7 +339,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.
 
@@ -351,7 +381,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.
@@ -375,7 +405,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.
@@ -395,8 +425,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/jina_cli/autocomplete.py b/jina_cli/autocomplete.py
index 9d2a307bd3c52..2d646df0c24a9 100644
--- a/jina_cli/autocomplete.py
+++ b/jina_cli/autocomplete.py
@@ -59,6 +59,7 @@
             '--floating',
             '--port',
             '--port-in',
+            '--reload',
             '--monitoring',
             '--port-monitoring',
             '--retries',
@@ -273,6 +274,7 @@
             '--floating',
             '--port',
             '--port-in',
+            '--reload',
             '--monitoring',
             '--port-monitoring',
             '--retries',
@@ -332,6 +334,7 @@
             '--floating',
             '--port',
             '--port-in',
+            '--reload',
             '--monitoring',
             '--port-monitoring',
             '--retries',
diff --git a/tests/integration/hot_reload/__init__.py b/tests/integration/hot_reload/__init__.py
new file mode 100644
index 0000000000000..e69de29bb2d1d
diff --git a/tests/integration/hot_reload/exec1/__init__.py b/tests/integration/hot_reload/exec1/__init__.py
new file mode 100644
index 0000000000000..e69de29bb2d1d
diff --git a/tests/integration/hot_reload/exec1/my_executor1.py b/tests/integration/hot_reload/exec1/my_executor1.py
new file mode 100644
index 0000000000000..03d45fc4ba867
--- /dev/null
+++ b/tests/integration/hot_reload/exec1/my_executor1.py
@@ -0,0 +1,12 @@
+from jina import Executor, requests
+
+
+class MyExecutorToReload1(Executor):
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    @requests()
+    def foo(self, docs, **kwargs):
+        for doc in docs:
+            doc.text = 'MyExecutorBeforeReload'
diff --git a/tests/integration/hot_reload/exec2/__init__.py b/tests/integration/hot_reload/exec2/__init__.py
new file mode 100644
index 0000000000000..e69de29bb2d1d
diff --git a/tests/integration/hot_reload/exec2/helper.py b/tests/integration/hot_reload/exec2/helper.py
new file mode 100644
index 0000000000000..5816d9310ec4d
--- /dev/null
+++ b/tests/integration/hot_reload/exec2/helper.py
@@ -0,0 +1,3 @@
+
+def get_doc_value():
+    return 'MyExecutorBeforeReload'
diff --git a/tests/integration/hot_reload/exec2/my_executor2.py b/tests/integration/hot_reload/exec2/my_executor2.py
new file mode 100644
index 0000000000000..261c448498f69
--- /dev/null
+++ b/tests/integration/hot_reload/exec2/my_executor2.py
@@ -0,0 +1,14 @@
+from jina import Executor, requests
+
+from .helper import get_doc_value
+
+
+class MyExecutorToReload2(Executor):
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    @requests()
+    def foo(self, docs, **kwargs):
+        for doc in docs:
+            doc.text = get_doc_value()
diff --git a/tests/integration/hot_reload/exec3/__init__.py b/tests/integration/hot_reload/exec3/__init__.py
new file mode 100644
index 0000000000000..e69de29bb2d1d
diff --git a/tests/integration/hot_reload/exec3/my_executor3.py b/tests/integration/hot_reload/exec3/my_executor3.py
new file mode 100644
index 0000000000000..3d649f25fba15
--- /dev/null
+++ b/tests/integration/hot_reload/exec3/my_executor3.py
@@ -0,0 +1,15 @@
+from jina import Executor, requests
+
+
+class A(Executor):
+    @requests
+    def x(self, docs, **kwargs):
+        for doc in docs:
+            doc.text = 'ABeforeReload'
+
+
+class EnhancedExecutor(A):
+    @requests
+    def y(self, docs, **kwargs):
+        for doc in docs:
+            doc.text = 'EnhancedBeforeReload'
diff --git a/tests/integration/hot_reload/helper2.py b/tests/integration/hot_reload/helper2.py
new file mode 100644
index 0000000000000..c87e9eb66fc25
--- /dev/null
+++ b/tests/integration/hot_reload/helper2.py
@@ -0,0 +1,3 @@
+
+def get_doc_value():
+    return 'MyExecutorAfterReload'
diff --git a/tests/integration/hot_reload/my_executor_1_new.py b/tests/integration/hot_reload/my_executor_1_new.py
new file mode 100644
index 0000000000000..26c5dea85d6c4
--- /dev/null
+++ b/tests/integration/hot_reload/my_executor_1_new.py
@@ -0,0 +1,12 @@
+from jina import Executor, requests
+
+
+class MyExecutorToReload1(Executor):
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    @requests()
+    def foo(self, docs, **kwargs):
+        for doc in docs:
+            doc.text = 'MyExecutorAfterReload'
diff --git a/tests/integration/hot_reload/my_executor_3_new.py b/tests/integration/hot_reload/my_executor_3_new.py
new file mode 100644
index 0000000000000..99e257151369f
--- /dev/null
+++ b/tests/integration/hot_reload/my_executor_3_new.py
@@ -0,0 +1,15 @@
+from jina import Executor, requests
+
+
+class A(Executor):
+    @requests
+    def x(self, docs, **kwargs):
+        for doc in docs:
+            doc.text = 'AAfterReload'
+
+
+class EnhancedExecutor(A):
+    @requests
+    def y(self, docs, **kwargs):
+        for doc in docs:
+            doc.text = 'EnhancedAfterReload'
diff --git a/tests/integration/hot_reload/test_hot_reload.py b/tests/integration/hot_reload/test_hot_reload.py
new file mode 100644
index 0000000000000..384e706e6a765
--- /dev/null
+++ b/tests/integration/hot_reload/test_hot_reload.py
@@ -0,0 +1,84 @@
+import os
+import time
+import shutil
+import contextlib
+
+from jina import Flow, DocumentArray
+
+cur_dir = os.path.dirname(os.path.abspath(__file__))
+
+
+@contextlib.contextmanager
+def _update_file(input_file_path, output_file_path, temp_path):
+    backup_file = os.path.join(temp_path, 'backup.py')
+    try:
+        shutil.copy2(output_file_path, backup_file)
+        shutil.copy(input_file_path, output_file_path)
+        time.sleep(2.0)
+        yield
+    finally:
+        shutil.copy2(backup_file, output_file_path)
+        time.sleep(2.0)
+
+
+def test_reload_simple_executor(tmpdir):
+    from tests.integration.hot_reload.exec1.my_executor1 import MyExecutorToReload1
+
+    f = Flow().add(uses=MyExecutorToReload1, reload=True)
+    with f:
+        res = f.post(on='/', inputs=DocumentArray.empty(10))
+        assert len(res) == 10
+        for doc in res:
+            assert doc.text == 'MyExecutorBeforeReload'
+        with _update_file(os.path.join(cur_dir, 'my_executor_1_new.py'), os.path.join(cur_dir, 'exec1/my_executor1.py'),
+                          str(tmpdir)):
+            res = f.post(on='/', inputs=DocumentArray.empty(10))
+            assert len(res) == 10
+            for doc in res:
+                assert doc.text == 'MyExecutorAfterReload'
+        res = f.post(on='/', inputs=DocumentArray.empty(10))
+        assert len(res) == 10
+        for doc in res:
+            assert doc.text == 'MyExecutorBeforeReload'
+
+
+def test_reload_helper(tmpdir):
+    from tests.integration.hot_reload.exec2.my_executor2 import MyExecutorToReload2
+
+    f = Flow().add(uses=MyExecutorToReload2, reload=True)
+    with f:
+        res = f.post(on='/', inputs=DocumentArray.empty(10))
+        assert len(res) == 10
+        for doc in res:
+            assert doc.text == 'MyExecutorBeforeReload'
+        with _update_file(os.path.join(cur_dir, 'helper2.py'), os.path.join(cur_dir, 'exec2/helper.py'),
+                          str(tmpdir)):
+            res = f.post(on='/', inputs=DocumentArray.empty(10))
+            assert len(res) == 10
+            for doc in res:
+                assert doc.text == 'MyExecutorAfterReload'
+        res = f.post(on='/', inputs=DocumentArray.empty(10))
+        assert len(res) == 10
+        for doc in res:
+            assert doc.text == 'MyExecutorBeforeReload'
+
+
+def test_reload_with_inheritance(tmpdir):
+    from tests.integration.hot_reload.exec3.my_executor3 import A, EnhancedExecutor
+
+    f = Flow().add(uses=A, reload=True)
+    with f:
+        res = f.post(on='/', inputs=DocumentArray.empty(10))
+        assert len(res) == 10
+        for doc in res:
+            assert doc.text == 'ABeforeReload'
+        with _update_file(os.path.join(cur_dir, 'my_executor_3_new.py'), os.path.join(cur_dir, 'exec3/my_executor3.py'),
+                          str(tmpdir)):
+            res = f.post(on='/', inputs=DocumentArray.empty(10))
+            assert len(res) == 10
+            for doc in res:
+                assert doc.text == 'AAfterReload'
+        res = f.post(on='/', inputs=DocumentArray.empty(10))
+        assert len(res) == 10
+        for doc in res:
+            assert doc.text == 'ABeforeReload'
\ No newline at end of file


Code review comments:
diff_hunk: @@ -29,6 +29,7 @@
 | `timeout_ready` | The timeout in milliseconds of a Pod waits for the runtime to be ready, -1 for waiting forever | `number` | `600000` |
 | `env` | The map of environment variables that are available inside runtime | `object` | `None` |
 | `floating` | If set, the current Pod/Deployment can not be further chained, and the next `.add()` will chain after the last Pod/Deployment not this current one. | `boolean` | `False` |
+| `reload` | If set, the Executor will reload the modules as they change | `boolean` | `False` |
path: docs/fundamentals/flow/executor-args.md
text: @user1:
```suggestion
| `reload` | If set, the Executor reloads the modules as they change | `boolean` | `False` |
```