Fix ConnectionPool to raise MaxConnectionsError instead of Connection…

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/cluster.py

@@ -37,6 +37,7 @@
     ConnectionError,
     CrossSlotTransactionError,
     DataError,
+    MaxConnectionsError,
     ExecAbortError,
     InvalidPipelineStack,
     MovedError,
@@ -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,88 @@
+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) as mock_get_conn:
+        # 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()