diff --git a/.github/workflows/mirror-rolling-to-master.yaml b/.github/workflows/mirror-rolling-to-master.yaml
new file mode 100644
index 000000000..2885eb4a4
--- /dev/null
+++ b/.github/workflows/mirror-rolling-to-master.yaml
@@ -0,0 +1,13 @@
+name: Mirror rolling to master
+
+on:
+  push:
+    branches: [ rolling ]
+
+jobs:
+  mirror-to-master:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: zofrex/mirror-branch@v1
+      with:
+        target-branch: master
diff --git a/rclpy/docs/source/api/parameters.rst b/rclpy/docs/source/api/parameters.rst
index c38ee7d9c..5dee741fd 100644
--- a/rclpy/docs/source/api/parameters.rst
+++ b/rclpy/docs/source/api/parameters.rst
@@ -10,3 +10,8 @@ Parameter Service
 -----------------
 
 .. automodule:: rclpy.parameter_service
+
+Parameter Client
+-----------------
+
+.. automodule:: rclpy.parameter_client
diff --git a/rclpy/docs/source/conf.py b/rclpy/docs/source/conf.py
index ac689dd2c..b218279b1 100644
--- a/rclpy/docs/source/conf.py
+++ b/rclpy/docs/source/conf.py
@@ -58,6 +58,7 @@
     'sphinx.ext.autosummary',
     'sphinx.ext.doctest',
     'sphinx.ext.coverage',
+    'sphinx_rtd_theme',
 ]
 
 # Add any paths that contain templates here, relative to this directory.
@@ -93,7 +94,7 @@
 # The theme to use for HTML and HTML Help pages.  See the documentation for
 # a list of builtin themes.
 #
-html_theme = 'alabaster'
+html_theme = 'sphinx_rtd_theme'
 
 # Theme options are theme-specific and customize the look and feel of a theme
 # further.  For a list of options available for each theme, see the
diff --git a/rclpy/package.xml b/rclpy/package.xml
index cb419d2a5..9e2f0b4fd 100644
--- a/rclpy/package.xml
+++ b/rclpy/package.xml
@@ -48,6 +48,7 @@
   <test_depend>test_msgs</test_depend>
 
   <doc_depend>python3-sphinx</doc_depend>
+  <doc_depend>python3-sphinx-rtd-theme</doc_depend>
 
   <export>
     <build_type>ament_cmake</build_type>
diff --git a/rclpy/rclpy/client.py b/rclpy/rclpy/client.py
index 4e5528f15..92e1b4782 100644
--- a/rclpy/rclpy/client.py
+++ b/rclpy/rclpy/client.py
@@ -101,7 +101,7 @@ def unblock(future):
 
     def call_async(self, request: SrvTypeRequest) -> Future:
         """
-        Make a service request and asyncronously get the result.
+        Make a service request and asynchronously get the result.
 
         :param request: The service request.
         :return: A future that completes when the request does.
@@ -170,6 +170,7 @@ def wait_for_service(self, timeout_sec: float = None) -> bool:
         """
         # TODO(sloretz) Return as soon as the service is available
         # This is a temporary implementation. The sleep time is arbitrary.
+        # https://github.com/ros2/rclpy/issues/58
         sleep_time = 0.25
         if timeout_sec is None:
             timeout_sec = float('inf')
diff --git a/rclpy/rclpy/executors.py b/rclpy/rclpy/executors.py
index 8cc3c618b..0ac57f2d9 100644
--- a/rclpy/rclpy/executors.py
+++ b/rclpy/rclpy/executors.py
@@ -79,7 +79,7 @@ def __exit__(self, t, v, tb):
             self._num_work_executing -= 1
             self._work_condition.notify_all()
 
-    def wait(self, timeout_sec=None):
+    def wait(self, timeout_sec: Optional[float] = None):
         """
         Wait until all work completes.
 
@@ -87,7 +87,7 @@ def wait(self, timeout_sec=None):
         :type timeout_sec: float or None
         :rtype: bool True if all work completed
         """
-        if timeout_sec is not None and timeout_sec < 0:
+        if timeout_sec is not None and timeout_sec < 0.0:
             timeout_sec = None
         # Wait for all work to complete
         with self._work_condition:
@@ -204,9 +204,9 @@ def shutdown(self, timeout_sec: float = None) -> bool:
                 self._is_shutdown = True
                 # Tell executor it's been shut down
                 self._guard.trigger()
-
-        if not self._work_tracker.wait(timeout_sec):
-            return False
+        if not self._is_shutdown:
+            if not self._work_tracker.wait(timeout_sec):
+                return False
 
         # Clean up stuff that won't be used anymore
         with self._nodes_lock:
@@ -464,7 +464,13 @@ async def handler(entity, gc, is_shutdown, work_tracker):
                     entity.callback_group.ending_execution(entity)
                     # Signal that work has been done so the next callback in a mutually exclusive
                     # callback group can get executed
-                    gc.trigger()
+
+                    # Catch expected error where calling executor.shutdown()
+                    # from callback causes the GuardCondition to be destroyed
+                    try:
+                        gc.trigger()
+                    except InvalidHandle:
+                        pass
         task = Task(
             handler, (entity, self._guard, self._is_shutdown, self._work_tracker),
             executor=self)
diff --git a/rclpy/rclpy/node.py b/rclpy/rclpy/node.py
index 983c4c49c..5b9c4c634 100644
--- a/rclpy/rclpy/node.py
+++ b/rclpy/rclpy/node.py
@@ -349,6 +349,9 @@ def declare_parameter(
         This method, if successful, will result in any callback registered with
         :func:`add_on_set_parameters_callback` to be called.
 
+        The name and type in the given descriptor is ignored, and should be specified using
+        the name argument to this function and the default value's type instead.
+
         :param name: Fully-qualified name of the parameter, including its namespace.
         :param value: Value of the parameter to declare.
         :param descriptor: Descriptor for the parameter to declare.
@@ -381,6 +384,8 @@ def declare_parameters(
 
         The tuples in the given parameter list shall contain the name for each parameter,
         optionally providing a value and a descriptor.
+        The name and type in the given descriptors are ignored, and should be specified using
+        the name argument to this function and the default value's type instead.
         For each entry in the list, a parameter with a name of "namespace.name"
         will be declared.
         The resulting value for each declared parameter will be returned, considering
@@ -750,7 +755,7 @@ def set_parameters_atomically(self, parameter_list: List[Parameter]) -> SetParam
         allowed for the node, this method will raise a ParameterNotDeclaredException exception.
 
         Parameters are set all at once.
-        If setting a parameter fails due to not being declared, then no parameter will be set set.
+        If setting a parameter fails due to not being declared, then no parameter will be set.
         Either all of the parameters are set or none of them are set.
 
         If undeclared parameters are allowed for the node, then all the parameters will be
diff --git a/rclpy/rclpy/parameter.py b/rclpy/rclpy/parameter.py
index daad33146..eede4bcbf 100644
--- a/rclpy/rclpy/parameter.py
+++ b/rclpy/rclpy/parameter.py
@@ -14,9 +14,14 @@
 
 import array
 from enum import Enum
+from typing import Dict
+from typing import List
+from typing import Optional
 
 from rcl_interfaces.msg import Parameter as ParameterMsg
-from rcl_interfaces.msg import ParameterType, ParameterValue
+from rcl_interfaces.msg import ParameterType
+from rcl_interfaces.msg import ParameterValue
+import yaml
 
 PARAMETER_SEPARATOR_STRING = '.'
 
@@ -177,6 +182,50 @@ def to_parameter_msg(self):
         return ParameterMsg(name=self.name, value=self.get_parameter_value())
 
 
+def get_parameter_value(string_value: str) -> ParameterValue:
+    """
+    Guess the desired type of the parameter based on the string value.
+
+    :param string_value: The string value to be converted to a ParameterValue.
+    :return: The ParameterValue.
+    """
+    value = ParameterValue()
+    try:
+        yaml_value = yaml.safe_load(string_value)
+    except yaml.parser.ParserError:
+        yaml_value = string_value
+
+    if isinstance(yaml_value, bool):
+        value.type = ParameterType.PARAMETER_BOOL
+        value.bool_value = yaml_value
+    elif isinstance(yaml_value, int):
+        value.type = ParameterType.PARAMETER_INTEGER
+        value.integer_value = yaml_value
+    elif isinstance(yaml_value, float):
+        value.type = ParameterType.PARAMETER_DOUBLE
+        value.double_value = yaml_value
+    elif isinstance(yaml_value, list):
+        if all((isinstance(v, bool) for v in yaml_value)):
+            value.type = ParameterType.PARAMETER_BOOL_ARRAY
+            value.bool_array_value = yaml_value
+        elif all((isinstance(v, int) for v in yaml_value)):
+            value.type = ParameterType.PARAMETER_INTEGER_ARRAY
+            value.integer_array_value = yaml_value
+        elif all((isinstance(v, float) for v in yaml_value)):
+            value.type = ParameterType.PARAMETER_DOUBLE_ARRAY
+            value.double_array_value = yaml_value
+        elif all((isinstance(v, str) for v in yaml_value)):
+            value.type = ParameterType.PARAMETER_STRING_ARRAY
+            value.string_array_value = yaml_value
+        else:
+            value.type = ParameterType.PARAMETER_STRING
+            value.string_value = string_value
+    else:
+        value.type = ParameterType.PARAMETER_STRING
+        value.string_value = yaml_value
+    return value
+
+
 def parameter_value_to_python(parameter_value: ParameterValue):
     """
     Get the value for the Python builtin type from a rcl_interfaces/msg/ParameterValue object.
@@ -211,3 +260,79 @@ def parameter_value_to_python(parameter_value: ParameterValue):
         raise RuntimeError(f'unexpected parameter type {parameter_value.type}')
 
     return value
+
+
+def parameter_dict_from_yaml_file(
+    parameter_file: str,
+    use_wildcard: bool = False,
+    target_nodes: Optional[List[str]] = None,
+    namespace: str = ''
+) -> Dict[str, ParameterMsg]:
+    """
+    Build a dict of parameters from a YAML file.
+
+    Will load all parameters if ``target_nodes`` is None or empty.
+
+    :raises RuntimeError: if a target node is not in the file
+    :raises RuntimeError: if the is not a valid ROS parameter file
+
+    :param parameter_file: Path to the YAML file to load parameters from.
+    :param use_wildcard: Use wildcard matching for the target nodes.
+    :param target_nodes: List of nodes in the YAML file to load parameters from.
+    :param namespace: Namespace to prepend to all parameters.
+    :return: A dict of Parameter messages keyed by the parameter names
+    """
+    with open(parameter_file, 'r') as f:
+        param_file = yaml.safe_load(f)
+        param_keys = []
+        param_dict = {}
+
+        if use_wildcard and '/**' in param_file:
+            param_keys.append('/**')
+
+        if target_nodes:
+            for n in target_nodes:
+                if n not in param_file.keys():
+                    raise RuntimeError(f'Param file does not contain parameters for {n},'
+                                       f'only for nodes: {list(param_file.keys())} ')
+                param_keys.append(n)
+        else:
+            # wildcard key must go to the front of param_keys so that
+            # node-namespaced parameters will override the wildcard parameters
+            keys = set(param_file.keys())
+            keys.discard('/**')
+            param_keys.extend(keys)
+
+        if len(param_keys) == 0:
+            raise RuntimeError('Param file does not contain selected parameters')
+
+        for n in param_keys:
+            value = param_file[n]
+            if type(value) != dict or 'ros__parameters' not in value:
+                raise RuntimeError(f'YAML file is not a valid ROS parameter file for node {n}')
+            param_dict.update(value['ros__parameters'])
+        return _unpack_parameter_dict(namespace, param_dict)
+
+
+def _unpack_parameter_dict(namespace, parameter_dict):
+    """
+    Flatten a parameter dictionary recursively.
+
+    :param namespace: The namespace to prepend to the parameter names.
+    :param parameter_dict: A dictionary of parameters keyed by the parameter names
+    :return: A dict of Parameter objects keyed by the parameter names
+    """
+    parameters: Dict[str, ParameterMsg] = {}
+    for param_name, param_value in parameter_dict.items():
+        full_param_name = namespace + param_name
+        # Unroll nested parameters
+        if type(param_value) == dict:
+            parameters.update(_unpack_parameter_dict(
+                    namespace=full_param_name + PARAMETER_SEPARATOR_STRING,
+                    parameter_dict=param_value))
+        else:
+            parameter = ParameterMsg()
+            parameter.name = full_param_name
+            parameter.value = get_parameter_value(str(param_value))
+            parameters[full_param_name] = parameter
+    return parameters
diff --git a/rclpy/rclpy/parameter_client.py b/rclpy/rclpy/parameter_client.py
new file mode 100644
index 000000000..48b97ad25
--- /dev/null
+++ b/rclpy/rclpy/parameter_client.py
@@ -0,0 +1,331 @@
+# Copyright 2022 Open Source Robotics Foundation, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time
+from typing import Callable, List, Optional, Sequence, Union
+
+from rcl_interfaces.msg import Parameter as ParameterMsg
+from rcl_interfaces.srv import DescribeParameters
+from rcl_interfaces.srv import GetParameters
+from rcl_interfaces.srv import GetParameterTypes
+from rcl_interfaces.srv import ListParameters
+from rcl_interfaces.srv import SetParameters
+from rcl_interfaces.srv import SetParametersAtomically
+from rclpy.callback_groups import CallbackGroup
+from rclpy.node import Node
+from rclpy.parameter import Parameter as Parameter
+from rclpy.parameter import parameter_dict_from_yaml_file
+from rclpy.qos import qos_profile_services_default
+from rclpy.qos import QoSProfile
+from rclpy.task import Future
+
+
+class AsyncParameterClient:
+    def __init__(
+            self,
+            node: Node,
+            remote_node_name: str,
+            qos_profile: QoSProfile = qos_profile_services_default,
+            callback_group: Optional[CallbackGroup] = None):
+        """
+        Create an AsyncParameterClient.
+
+        An AsyncParameterClient that uses services offered by a remote node
+        to query and modify parameters in a streamlined way.
+        
+        Usage example:
+
+        .. code-block:: python
+
+            import rclpy
+            from rclpy.parameter import Parameter
+            node = rclpy.create_node('my_node')
+
+            client = AsyncParameterClient(node, 'example_node')
+
+            # set parameters on example node
+            future = client.set_parameters([
+                Parameter('int_param', Parameter.Type.INTEGER, 88),
+                Parameter('string/param', Parameter.Type.STRING, 'hello world').to_parameter_msg(),
+            ])
+            self.executor.spin_until_future_complete(future)
+            results = future.result()  # rcl_interfaces.srv.SetParameters.Response
+
+        For more on service names, see: `ROS 2 docs`_.
+
+        .. _ROS 2 docs: https://docs.ros.org/en/rolling/Concepts/About-ROS-2-Parameters.html#interacting-with-parameters # noqa E501
+
+        :param node: Node used to create clients that will interact with the remote node
+        :param remote_node_name: Name of remote node for which the parameters will be managed
+        """
+        self.remote_node_name = remote_node_name
+        self.node = node
+        self._get_parameter_client = self.node.create_client(
+            GetParameters, f'{remote_node_name}/get_parameters',
+            qos_profile=qos_profile, callback_group=callback_group
+        )
+        self._list_parameter_client = self.node.create_client(
+            ListParameters, f'{remote_node_name}/list_parameters',
+            qos_profile=qos_profile, callback_group=callback_group
+        )
+        self._set_parameter_client = self.node.create_client(
+            SetParameters, f'{remote_node_name}/set_parameters',
+            qos_profile=qos_profile, callback_group=callback_group
+        )
+        self._get_parameter_types_client = self.node.create_client(
+            GetParameterTypes, f'{remote_node_name}/get_parameter_types',
+            qos_profile=qos_profile, callback_group=callback_group
+        )
+        self._describe_parameters_client = self.node.create_client(
+            DescribeParameters, f'{remote_node_name}/describe_parameters',
+            qos_profile=qos_profile, callback_group=callback_group
+        )
+        self._set_parameters_atomically_client = self.node.create_client(
+            SetParametersAtomically, f'{remote_node_name}/set_parameters_atomically',
+            qos_profile=qos_profile, callback_group=callback_group
+        )
+
+    def services_are_ready(self) -> bool:
+        """
+        Check if all services are ready.
+
+        :return: ``True`` if all services are available, False otherwise.
+        """
+        return all([
+            self._list_parameter_client.service_is_ready(),
+            self._set_parameter_client.service_is_ready(),
+            self._get_parameter_client.service_is_ready(),
+            self._get_parameter_types_client.service_is_ready(),
+            self._describe_parameters_client.service_is_ready(),
+            self._set_parameters_atomically_client.service_is_ready(),
+        ])
+
+    def wait_for_services(self, timeout_sec: Optional[float] = None) -> bool:
+        """
+        Wait for all parameter services to be available.
+
+        :param timeout_sec: Seconds to wait. If ``None``, then wait forever.
+        :return: ``True`` if all services becomes avaliable, ``False`` otherwise.
+        """
+        # TODO(ihasdapie) See: rclpy.Client.wait_for_service
+        sleep_time = 0.25
+        if timeout_sec is None:
+            timeout_sec = float('inf')
+        while not self.services_are_ready() and timeout_sec > 0.0:
+            time.sleep(sleep_time)
+            timeout_sec -= sleep_time
+        return self.services_are_ready()
+
+    def list_parameters(
+        self,
+        prefixes: Optional[List[str]] = None,
+        depth: Optional[int] = None,
+        callback: Optional[Callable] = None
+    ) -> Future:
+        """
+        List all parameters with given prefixes.
+
+        :param prefixes: List of prefixes to filter by.
+        :param depth: Depth of the parameter tree to list. ``None`` means unlimited.
+        :param callback: Callback function to call when the request is complete.
+        :return: ``Future`` with the result of the request.
+        """
+        request = ListParameters.Request()
+        if prefixes:
+            request.prefixes = prefixes
+        if depth:
+            request.depth = depth
+        future = self._list_parameter_client.call_async(request)
+        if callback:
+            future.add_done_callback(callback)
+        return future
+
+    def get_parameters(self, names: List[str], callback: Optional[Callable] = None) -> Future:
+        """
+        Get parameters given names.
+
+        :param names: List of parameter names to get.
+        :param callback: Callback function to call when the request is complete.
+        :return: ``Future`` with the result of the request.
+        """
+        request = GetParameters.Request()
+        request.names = names
+        future = self._get_parameter_client.call_async(request)
+        if callback:
+            future.add_done_callback(callback)
+        return future
+
+    def set_parameters(
+        self,
+        parameters: Sequence[Union[Parameter, ParameterMsg]],
+        callback: Optional[Callable] = None
+    ) -> Future:
+        """
+        Set parameters given a list of parameters.
+
+        The result after the returned future is complete
+        will be of type ``rcl_interfaces.srv.SetParameters.Response``.
+
+        :param parameters: Sequence of parameters to set.
+        :param callback: Callback function to call when the request is complete.
+        :return: ``Future`` with the result of the request.
+        """
+        request = SetParameters.Request()
+        request.parameters = [
+            param.to_parameter_msg()
+            if isinstance(param, Parameter) else param
+            for param in parameters
+        ]
+        future = self._set_parameter_client.call_async(request)
+        if callback:
+            future.add_done_callback(callback)
+        return future
+
+    def describe_parameters(
+        self,
+        names: List[str],
+        callback: Optional[Callable] = None
+    ) -> Future:
+        """
+        Describe parameters given names.
+
+        The result after the returned future is complete
+        will be of type ``rcl_interfaces.srv.DescribeParameters.Response``.
+
+        :param names: List of parameter names to describe
+        :param callback: Callback function to call when the request is complete.
+        :return: ``Future`` with the result of the request.
+        """
+        request = DescribeParameters.Request()
+        request.names = names
+        future = self._describe_parameters_client.call_async(request)
+        if callback:
+            future.add_done_callback(callback)
+        return future
+
+    def get_parameter_types(
+        self,
+        names: List[str],
+        callback: Optional[Callable] = None
+    ) -> Future:
+        """
+        Get parameter types given names.
+
+        The result after the returned future is complete
+        will be of type ``rcl_interfaces.srv.GetParameterTypes.Response``.
+
+        Parameter type definitions are given in Parameter.Type
+
+        :param names: List of parameter names to get types for.
+        :param callback: Callback function to call when the request is complete.
+        :return: ``Future`` with the result of the request.
+        """
+        request = GetParameterTypes.Request()
+        request.names = names
+        future = self._get_parameter_types_client.call_async(request)
+        if callback:
+            future.add_done_callback(callback)
+        return future
+
+    def set_parameters_atomically(
+        self,
+        parameters: Sequence[Union[Parameter, ParameterMsg]],
+        callback: Optional[Callable] = None
+    ) -> Future:
+        """
+        Set parameters atomically.
+
+        The result after the returned future is complete
+        will be of type ``rcl_interfaces.srv.SetParametersAtomically.Response``.
+
+        :param parameters: Sequence of parameters to set.
+        :param callback: Callback function to call when the request is complete.
+        :return: ``Future`` with the result of the request.
+        """
+        request = SetParametersAtomically.Request()
+        request.parameters = [
+            param.to_parameter_msg()
+            if isinstance(param, Parameter) else param
+            for param in parameters
+        ]
+        future = self._set_parameters_atomically_client.call_async(request)
+        if callback:
+            future.add_done_callback(callback)
+        return future
+
+    def delete_parameters(
+        self, names: List[str], callback: Optional[Callable] = None
+    ) -> Future:
+        """
+        Unset parameters with given names.
+
+        The result after the returned future is complete
+        will be of type ``rcl_interfaces.srv.SetParameters.Response``.
+
+        Note: Only parameters that have been declared as dynamically typed can be unset.
+
+        :param names: List of parameter names to unset.
+        :param callback: Callback function to call when the request is complete.
+        :return: ``Future`` with the result of the request.
+        """
+        request = SetParameters.Request()
+        request.parameters = [Parameter(name=i).to_parameter_msg() for i in names]
+        future = self._set_parameter_client.call_async(request)
+        if callback:
+            future.add_done_callback(callback)
+        return future
+
+    def load_parameter_file(
+        self,
+        parameter_file: str,
+        use_wildcard: bool = False,
+        callback: Optional[Callable] = None
+    ) -> Future:
+        """
+        Load parameters from a yaml file.
+
+        Wrapper around `rclpy.parameter.parameter_dict_from_yaml_file`.
+
+        The result after the returned future is complete
+        will be of type ``rcl_interfaces.srv.SetParameters.Response``.
+
+        :param parameter_file: Path to the parameter file.
+        :param use_wildcard: Whether to use wildcard expansion.
+        :return: Future with the result from the set_parameters call.
+        """
+        param_dict = parameter_dict_from_yaml_file(parameter_file, use_wildcard)
+        future = self.set_parameters(list(param_dict.values()), callback=callback)
+        return future
+
+    def load_parameter_file_atomically(
+        self,
+        parameter_file: str,
+        use_wildcard: bool = False,
+        callback: Optional[Callable] = None
+    ) -> Future:
+        """
+        Load parameters from a yaml file atomically.
+
+        Wrapper around `rclpy.parameter.parameter_dict_from_yaml_file`.
+
+        The result after the returned future is complete
+        will be of type ``rcl_interfaces.srv.SetParameters.Response``.
+
+        :param parameter_file: Path to the parameter file.
+        :param use_wildcard: Whether to use wildcard expansion.
+        :return: Future with the result from the set_parameters_atomically call.
+        """
+        param_dict = parameter_dict_from_yaml_file(parameter_file, use_wildcard)
+        future = self.set_parameters_atomically(list(param_dict.values()), callback=callback)
+        return future
diff --git a/rclpy/test/test_executor.py b/rclpy/test/test_executor.py
index 65f878ac5..96e84ae22 100644
--- a/rclpy/test/test_executor.py
+++ b/rclpy/test/test_executor.py
@@ -534,6 +534,25 @@ def timer_callback():
             executor.shutdown()
             self.node.destroy_timer(tmr)
 
+    def shutdown_executor_from_callback(self):
+        """https://github.com/ros2/rclpy/issues/944: allow for executor shutdown from callback."""
+        self.assertIsNotNone(self.node.handle)
+        timer_period = 0.1
+        executor = SingleThreadedExecutor(context=self.context)
+        shutdown_event = threading.Event()
+
+        def timer_callback():
+            nonlocal shutdown_event, executor
+            executor.shutdown()
+            shutdown_event.set()
+
+        tmr = self.node.create_timer(timer_period, timer_callback)
+        executor.add_node(self.node)
+        t = threading.Thread(target=executor.spin, daemon=True)
+        t.start()
+        self.assertTrue(shutdown_event.wait(120))
+        self.node.destroy_timer(tmr)
+
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/rclpy/test/test_parameter.py b/rclpy/test/test_parameter.py
index 4a4a7f7d3..ce89a0497 100644
--- a/rclpy/test/test_parameter.py
+++ b/rclpy/test/test_parameter.py
@@ -13,14 +13,16 @@
 # limitations under the License.
 
 from array import array
+import os
+from tempfile import NamedTemporaryFile
 import unittest
 
 import pytest
-
 from rcl_interfaces.msg import Parameter as ParameterMsg
 from rcl_interfaces.msg import ParameterType
 from rcl_interfaces.msg import ParameterValue
 from rclpy.parameter import Parameter
+from rclpy.parameter import parameter_dict_from_yaml_file
 from rclpy.parameter import parameter_value_to_python
 
 
@@ -213,6 +215,29 @@ def test_parameter_value_to_python(self):
         with pytest.raises(RuntimeError):
             parameter_value_to_python(parameter_value)
 
+    def test_parameter_dict_from_yaml_file(self):
+        yaml_string = """/param_test_target:
+            ros__parameters:
+                param_1: 1
+                param_str: string
+            """
+        expected = {
+            'param_1': Parameter('param_1', Parameter.Type.INTEGER, 1).to_parameter_msg(),
+            'param_str': Parameter('param_str', Parameter.Type.STRING, 'string').to_parameter_msg()
+        }
+
+        try:
+            with NamedTemporaryFile(mode='w', delete=False) as f:
+                f.write(yaml_string)
+                f.flush()
+                f.close()
+                parameter_dict = parameter_dict_from_yaml_file(f.name)
+            assert parameter_dict == expected
+        finally:
+            if os.path.exists(f.name):
+                os.unlink(f.name)
+        self.assertRaises(FileNotFoundError, parameter_dict_from_yaml_file, 'unknown_file')
+
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/rclpy/test/test_parameter_client.py b/rclpy/test/test_parameter_client.py
new file mode 100644
index 000000000..7c4882fc7
--- /dev/null
+++ b/rclpy/test/test_parameter_client.py
@@ -0,0 +1,172 @@
+# Copyright 2022 Open Source Robotics Foundation, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+from tempfile import NamedTemporaryFile
+import unittest
+
+import rcl_interfaces.msg
+from rcl_interfaces.msg import ParameterType
+import rcl_interfaces.srv
+import rclpy
+import rclpy.context
+from rclpy.executors import SingleThreadedExecutor
+from rclpy.parameter import Parameter
+from rclpy.parameter_client import AsyncParameterClient
+
+
+class TestParameterClient(unittest.TestCase):
+
+    def setUp(self):
+        self.context = rclpy.context.Context()
+        rclpy.init(context=self.context)
+        self.client_node = rclpy.create_node(
+            'test_parameter_client',
+            namespace='/rclpy',
+            context=self.context)
+        self.target_node = rclpy.create_node(
+            'test_parameter_client_target',
+            namespace='/rclpy',
+            allow_undeclared_parameters=True,
+            context=self.context)
+        self.target_node.declare_parameter('int_arr_param', [1, 2, 3])
+        self.target_node.declare_parameter('float.param..', 3.14)
+
+        self.client = AsyncParameterClient(self.client_node, 'test_parameter_client_target')
+        self.executor = SingleThreadedExecutor(context=self.context)
+        self.executor.add_node(self.client_node)
+        self.executor.add_node(self.target_node)
+
+    def tearDown(self):
+        self.executor.shutdown()
+        self.client_node.destroy_node()
+        self.target_node.destroy_node()
+        rclpy.shutdown(context=self.context)
+
+    def test_set_parameter(self):
+        future = self.client.set_parameters([
+            Parameter('int_param', Parameter.Type.INTEGER, 88).to_parameter_msg(),
+            Parameter('string.param', Parameter.Type.STRING, 'hello world'),
+        ])
+        self.executor.spin_until_future_complete(future)
+        results = future.result()
+        assert results is not None
+        assert len(results.results) == 2
+        res = [i.successful for i in results.results]
+        assert all(res)
+
+    def test_get_parameter(self):
+        future = self.client.get_parameters(['int_arr_param', 'float.param..'])
+        self.executor.spin_until_future_complete(future)
+        results = future.result()
+        assert results is not None
+        assert len(results.values) == 2
+        assert list(results.values[0].integer_array_value) == [1, 2, 3]
+        assert results.values[1].double_value == 3.14
+
+    def test_list_parameters(self):
+        future = self.client.list_parameters()
+        self.executor.spin_until_future_complete(future)
+        results = future.result()
+        assert results is not None
+        assert 'int_arr_param' in results.result.names
+        assert 'float.param..' in results.result.names
+
+    def test_describe_parameters(self):
+        future = self.client.describe_parameters(['int_arr_param'])
+        self.executor.spin_until_future_complete(future)
+        results = future.result()
+        assert results is not None
+        assert len(results.descriptors) == 1
+        assert results.descriptors[0].type == ParameterType.PARAMETER_INTEGER_ARRAY
+        assert results.descriptors[0].name == 'int_arr_param'
+
+    def test_get_paramter_types(self):
+        future = self.client.get_parameter_types(['int_arr_param'])
+        self.executor.spin_until_future_complete(future)
+        results = future.result()
+        assert results is not None
+        assert len(results.types) == 1
+        assert results.types[0] == ParameterType.PARAMETER_INTEGER_ARRAY
+
+    def test_set_parameters_atomically(self):
+        future = self.client.set_parameters_atomically([
+            Parameter('int_param', Parameter.Type.INTEGER, 888),
+            Parameter('string.param', Parameter.Type.STRING, 'Hello World').to_parameter_msg(),
+        ])
+        self.executor.spin_until_future_complete(future)
+        results = future.result()
+        assert results is not None
+        assert results.result.successful
+
+    def test_delete_parameters(self):
+        self.target_node.declare_parameter('delete_param', 10)
+        descriptor = rcl_interfaces.msg.ParameterDescriptor(dynamic_typing=True)
+        self.target_node.declare_parameter('delete_param_dynamic', 10, descriptor=descriptor)
+
+        future = self.client.delete_parameters(['delete_param'])
+        self.executor.spin_until_future_complete(future)
+        result = future.result()
+        assert result is not None
+        assert len(result.results) == 1
+        assert not result.results[0].successful
+        assert result.results[0].reason == 'Static parameter cannot be undeclared'
+
+        future = self.client.delete_parameters(['delete_param_dynamic'])
+        self.executor.spin_until_future_complete(future)
+        result = future.result()
+        assert result is not None
+        assert len(result.results) == 1
+        assert result.results[0].successful
+
+    def test_load_parameter_file(self):
+        yaml_string = """/param_test_target:
+            ros__parameters:
+                param_1: 1
+                param_str: "string"
+            """
+        try:
+            with NamedTemporaryFile(mode='w', delete=False) as f:
+                f.write(yaml_string)
+                f.flush()
+                f.close()
+                future = self.client.load_parameter_file(f.name)
+            self.executor.spin_until_future_complete(future)
+            result = future.result()
+            assert result is not None
+            assert len(result.results) == 2
+            assert all([i.successful for i in result.results])
+        finally:
+            if os.path.exists(f.name):
+                os.unlink(f.name)
+
+    def test_load_parameter_file_atomically(self):
+        yaml_string = """/param_test_target:
+            ros__parameters:
+                param_1: 1
+                param_str: "string"
+            """
+        try:
+            with NamedTemporaryFile(mode='w', delete=False) as f:
+                f.write(yaml_string)
+                f.flush()
+                f.close()
+                future = self.client.load_parameter_file_atomically(f.name)
+            self.executor.spin_until_future_complete(future)
+            result = future.result()
+            assert result is not None
+            assert result.result.successful
+        finally:
+            if os.path.exists(f.name):
+                os.unlink(f.name)