Fix ConnectionPool to raise MaxConnectionsError instead of Connection… (#3698)

* Fix ConnectionPool to raise MaxConnectionsError instead of ConnectionError

- Added MaxConnectionsError class as a subclass of ConnectionError
- Updated connection.py to raise the more specific error
- Updated cluster.py to handle this specific error type
- Added tests to verify the behavior

Fixes #3684

* Update async cluster to handle MaxConnectionsError separately

Add an explanatory comment to the exception handling for MaxConnectionsError in the
async cluster implementation, matching the synchronous version.
This ensures consistent behavior between sync and async implementations when
connection pool is exhausted, preventing unnecessary cluster reinitialization.
---------

Author: ngabhanenetskope

redis/__init__.py

@@ -20,6 +20,7 @@
     DataError,
     InvalidPipelineStack,
     InvalidResponse,
+    MaxConnectionsError,
     OutOfMemoryError,
     PubSubError,
     ReadOnlyError,
@@ -65,6 +66,7 @@ def int_or_str(value):
     "default_backoff",
     "InvalidPipelineStack",
     "InvalidResponse",
+    "MaxConnectionsError",
     "OutOfMemoryError",
     "PubSubError",
     "ReadOnlyError",

redis/asyncio/cluster.py

@@ -814,7 +814,13 @@ async def _execute_command(
                     moved = False
 
                 return await target_node.execute_command(*args, **kwargs)
-            except (BusyLoadingError, MaxConnectionsError):
+            except BusyLoadingError:
+                raise
+            except MaxConnectionsError:
+                # MaxConnectionsError indicates client-side resource exhaustion
+                # (too many connections in the pool), not a node failure.
+                # Don't treat this as a node failure - just re-raise the error
+                # without reinitializing the cluster.
                 raise
             except (ConnectionError, TimeoutError):
                 # Connection retries are being handled in the node's

redis/cluster.py

@@ -39,6 +39,7 @@
     DataError,
     ExecAbortError,
     InvalidPipelineStack,
+    MaxConnectionsError,
     MovedError,
     RedisClusterException,
     RedisError,
@@ -1235,6 +1236,12 @@ def _execute_command(self, target_node, *args, **kwargs):
                 return response
             except AuthenticationError:
                 raise
+            except MaxConnectionsError:
+                # MaxConnectionsError indicates client-side resource exhaustion
+                # (too many connections in the pool), not a node failure.
+                # Don't treat this as a node failure - just re-raise the error
+                # without reinitializing the cluster.
+                raise
             except (ConnectionError, TimeoutError) as e:
                 # ConnectionError can also be raised if we couldn't get a
                 # connection from the pool before timing out, so check that

redis/connection.py

@@ -31,6 +31,7 @@
     ChildDeadlockedError,
     ConnectionError,
     DataError,
+    MaxConnectionsError,
     RedisError,
     ResponseError,
     TimeoutError,
@@ -1556,7 +1557,7 @@ def get_encoder(self) -> Encoder:
     def make_connection(self) -> "ConnectionInterface":
         "Create a new connection"
         if self._created_connections >= self.max_connections:
-            raise ConnectionError("Too many connections")
+            raise MaxConnectionsError("Too many connections")
         self._created_connections += 1
 
         if self.cache is not None:

redis/exceptions.py

@@ -220,7 +220,13 @@ class SlotNotCoveredError(RedisClusterException):
     pass
 
 
-class MaxConnectionsError(ConnectionError): ...
+class MaxConnectionsError(ConnectionError):
+    """
+    Raised when a connection pool has reached its max_connections limit.
+    This indicates pool exhaustion rather than an actual connection failure.
+    """
+
+    pass
 
 
 class CrossSlotTransactionError(RedisClusterException):

tests/test_connection_pool.py

@@ -76,11 +76,13 @@ def test_multiple_connections(self, master_host):
         assert c1 != c2
 
     def test_max_connections(self, master_host):
-        connection_kwargs = {"host": master_host[0], "port": master_host[1]}
-        pool = self.get_pool(max_connections=2, connection_kwargs=connection_kwargs)
+        # Use DummyConnection to avoid actual connection to Redis
+        # This prevents authentication issues and makes the test more reliable
+        # while still properly testing the MaxConnectionsError behavior
+        pool = self.get_pool(max_connections=2, connection_class=DummyConnection)
         pool.get_connection()
         pool.get_connection()
-        with pytest.raises(redis.ConnectionError):
+        with pytest.raises(redis.MaxConnectionsError):
             pool.get_connection()
 
     def test_reuse_previously_released_connection(self, master_host):

tests/test_max_connections_error.py

@@ -0,0 +1,112 @@
+import pytest
+import redis
+from unittest import mock
+from redis.connection import ConnectionInterface
+
+
+class DummyConnection(ConnectionInterface):
+    """A dummy connection class for testing that doesn't actually connect to Redis"""
+
+    def __init__(self, *args, **kwargs):
+        self.connected = False
+
+    def connect(self):
+        self.connected = True
+
+    def disconnect(self):
+        self.connected = False
+
+    def register_connect_callback(self, callback):
+        pass
+
+    def deregister_connect_callback(self, callback):
+        pass
+
+    def set_parser(self, parser_class):
+        pass
+
+    def get_protocol(self):
+        return 2
+
+    def on_connect(self):
+        pass
+
+    def check_health(self):
+        return True
+
+    def send_packed_command(self, command, check_health=True):
+        pass
+
+    def send_command(self, *args, **kwargs):
+        pass
+
+    def can_read(self, timeout=0):
+        return False
+
+    def read_response(self, disable_decoding=False, **kwargs):
+        return "PONG"
+
+
+@pytest.mark.onlynoncluster
+def test_max_connections_error_inheritance():
+    """Test that MaxConnectionsError is a subclass of ConnectionError"""
+    assert issubclass(redis.MaxConnectionsError, redis.ConnectionError)
+
+
+@pytest.mark.onlynoncluster
+def test_connection_pool_raises_max_connections_error():
+    """Test that ConnectionPool raises MaxConnectionsError and not ConnectionError"""
+    # Use a dummy connection class that doesn't try to connect to a real Redis server
+    pool = redis.ConnectionPool(max_connections=1, connection_class=DummyConnection)
+    pool.get_connection()
+
+    with pytest.raises(redis.MaxConnectionsError):
+        pool.get_connection()
+
+
+@pytest.mark.skipif(
+    not hasattr(redis, "RedisCluster"), reason="RedisCluster not available"
+)
+def test_cluster_handles_max_connections_error():
+    """
+    Test that RedisCluster doesn't reinitialize when MaxConnectionsError is raised
+    """
+    # Create a more complete mock cluster
+    cluster = mock.MagicMock(spec=redis.RedisCluster)
+    cluster.cluster_response_callbacks = {}
+    cluster.RedisClusterRequestTTL = 3  # Set the TTL to avoid infinite loops
+    cluster.nodes_manager = mock.MagicMock()
+    node = mock.MagicMock()
+
+    # Mock get_redis_connection to return a mock Redis client
+    redis_conn = mock.MagicMock()
+    cluster.get_redis_connection.return_value = redis_conn
+
+    # Setup get_connection to be called and return a connection that will raise
+    connection = mock.MagicMock()
+
+    # Patch the get_connection function in the cluster module
+    with mock.patch("redis.cluster.get_connection", return_value=connection):
+        # Test MaxConnectionsError
+        connection.send_command.side_effect = redis.MaxConnectionsError(
+            "Too many connections"
+        )
+
+        # Call the method and check that the exception is raised
+        with pytest.raises(redis.MaxConnectionsError):
+            redis.RedisCluster._execute_command(cluster, node, "GET", "key")
+
+        # Verify nodes_manager.initialize was NOT called
+        cluster.nodes_manager.initialize.assert_not_called()
+
+        # Reset the mock for the next test
+        cluster.nodes_manager.initialize.reset_mock()
+
+        # Now test with regular ConnectionError to ensure it DOES reinitialize
+        connection.send_command.side_effect = redis.ConnectionError("Connection lost")
+
+        with pytest.raises(redis.ConnectionError):
+            redis.RedisCluster._execute_command(cluster, node, "GET", "key")
+
+        # Verify nodes_manager.initialize WAS called
+        cluster.nodes_manager.initialize.assert_called_once()