Repository: redis/redis-py
Pull request #1821 (2021-12-21 10:30:28)

Retry on error exception and timeout fixes
…onError import

### Pull Request check-list

_Please make sure to review and check all of these items:_

- [X ] Does `$ tox` pass with this change (including linting)?
- [ X] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [X ] Is the new or changed code fully tested?
- [ X] Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
- [X ] Is there an example added to the examples folder (if applicable)?

_NOTE: these things are not required to open a PR and can be done
afterwards / while the PR is open._

### Description of change

1. Added timeout error to the list of errors to retry.
2. Changed TimeoutError import from redis.exception to RedisTimeoutError so python's TimeoutError won't get override, as it can be also thrown when we try to establish a connection.
3. A ConnectionError import was added from redis.exceptions, since it was mistakenly not imported and the default ConnectionError was being used.
4. Increased sleep time to 0.25sec when ClusterDownError is thrown, so it will let more time for the cluster to get healed. 


Patch:
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 04f989a3..0f9ca0b1 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -32,7 +32,7 @@ Here's how to get started with your code contribution:
 
 1.  Create your own fork of redis-py
 2.  Do the changes in your fork
-3.  
+3.
     *Create a virtualenv and install the development dependencies from the dev_requirements.txt file:*
 
         a.  python -m venv .venv
@@ -69,12 +69,12 @@ configuration](https://redis.io/topics/sentinel).
 
 ## Testing
 
-Call `invoke tests` to run all tests, or `invoke all-tests` to run linters 
-tests as well. With the 'tests' and 'all-tests' targets, all Redis and 
-RedisCluster tests will be run. 
+Call `invoke tests` to run all tests, or `invoke all-tests` to run linters
+tests as well. With the 'tests' and 'all-tests' targets, all Redis and
+RedisCluster tests will be run.
 
-It is possible to run only Redis client tests (with cluster mode disabled) by 
-using `invoke standalone-tests`; similarly, RedisCluster tests can be run by using 
+It is possible to run only Redis client tests (with cluster mode disabled) by
+using `invoke standalone-tests`; similarly, RedisCluster tests can be run by using
 `invoke cluster-tests`.
 
 Each run of tox starts and stops the various dockers required. Sometimes
@@ -85,9 +85,7 @@ tests against multiple versions of python. Feel free to test your
 changes against all the python versions supported, as declared by the
 tox.ini file (eg: tox -e py39). If you have the various python versions
 on your desktop, you can run *tox* by itself, to test all supported
-versions. Alternatively, as your source code is mounted in the
-**lots-of-pythons** docker, you can start exploring from there, with all
-supported python versions!
+versions.
 
 ### Docker Tips
 
diff --git a/docs/redismodules.rst b/docs/redismodules.rst
index 0cb8c496..53a50081 100644
--- a/docs/redismodules.rst
+++ b/docs/redismodules.rst
@@ -44,7 +44,7 @@ These are the commands for interacting with the `RedisJSON module <https://redis
 RediSearch Commands
 *******************
 
-These are the commands for interacting with the `RediSearch module <https://redisjson.io>`_. Below is a brief example, as well as documentation on the commands themselves.
+These are the commands for interacting with the `RediSearch module <https://redisearch.io>`_. Below is a brief example, as well as documentation on the commands themselves.
 
 **Create a search index, and display its information**
 
@@ -64,7 +64,7 @@ These are the commands for interacting with the `RediSearch module <https://redi
 RedisGraph Commands
 *******************
 
-These are the commands for interacting with the `RedisGraph module <https://redisjson.io>`_. Below is a brief example, as well as documentation on the commands themselves.
+These are the commands for interacting with the `RedisGraph module <https://redisgraph.io>`_. Below is a brief example, as well as documentation on the commands themselves.
 
 **Create a graph, adding two nodes**
 
@@ -97,7 +97,7 @@ These are the commands for interacting with the `RedisGraph module <https://redi
 RedisBloom Commands
 *******************
 
-These are the commands for interacting with the `RedisBloom module <https://redisjson.io>`_. Below is a brief example, as well as documentation on the commands themselves.
+These are the commands for interacting with the `RedisBloom module <https://redisbloom.io>`_. Below is a brief example, as well as documentation on the commands themselves.
 
 **Create and add to a bloom filter**
 
diff --git a/redis/client.py b/redis/client.py
index ae4fae2a..5116482c 100755
--- a/redis/client.py
+++ b/redis/client.py
@@ -768,6 +768,7 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
         "STRALGO": parse_stralgo,
         "PUBSUB NUMSUB": parse_pubsub_numsub,
         "RANDOMKEY": lambda r: r and r or None,
+        "RESET": str_if_bytes,
         "SCAN": parse_scan,
         "SCRIPT EXISTS": lambda r: list(map(bool, r)),
         "SCRIPT FLUSH": bool_ok,
@@ -868,6 +869,7 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
         errors=None,
         decode_responses=False,
         retry_on_timeout=False,
+        retry_on_error=[],
         ssl=False,
         ssl_keyfile=None,
         ssl_certfile=None,
@@ -886,8 +888,10 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
     ):
         """
         Initialize a new Redis client.
-        To specify a retry policy, first set `retry_on_timeout` to `True`
-        then set `retry` to a valid `Retry` object
+        To specify a retry policy for specific errors, first set
+        `retry_on_error` to a list of the error/s to retry on, then set
+        `retry` to a valid `Retry` object.
+        To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
         if not connection_pool:
             if charset is not None:
@@ -904,7 +908,8 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
                     )
                 )
                 encoding_errors = errors
-
+            if retry_on_timeout is True:
+                retry_on_error.append(TimeoutError)
             kwargs = {
                 "db": db,
                 "username": username,
@@ -913,7 +918,7 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
                 "encoding": encoding,
                 "encoding_errors": encoding_errors,
                 "decode_responses": decode_responses,
-                "retry_on_timeout": retry_on_timeout,
+                "retry_on_error": retry_on_error,
                 "retry": copy.deepcopy(retry),
                 "max_connections": max_connections,
                 "health_check_interval": health_check_interval,
@@ -1145,11 +1150,14 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
     def _disconnect_raise(self, conn, error):
         """
         Close the connection and raise an exception
-        if retry_on_timeout is not set or the error
-        is not a TimeoutError
+        if retry_on_error is not set or the error
+        is not one of the specified error types
         """
         conn.disconnect()
-        if not (conn.retry_on_timeout and isinstance(error, TimeoutError)):
+        if (
+            conn.retry_on_error is None
+            or isinstance(error, tuple(conn.retry_on_error)) is False
+        ):
             raise error
 
     # COMMAND EXECUTION AND PROTOCOL PARSING
@@ -1280,18 +1288,17 @@ class PubSub:
         self.shard_hint = shard_hint
         self.ignore_subscribe_messages = ignore_subscribe_messages
         self.connection = None
+        self.subscribed_event = threading.Event()
         # we need to know the encoding options for this connection in order
         # to lookup channel and pattern names for callback handlers.
         self.encoder = encoder
         if self.encoder is None:
             self.encoder = self.connection_pool.get_encoder()
+        self.health_check_response_b = self.encoder.encode(self.HEALTH_CHECK_MESSAGE)
         if self.encoder.decode_responses:
             self.health_check_response = ["pong", self.HEALTH_CHECK_MESSAGE]
         else:
-            self.health_check_response = [
-                b"pong",
-                self.encoder.encode(self.HEALTH_CHECK_MESSAGE),
-            ]
+            self.health_check_response = [b"pong", self.health_check_response_b]
         self.reset()
 
     def __enter__(self):
@@ -1316,9 +1323,11 @@ class PubSub:
             self.connection_pool.release(self.connection)
             self.connection = None
         self.channels = {}
+        self.health_check_response_counter = 0
         self.pending_unsubscribe_channels = set()
         self.patterns = {}
         self.pending_unsubscribe_patterns = set()
+        self.subscribed_event.clear()
 
     def close(self):
         self.reset()
@@ -1344,7 +1353,7 @@ class PubSub:
     @property
     def subscribed(self):
         "Indicates if there are subscriptions to any channels or patterns"
-        return bool(self.channels or self.patterns)
+        return self.subscribed_event.is_set()
 
     def execute_command(self, *args):
         "Execute a publish/subscribe command"
@@ -1362,8 +1371,28 @@ class PubSub:
             self.connection.register_connect_callback(self.on_connect)
         connection = self.connection
         kwargs = {"check_health": not self.subscribed}
+        if not self.subscribed:
+            self.clean_health_check_responses()
         self._execute(connection, connection.send_command, *args, **kwargs)
 
+    def clean_health_check_responses(self):
+        """
+        If any health check responses are present, clean them
+        """
+        ttl = 10
+        conn = self.connection
+        while self.health_check_response_counter > 0 and ttl > 0:
+            if self._execute(conn, conn.can_read, timeout=conn.socket_timeout):
+                response = self._execute(conn, conn.read_response)
+                if self.is_health_check_response(response):
+                    self.health_check_response_counter -= 1
+                else:
+                    raise PubSubError(
+                        "A non health check response was cleaned by "
+                        "execute_command: {0}".format(response)
+                    )
+            ttl -= 1
+
     def _disconnect_raise_connect(self, conn, error):
         """
         Close the connection and raise an exception
@@ -1403,11 +1432,23 @@ class PubSub:
             return None
         response = self._execute(conn, conn.read_response)
 
-        if conn.health_check_interval and response == self.health_check_response:
+        if self.is_health_check_response(response):
             # ignore the health check message as user might not expect it
+            self.health_check_response_counter -= 1
             return None
         return response
 
+    def is_health_check_response(self, response):
+        """
+        Check if the response is a health check response.
+        If there are no subscriptions redis responds to PING command with a
+        bulk response, instead of a multi-bulk with "pong" and the response.
+        """
+        return response in [
+            self.health_check_response,  # If there was a subscription
+            self.health_check_response_b,  # If there wasn't
+        ]
+
     def check_health(self):
         conn = self.connection
         if conn is None:
@@ -1418,6 +1459,7 @@ class PubSub:
 
         if conn.health_check_interval and time.time() > conn.next_health_check:
             conn.send_command("PING", self.HEALTH_CHECK_MESSAGE, check_health=False)
+            self.health_check_response_counter += 1
 
     def _normalize_keys(self, data):
         """
@@ -1447,6 +1489,11 @@ class PubSub:
         # for the reconnection.
         new_patterns = self._normalize_keys(new_patterns)
         self.patterns.update(new_patterns)
+        if not self.subscribed:
+            # Set the subscribed_event flag to True
+            self.subscribed_event.set()
+            # Clear the health check counter
+            self.health_check_response_counter = 0
         self.pending_unsubscribe_patterns.difference_update(new_patterns)
         return ret_val
 
@@ -1481,6 +1528,11 @@ class PubSub:
         # for the reconnection.
         new_channels = self._normalize_keys(new_channels)
         self.channels.update(new_channels)
+        if not self.subscribed:
+            # Set the subscribed_event flag to True
+            self.subscribed_event.set()
+            # Clear the health check counter
+            self.health_check_response_counter = 0
         self.pending_unsubscribe_channels.difference_update(new_channels)
         return ret_val
 
@@ -1512,6 +1564,20 @@ class PubSub:
         before returning. Timeout should be specified as a floating point
         number.
         """
+        if not self.subscribed:
+            # Wait for subscription
+            start_time = time.time()
+            if self.subscribed_event.wait(timeout) is True:
+                # The connection was subscribed during the timeout time frame.
+                # The timeout should be adjusted based on the time spent
+                # waiting for the subscription
+                time_spent = time.time() - start_time
+                timeout = max(0.0, timeout - time_spent)
+            else:
+                # The connection isn't subscribed to any channels or patterns,
+                # so no messages are available
+                return None
+
         response = self.parse_response(block=False, timeout=timeout)
         if response:
             return self.handle_message(response, ignore_subscribe_messages)
@@ -1530,6 +1596,8 @@ class PubSub:
         with a message handler, the handler is invoked instead of a parsed
         message being returned.
         """
+        if response is None:
+            return None
         message_type = str_if_bytes(response[0])
         if message_type == "pmessage":
             message = {
@@ -1565,6 +1633,10 @@ class PubSub:
                 if channel in self.pending_unsubscribe_channels:
                     self.pending_unsubscribe_channels.remove(channel)
                     self.channels.pop(channel, None)
+            if not self.channels and not self.patterns:
+                # There are no subscriptions anymore, set subscribed_event flag
+                # to false
+                self.subscribed_event.clear()
 
         if message_type in self.PUBLISH_MESSAGE_TYPES:
             # if there's a message handler, invoke it
diff --git a/redis/cluster.py b/redis/cluster.py
index b1adeb73..5707a9da 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -17,6 +17,7 @@ from redis.exceptions import (
     ClusterCrossSlotError,
     ClusterDownError,
     ClusterError,
+    ConnectionError,
     DataError,
     MasterDownError,
     MovedError,
@@ -374,6 +375,12 @@ class RedisCluster(RedisClusterCommands):
         ),
     )
 
+    ERRORS_ALLOW_RETRY = (
+        ConnectionError,
+        TimeoutError,
+        ClusterDownError,
+    )
+
     def __init__(
         self,
         host=None,
@@ -385,8 +392,6 @@ class RedisCluster(RedisClusterCommands):
         reinitialize_steps=10,
         read_from_replicas=False,
         url=None,
-        retry_on_timeout=False,
-        retry=None,
         **kwargs,
     ):
         """
@@ -417,11 +422,6 @@ class RedisCluster(RedisClusterCommands):
         :cluster_error_retry_attempts: 'int'
              Retry command execution attempts when encountering ClusterDownError
              or ConnectionError
-        :retry_on_timeout: 'bool'
-             To specify a retry policy, first set `retry_on_timeout` to `True`
-             then set `retry` to a valid `Retry` object
-        :retry: 'Retry'
-              a `Retry` object
         :reinitialize_steps: 'int'
             Specifies the number of MOVED errors that need to occur before
             reinitializing the whole cluster topology. If a MOVED error occurs
@@ -452,9 +452,6 @@ class RedisCluster(RedisClusterCommands):
                 "Argument 'db' is not possible to use in cluster mode"
             )
 
-        if retry_on_timeout:
-            kwargs.update({"retry_on_timeout": retry_on_timeout, "retry": retry})
-
         # Get the startup node/s
         from_url = False
         if url is not None:
@@ -850,7 +847,7 @@ class RedisCluster(RedisClusterCommands):
 
     def execute_command(self, *args, **kwargs):
         """
-        Wrapper for ClusterDownError and ConnectionError error handling.
+        Wrapper for ERRORS_ALLOW_RETRY error handling.
 
         It will try the number of times specified by the config option
         "self.cluster_error_retry_attempts" which defaults to 3 unless manually
@@ -865,18 +862,19 @@ class RedisCluster(RedisClusterCommands):
             dict<Any, ClusterNode>
         """
         target_nodes_specified = False
-        target_nodes = kwargs.pop("target_nodes", None)
-        if target_nodes is not None and not self._is_nodes_flag(target_nodes):
-            target_nodes = self._parse_target_nodes(target_nodes)
+        target_nodes = None
+        passed_targets = kwargs.pop("target_nodes", None)
+        if passed_targets is not None and not self._is_nodes_flag(passed_targets):
+            target_nodes = self._parse_target_nodes(passed_targets)
             target_nodes_specified = True
-        # If ClusterDownError/ConnectionError were thrown, the nodes
-        # and slots cache were reinitialized. We will retry executing the
-        # command with the updated cluster setup only when the target nodes
-        # can be determined again with the new cache tables. Therefore,
-        # when target nodes were passed to this function, we cannot retry
-        # the command execution since the nodes may not be valid anymore
-        # after the tables were reinitialized. So in case of passed target
-        # nodes, retry_attempts will be set to 1.
+        # If an error that allows retrying was thrown, the nodes and slots
+        # cache were reinitialized. We will retry executing the command with
+        # the updated cluster setup only when the target nodes can be
+        # determined again with the new cache tables. Therefore, when target
+        # nodes were passed to this function, we cannot retry the command
+        # execution since the nodes may not be valid anymore after the tables
+        # were reinitialized. So in case of passed target nodes,
+        # retry_attempts will be set to 1.
         retry_attempts = (
             1 if target_nodes_specified else self.cluster_error_retry_attempts
         )
@@ -887,7 +885,7 @@ class RedisCluster(RedisClusterCommands):
                 if not target_nodes_specified:
                     # Determine the nodes to execute the command on
                     target_nodes = self._determine_nodes(
-                        *args, **kwargs, nodes_flag=target_nodes
+                        *args, **kwargs, nodes_flag=passed_targets
                     )
                     if not target_nodes:
                         raise RedisClusterException(
@@ -897,11 +895,14 @@ class RedisCluster(RedisClusterCommands):
                     res[node.name] = self._execute_command(node, *args, **kwargs)
                 # Return the processed result
                 return self._process_result(args[0], res, **kwargs)
-            except (ClusterDownError, ConnectionError) as e:
-                # The nodes and slots cache were reinitialized.
-                # Try again with the new cluster setup. All other errors
-                # should be raised.
-                exception = e
+            except BaseException as e:
+                if type(e) in RedisCluster.ERRORS_ALLOW_RETRY:
+                    # The nodes and slots cache were reinitialized.
+                    # Try again with the new cluster setup.
+                    exception = e
+                else:
+                    # All other errors should be raised.
+                    raise e
 
         # If it fails the configured number of times then raise exception back
         # to caller of this method
@@ -953,11 +954,11 @@ class RedisCluster(RedisClusterCommands):
                     )
                 return response
 
-            except (RedisClusterException, BusyLoadingError):
-                log.exception("RedisClusterException || BusyLoadingError")
+            except (RedisClusterException, BusyLoadingError) as e:
+                log.exception(type(e))
                 raise
-            except ConnectionError:
-                log.exception("ConnectionError")
+            except (ConnectionError, TimeoutError) as e:
+                log.exception(type(e))
                 # ConnectionError can also be raised if we couldn't get a
                 # connection from the pool before timing out, so check that
                 # this is an actual connection before attempting to disconnect.
@@ -976,13 +977,6 @@ class RedisCluster(RedisClusterCommands):
                     # and try again with the new setup
                     self.nodes_manager.initialize()
                     raise
-            except TimeoutError:
-                log.exception("TimeoutError")
-                if connection is not None:
-                    connection.disconnect()
-
-                if ttl < self.RedisClusterRequestTTL / 2:
-                    time.sleep(0.05)
             except MovedError as e:
                 # First, we will try to patch the slots/nodes cache with the
                 # redirected node output and try again. If MovedError exceeds
@@ -996,9 +990,11 @@ class RedisCluster(RedisClusterCommands):
                 self.reinitialize_counter += 1
                 if self._should_reinitialized():
                     self.nodes_manager.initialize()
+                    # Reset the counter
+                    self.reinitialize_counter = 0
                 else:
                     self.nodes_manager.update_moved_exception(e)
-                    moved = True
+                moved = True
             except TryAgainError:
                 log.exception("TryAgainError")
 
@@ -1014,7 +1010,7 @@ class RedisCluster(RedisClusterCommands):
                 # ClusterDownError can occur during a failover and to get
                 # self-healed, we will try to reinitialize the cluster layout
                 # and retry executing the command
-                time.sleep(0.05)
+                time.sleep(0.25)
                 self.nodes_manager.initialize()
                 raise e
             except ResponseError as e:
@@ -1320,6 +1316,7 @@ class NodesManager:
         tmp_slots = {}
         disagreements = []
         startup_nodes_reachable = False
+        fully_covered = False
         kwargs = self.connection_kwargs
         for startup_node in self.startup_nodes.values():
             try:
@@ -1339,7 +1336,7 @@ class NodesManager:
                     raise RedisClusterException(
                         "Cluster mode is not enabled on this node"
                     )
-                cluster_slots = r.execute_command("CLUSTER SLOTS")
+                cluster_slots = str_if_bytes(r.execute_command("CLUSTER SLOTS"))
                 startup_nodes_reachable = True
             except (ConnectionError, TimeoutError) as e:
                 msg = e.__str__
@@ -1431,6 +1428,12 @@ class NodesManager:
                                     f'slots cache: {", ".join(disagreements)}'
                                 )
 
+            fully_covered = self.check_slots_coverage(tmp_slots)
+            if fully_covered:
+                # Don't need to continue to the next startup node if all
+                # slots are covered
+                break
+
         if not startup_nodes_reachable:
             raise RedisClusterException(
                 "Redis Cluster cannot be connected. Please provide at least "
@@ -1440,7 +1443,6 @@ class NodesManager:
         # Create Redis connections to all nodes
         self.create_redis_connections(list(tmp_nodes_cache.values()))
 
-        fully_covered = self.check_slots_coverage(tmp_slots)
         # Check if the slots are not fully covered
         if not fully_covered and self._require_full_coverage:
             # Despite the requirement that the slots be covered, there
@@ -1478,6 +1480,8 @@ class NodesManager:
         self.default_node = self.get_nodes_by_server_type(PRIMARY)[0]
         # Populate the startup nodes with all discovered nodes
         self.populate_startup_nodes(self.nodes_cache.values())
+        # If initialize was called after a MovedError, clear it
+        self._moved_exception = None
 
     def close(self):
         self.default_node = None
@@ -1621,21 +1625,20 @@ class ClusterPubSub(PubSub):
             return self.node.redis_connection
 
 
-ERRORS_ALLOW_RETRY = (
-    ConnectionError,
-    TimeoutError,
-    MovedError,
-    AskError,
-    TryAgainError,
-)
-
-
 class ClusterPipeline(RedisCluster):
     """
     Support for Redis pipeline
     in cluster mode
     """
 
+    ERRORS_ALLOW_RETRY = (
+        ConnectionError,
+        TimeoutError,
+        MovedError,
+        AskError,
+        TryAgainError,
+    )
+
     def __init__(
         self,
         nodes_manager,
@@ -1643,7 +1646,7 @@ class ClusterPipeline(RedisCluster):
         cluster_response_callbacks=None,
         startup_nodes=None,
         read_from_replicas=False,
-        cluster_error_retry_attempts=3,
+        cluster_error_retry_attempts=5,
         reinitialize_steps=10,
         **kwargs,
     ):
@@ -1905,7 +1908,11 @@ class ClusterPipeline(RedisCluster):
         # collect all the commands we are allowed to retry.
         # (MOVED, ASK, or connection errors or timeout errors)
         attempt = sorted(
-            (c for c in attempt if isinstance(c.result, ERRORS_ALLOW_RETRY)),
+            (
+                c
+                for c in attempt
+                if isinstance(c.result, ClusterPipeline.ERRORS_ALLOW_RETRY)
+            ),
             key=lambda x: x.position,
         )
         if attempt and allow_redirections:
diff --git a/redis/commands/core.py b/redis/commands/core.py
index 835ea612..0823315a 100644
--- a/redis/commands/core.py
+++ b/redis/commands/core.py
@@ -479,6 +479,104 @@ class ManagementCommands:
         """
         return self.execute_command("CLIENT ID", **kwargs)
 
+    def client_tracking_on(
+        self,
+        clientid=None,
+        prefix=[],
+        bcast=False,
+        optin=False,
+        optout=False,
+        noloop=False,
+    ):
+        """
+        Turn on the tracking mode.
+        For more information about the options look at client_tracking func.
+
+        See https://redis.io/commands/client-tracking
+        """
+        return self.client_tracking(
+            True, clientid, prefix, bcast, optin, optout, noloop
+        )
+
+    def client_tracking_off(
+        self,
+        clientid=None,
+        prefix=[],
+        bcast=False,
+        optin=False,
+        optout=False,
+        noloop=False,
+    ):
+        """
+        Turn off the tracking mode.
+        For more information about the options look at client_tracking func.
+
+        See https://redis.io/commands/client-tracking
+        """
+        return self.client_tracking(
+            False, clientid, prefix, bcast, optin, optout, noloop
+        )
+
+    def client_tracking(
+        self,
+        on=True,
+        clientid=None,
+        prefix=[],
+        bcast=False,
+        optin=False,
+        optout=False,
+        noloop=False,
+        **kwargs,
+    ):
+        """
+        Enables the tracking feature of the Redis server, that is used
+        for server assisted client side caching.
+
+        ``on`` indicate for tracking on or tracking off. The dafualt is on.
+
+        ``clientid`` send invalidation messages to the connection with
+        the specified ID.
+
+        ``bcast`` enable tracking in broadcasting mode. In this mode
+        invalidation messages are reported for all the prefixes
+        specified, regardless of the keys requested by the connection.
+
+        ``optin``  when broadcasting is NOT active, normally don't track
+        keys in read only commands, unless they are called immediately
+        after a CLIENT CACHING yes command.
+
+        ``optout`` when broadcasting is NOT active, normally track keys in
+        read only commands, unless they are called immediately after a
+        CLIENT CACHING no command.
+
+        ``noloop`` don't send notifications about keys modified by this
+        connection itself.
+
+        ``prefix``  for broadcasting, register a given key prefix, so that
+        notifications will be provided only for keys starting with this string.
+
+        See https://redis.io/commands/client-tracking
+        """
+
+        if len(prefix) != 0 and bcast is False:
+            raise DataError("Prefix can only be used with bcast")
+
+        pieces = ["ON"] if on else ["OFF"]
+        if clientid is not None:
+            pieces.extend(["REDIRECT", clientid])
+        for p in prefix:
+            pieces.extend(["PREFIX", p])
+        if bcast:
+            pieces.append("BCAST")
+        if optin:
+            pieces.append("OPTIN")
+        if optout:
+            pieces.append("OPTOUT")
+        if noloop:
+            pieces.append("NOLOOP")
+
+        return self.execute_command("CLIENT TRACKING", *pieces)
+
     def client_trackinginfo(self, **kwargs):
         """
         Returns the information about the current client connection's
@@ -510,16 +608,28 @@ class ManagementCommands:
             args.append(b"ERROR")
         return self.execute_command(*args, **kwargs)
 
-    def client_pause(self, timeout, **kwargs):
+    def client_pause(self, timeout, all=True, **kwargs):
         """
         Suspend all the Redis clients for the specified amount of time
         :param timeout: milliseconds to pause clients
 
         For more information check https://redis.io/commands/client-pause
-        """
+        :param all: If true (default) all client commands are blocked.
+             otherwise, clients are only blocked if they attempt to execute
+             a write command.
+             For the WRITE mode, some commands have special behavior:
+                 EVAL/EVALSHA: Will block client for all scripts.
+                 PUBLISH: Will block client.
+                 PFCOUNT: Will block client.
+                 WAIT: Acknowledgments will be delayed, so this command will
+                 appear blocked.
+        """
+        args = ["CLIENT PAUSE", str(timeout)]
         if not isinstance(timeout, int):
             raise DataError("CLIENT PAUSE timeout must be an integer")
-        return self.execute_command("CLIENT PAUSE", str(timeout), **kwargs)
+        if not all:
+            args.append("WRITE")
+        return self.execute_command(*args, **kwargs)
 
     def client_unpause(self, **kwargs):
         """
@@ -707,6 +817,13 @@ class ManagementCommands:
         else:
             return self.execute_command("LOLWUT", **kwargs)
 
+    def reset(self):
+        """Perform a full reset on the connection's server side contenxt.
+
+        See: https://redis.io/commands/reset
+        """
+        return self.execute_command("RESET")
+
     def migrate(
         self,
         host,
diff --git a/redis/commands/json/commands.py b/redis/commands/json/commands.py
index e7f07b61..a132b8ee 100644
--- a/redis/commands/json/commands.py
+++ b/redis/commands/json/commands.py
@@ -1,3 +1,6 @@
+import os
+from json import JSONDecodeError, loads
+
 from deprecated import deprecated
 
 from redis.exceptions import DataError
@@ -213,6 +216,54 @@ class JSONCommands:
             pieces.append("XX")
         return self.execute_command("JSON.SET", *pieces)
 
+    def set_file(self, name, path, file_name, nx=False, xx=False, decode_keys=False):
+        """
+        Set the JSON value at key ``name`` under the ``path`` to the content
+        of the json file ``file_name``.
+
+        ``nx`` if set to True, set ``value`` only if it does not exist.
+        ``xx`` if set to True, set ``value`` only if it exists.
+        ``decode_keys`` If set to True, the keys of ``obj`` will be decoded
+        with utf-8.
+
+        """
+
+        with open(file_name, "r") as fp:
+            file_content = loads(fp.read())
+
+        return self.set(name, path, file_content, nx=nx, xx=xx, decode_keys=decode_keys)
+
+    def set_path(self, json_path, root_folder, nx=False, xx=False, decode_keys=False):
+        """
+        Iterate over ``root_folder`` and set each JSON file to a value
+        under ``json_path`` with the file name as the key.
+
+        ``nx`` if set to True, set ``value`` only if it does not exist.
+        ``xx`` if set to True, set ``value`` only if it exists.
+        ``decode_keys`` If set to True, the keys of ``obj`` will be decoded
+        with utf-8.
+
+        """
+        set_files_result = {}
+        for root, dirs, files in os.walk(root_folder):
+            for file in files:
+                file_path = os.path.join(root, file)
+                try:
+                    file_name = file_path.rsplit(".")[0]
+                    self.set_file(
+                        file_name,
+                        json_path,
+                        file_path,
+                        nx=nx,
+                        xx=xx,
+                        decode_keys=decode_keys,
+                    )
+                    set_files_result[file_path] = True
+                except JSONDecodeError:
+                    set_files_result[file_path] = False
+
+        return set_files_result
+
     def strlen(self, name, path=None):
         """Return the length of the string JSON value under ``path`` at key
         ``name``.
diff --git a/redis/connection.py b/redis/connection.py
index 3fe8543c..a349a0f2 100755
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -513,6 +513,7 @@ class Connection:
         socket_keepalive_options=None,
         socket_type=0,
         retry_on_timeout=False,
+        retry_on_error=[],
         encoding="utf-8",
         encoding_errors="strict",
         decode_responses=False,
@@ -526,8 +527,10 @@ class Connection:
     ):
         """
         Initialize a new Connection.
-        To specify a retry policy, first set `retry_on_timeout` to `True`
-        then set `retry` to a valid `Retry` object
+        To specify a retry policy for specific errors, first set
+        `retry_on_error` to a list of the error/s to retry on, then set
+        `retry` to a valid `Retry` object.
+        To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
         self.pid = os.getpid()
         self.host = host
@@ -543,11 +546,17 @@ class Connection:
         self.socket_type = socket_type
         self.retry_on_timeout = retry_on_timeout
         if retry_on_timeout:
+            # Add TimeoutError to the errors list to retry on
+            retry_on_error.append(TimeoutError)
+        self.retry_on_error = retry_on_error
+        if retry_on_error:
             if retry is None:
                 self.retry = Retry(NoBackoff(), 1)
             else:
                 # deep-copy the Retry object as it is mutable
                 self.retry = copy.deepcopy(retry)
+            # Update the retry's supported errors with the specified errors
+            self.retry.update_supported_erros(retry_on_error)
         else:
             self.retry = Retry(NoBackoff(), 0)
         self.health_check_interval = health_check_interval
@@ -969,6 +978,7 @@ class UnixDomainSocketConnection(Connection):
         encoding_errors="strict",
         decode_responses=False,
         retry_on_timeout=False,
+        retry_on_error=[],
         parser_class=DefaultParser,
         socket_read_size=65536,
         health_check_interval=0,
@@ -978,8 +988,10 @@ class UnixDomainSocketConnection(Connection):
     ):
         """
         Initialize a new UnixDomainSocketConnection.
-        To specify a retry policy, first set `retry_on_timeout` to `True`
-        then set `retry` to a valid `Retry` object
+        To specify a retry policy for specific errors, first set
+        `retry_on_error` to a list of the error/s to retry on, then set
+        `retry` to a valid `Retry` object.
+        To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
         self.pid = os.getpid()
         self.path = path
@@ -990,11 +1002,17 @@ class UnixDomainSocketConnection(Connection):
         self.socket_timeout = socket_timeout
         self.retry_on_timeout = retry_on_timeout
         if retry_on_timeout:
+            # Add TimeoutError to the errors list to retry on
+            retry_on_error.append(TimeoutError)
+        self.retry_on_error = retry_on_error
+        if self.retry_on_error:
             if retry is None:
                 self.retry = Retry(NoBackoff(), 1)
             else:
                 # deep-copy the Retry object as it is mutable
                 self.retry = copy.deepcopy(retry)
+            # Update the retry's supported errors with the specified errors
+            self.retry.update_supported_erros(retry_on_error)
         else:
             self.retry = Retry(NoBackoff(), 0)
         self.health_check_interval = health_check_interval
@@ -1052,6 +1070,7 @@ URL_QUERY_ARGUMENT_PARSERS = {
     "socket_connect_timeout": float,
     "socket_keepalive": to_bool,
     "retry_on_timeout": to_bool,
+    "retry_on_error": list,
     "max_connections": int,
     "health_check_interval": int,
     "ssl_check_hostname": to_bool,
diff --git a/redis/retry.py b/redis/retry.py
index 75504c77..6147fbd9 100644
--- a/redis/retry.py
+++ b/redis/retry.py
@@ -19,6 +19,14 @@ class Retry:
         self._retries = retries
         self._supported_errors = supported_errors
 
+    def update_supported_erros(self, specified_errors: list):
+        """
+        Updates the supported errors with the specified error types
+        """
+        self._supported_errors = tuple(
+            set(self._supported_errors + tuple(specified_errors))
+        )
+
     def call_with_retry(self, do, fail):
         """
         Execute an operation that might fail and returns its result, or
diff --git a/tox.ini b/tox.ini
index b574d17b..bceab0bf 100644
--- a/tox.ini
+++ b/tox.ini
@@ -73,12 +73,6 @@ ports =
     36379:6379/tcp
 healtcheck_cmd = python -c "import socket;print(True) if 0 == socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect_ex(('127.0.0.1',36379)) else False"
 
-[docker:lots-of-pythons]
-name = lots-of-pythons
-image = redisfab/lots-of-pythons:latest
-volumes =
-    bind:rw:{toxinidir}:/data
-
 [docker:redis_cluster]
 name = redis_cluster
 image = redisfab/redis-py-cluster:6.2.6-buster


Test patch:
diff --git a/tests/test_cluster.py b/tests/test_cluster.py
index e1c8c347..496ed981 100644
--- a/tests/test_cluster.py
+++ b/tests/test_cluster.py
@@ -22,6 +22,7 @@ from redis.crc import key_slot
 from redis.exceptions import (
     AskError,
     ClusterDownError,
+    ConnectionError,
     DataError,
     MovedError,
     NoPermissionError,
@@ -555,46 +556,24 @@ class TestRedisClusterObj:
         for node in r.get_primaries():
             assert node in nodes
 
-    def test_cluster_down_overreaches_retry_attempts(self):
+    @pytest.mark.parametrize("error", RedisCluster.ERRORS_ALLOW_RETRY)
+    def test_cluster_down_overreaches_retry_attempts(self, error):
         """
-        When ClusterDownError is thrown, test that we retry executing the
-        command as many times as configured in cluster_error_retry_attempts
+        When error that allows retry is thrown, test that we retry executing
+        the command as many times as configured in cluster_error_retry_attempts
         and then raise the exception
         """
         with patch.object(RedisCluster, "_execute_command") as execute_command:
 
-            def raise_cluster_down_error(target_node, *args, **kwargs):
+            def raise_error(target_node, *args, **kwargs):
                 execute_command.failed_calls += 1
-                raise ClusterDownError(
-                    "CLUSTERDOWN The cluster is down. Use CLUSTER INFO for "
-                    "more information"
-                )
+                raise error("mocked error")
 
-            execute_command.side_effect = raise_cluster_down_error
+            execute_command.side_effect = raise_error
 
             rc = get_mocked_redis_client(host=default_host, port=default_port)
 
-            with pytest.raises(ClusterDownError):
-                rc.get("bar")
-                assert execute_command.failed_calls == rc.cluster_error_retry_attempts
-
-    def test_connection_error_overreaches_retry_attempts(self):
-        """
-        When ConnectionError is thrown, test that we retry executing the
-        command as many times as configured in cluster_error_retry_attempts
-        and then raise the exception
-        """
-        with patch.object(RedisCluster, "_execute_command") as execute_command:
-
-            def raise_conn_error(target_node, *args, **kwargs):
-                execute_command.failed_calls += 1
-                raise ConnectionError()
-
-            execute_command.side_effect = raise_conn_error
-
-            rc = get_mocked_redis_client(host=default_host, port=default_port)
-
-            with pytest.raises(ConnectionError):
+            with pytest.raises(error):
                 rc.get("bar")
                 assert execute_command.failed_calls == rc.cluster_error_retry_attempts
 
diff --git a/tests/test_commands.py b/tests/test_commands.py
index b8dc69f9..09121617 100644
--- a/tests/test_commands.py
+++ b/tests/test_commands.py
@@ -396,6 +396,28 @@ class TestRedisCommands:
         assert len(res) > 2
         assert "prefixes" in res
 
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("6.0.0")
+    def test_client_tracking(self, r, r2):
+
+        # simple case
+        assert r.client_tracking_on()
+        assert r.client_tracking_off()
+
+        # id based
+        client_id = r.client_id()
+        assert r.client_tracking_on(client_id)
+        assert r.client_tracking_off(client_id)
+
+        # id exists
+        client_id = r2.client_id()
+        assert r.client_tracking_on(client_id)
+        assert r2.client_tracking_off(client_id)
+
+        # now with some prefixes
+        with pytest.raises(exceptions.DataError):
+            assert r.client_tracking_on(prefix=["foo", "bar", "blee"])
+
     @pytest.mark.onlynoncluster
     @skip_if_server_version_lt("5.0.0")
     def test_client_unblock(self, r):
@@ -557,6 +579,13 @@ class TestRedisCommands:
         with pytest.raises(exceptions.RedisError):
             r.client_pause(timeout="not an integer")
 
+    @skip_if_server_version_lt("6.2.0")
+    def test_client_pause_all(self, r, r2):
+        assert r.client_pause(1, all=False)
+        assert r2.set("foo", "bar")
+        assert r2.get("foo") == b"bar"
+        assert r.get("foo") == b"bar"
+
     @pytest.mark.onlynoncluster
     @skip_if_server_version_lt("6.2.0")
     @skip_if_redis_enterprise()
@@ -640,6 +669,11 @@ class TestRedisCommands:
         lolwut = r.lolwut(5, 6, 7, 8).decode("utf-8")
         assert "Redis ver." in lolwut
 
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("6.2.0")
+    def test_reset(self, r):
+        assert r.reset() == "RESET"
+
     def test_object(self, r):
         r["a"] = "foo"
         assert isinstance(r.object("refcount", "a"), int)
diff --git a/tests/test_json.py b/tests/test_json.py
index a99547dd..6980e673 100644
--- a/tests/test_json.py
+++ b/tests/test_json.py
@@ -1392,3 +1392,41 @@ def test_custom_decoder(client):
     assert client.exists("foo") == 0
     assert not isinstance(cj.__encoder__, json.JSONEncoder)
     assert not isinstance(cj.__decoder__, json.JSONDecoder)
+
+
+@pytest.mark.redismod
+def test_set_file(client):
+    import json
+    import tempfile
+
+    obj = {"hello": "world"}
+    jsonfile = tempfile.NamedTemporaryFile(suffix=".json")
+    with open(jsonfile.name, "w+") as fp:
+        fp.write(json.dumps(obj))
+
+    nojsonfile = tempfile.NamedTemporaryFile()
+    nojsonfile.write(b"Hello World")
+
+    assert client.json().set_file("test", Path.rootPath(), jsonfile.name)
+    assert client.json().get("test") == obj
+    with pytest.raises(json.JSONDecodeError):
+        client.json().set_file("test2", Path.rootPath(), nojsonfile.name)
+
+
+@pytest.mark.redismod
+def test_set_path(client):
+    import json
+    import tempfile
+
+    root = tempfile.mkdtemp()
+    sub = tempfile.mkdtemp(dir=root)
+    jsonfile = tempfile.mktemp(suffix=".json", dir=sub)
+    nojsonfile = tempfile.mktemp(dir=root)
+
+    with open(jsonfile, "w+") as fp:
+        fp.write(json.dumps({"hello": "world"}))
+    open(nojsonfile, "a+").write("hello")
+
+    result = {jsonfile: True, nojsonfile: False}
+    assert client.json().set_path(Path.rootPath(), root) == result
+    assert client.json().get(jsonfile.rsplit(".")[0]) == {"hello": "world"}
diff --git a/tests/test_pubsub.py b/tests/test_pubsub.py
index 20ae0a05..23af4615 100644
--- a/tests/test_pubsub.py
+++ b/tests/test_pubsub.py
@@ -2,6 +2,7 @@ import platform
 import threading
 import time
 from unittest import mock
+from unittest.mock import patch
 
 import pytest
 
@@ -348,15 +349,6 @@ class TestPubSubMessages:
             "pmessage", channel, "test message", pattern=pattern
         )
 
-    def test_get_message_without_subscribe(self, r):
-        p = r.pubsub()
-        with pytest.raises(RuntimeError) as info:
-            p.get_message()
-        expect = (
-            "connection not set: " "did you forget to call subscribe() or psubscribe()?"
-        )
-        assert expect in info.exconly()
-
 
 class TestPubSubAutoDecoding:
     "These tests only validate that we get unicode values back"
@@ -549,6 +541,39 @@ class TestPubSubTimeouts:
         assert wait_for_message(p) == make_message("subscribe", "foo", 1)
         assert p.get_message(timeout=0.01) is None
 
+    def test_get_message_not_subscribed_return_none(self, r):
+        p = r.pubsub()
+        assert p.subscribed is False
+        assert p.get_message() is None
+        assert p.get_message(timeout=0.1) is None
+        with patch.object(threading.Event, "wait") as mock:
+            mock.return_value = False
+            assert p.get_message(timeout=0.01) is None
+            assert mock.called
+
+    def test_get_message_subscribe_during_waiting(self, r):
+        p = r.pubsub()
+
+        def poll(ps, expected_res):
+            assert ps.get_message() is None
+            message = ps.get_message(timeout=1)
+            assert message == expected_res
+
+        subscribe_response = make_message("subscribe", "foo", 1)
+        poller = threading.Thread(target=poll, args=(p, subscribe_response))
+        poller.start()
+        time.sleep(0.2)
+        p.subscribe("foo")
+        poller.join()
+
+    def test_get_message_wait_for_subscription_not_being_called(self, r):
+        p = r.pubsub()
+        p.subscribe("foo")
+        with patch.object(threading.Event, "wait") as mock:
+            assert p.subscribed is True
+            assert wait_for_message(p) == make_message("subscribe", "foo", 1)
+            assert mock.called is False
+
 
 class TestPubSubWorkerThread:
     @pytest.mark.skipif(
diff --git a/tests/test_retry.py b/tests/test_retry.py
index c4650bc6..00947872 100644
--- a/tests/test_retry.py
+++ b/tests/test_retry.py
@@ -1,10 +1,20 @@
+from unittest.mock import patch
+
 import pytest
 
 from redis.backoff import NoBackoff
+from redis.client import Redis
 from redis.connection import Connection, UnixDomainSocketConnection
-from redis.exceptions import ConnectionError
+from redis.exceptions import (
+    BusyLoadingError,
+    ConnectionError,
+    ReadOnlyError,
+    TimeoutError,
+)
 from redis.retry import Retry
 
+from .conftest import _get_client
+
 
 class BackoffMock:
     def __init__(self):
@@ -39,6 +49,37 @@ class TestConnectionConstructorWithRetry:
         assert isinstance(c.retry, Retry)
         assert c.retry._retries == retries
 
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error(self, Class):
+        c = Class(retry_on_error=[ReadOnlyError])
+        assert c.retry_on_error == [ReadOnlyError]
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == 1
+
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error_empty_value(self, Class):
+        c = Class(retry_on_error=[])
+        assert c.retry_on_error == []
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == 0
+
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error_and_timeout(self, Class):
+        c = Class(
+            retry_on_error=[ReadOnlyError, BusyLoadingError], retry_on_timeout=True
+        )
+        assert c.retry_on_error == [ReadOnlyError, BusyLoadingError, TimeoutError]
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == 1
+
+    @pytest.mark.parametrize("retries", range(10))
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error_retry(self, Class, retries):
+        c = Class(retry_on_error=[ReadOnlyError], retry=Retry(NoBackoff(), retries))
+        assert c.retry_on_error == [ReadOnlyError]
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == retries
+
 
 class TestRetry:
     "Test that Retry calls backoff and retries the expected number of times"
@@ -65,3 +106,85 @@ class TestRetry:
         assert self.actual_failures == 1 + retries
         assert backoff.reset_calls == 1
         assert backoff.calls == retries
+
+
+@pytest.mark.onlynoncluster
+class TestRedisClientRetry:
+    "Test the standalone Redis client behavior with retries"
+
+    def test_client_retry_on_error_with_success(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+
+            def mock_parse_response(connection, *args, **options):
+                def ok_response(connection, *args, **options):
+                    return "MOCK_OK"
+
+                parse_response.side_effect = ok_response
+                raise ReadOnlyError()
+
+            parse_response.side_effect = mock_parse_response
+            r = _get_client(Redis, request, retry_on_error=[ReadOnlyError])
+            assert r.get("foo") == "MOCK_OK"
+            assert parse_response.call_count == 2
+
+    def test_client_retry_on_error_raise(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = BusyLoadingError()
+            retries = 3
+            r = _get_client(
+                Redis,
+                request,
+                retry_on_error=[ReadOnlyError, BusyLoadingError],
+                retry=Retry(NoBackoff(), retries),
+            )
+            with pytest.raises(BusyLoadingError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == retries + 1
+
+    def test_client_retry_on_error_different_error_raised(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = TimeoutError()
+            retries = 3
+            r = _get_client(
+                Redis,
+                request,
+                retry_on_error=[ReadOnlyError],
+                retry=Retry(NoBackoff(), retries),
+            )
+            with pytest.raises(TimeoutError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == 1
+
+    def test_client_retry_on_error_and_timeout(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = TimeoutError()
+            retries = 3
+            r = _get_client(
+                Redis,
+                request,
+                retry_on_error=[ReadOnlyError],
+                retry_on_timeout=True,
+                retry=Retry(NoBackoff(), retries),
+            )
+            with pytest.raises(TimeoutError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == retries + 1
+
+    def test_client_retry_on_timeout(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = TimeoutError()
+            retries = 3
+            r = _get_client(
+                Redis, request, retry_on_timeout=True, retry=Retry(NoBackoff(), retries)
+            )
+            with pytest.raises(TimeoutError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == retries + 1