Skip to content

Commit 1091b37

Browse files
committed
[GROW-3134] release already acquired connections, when get_connections raised an exception (#3)
* [GROW-3134] release already acquired connections, when get_connections raised an exception * fix style (cherry picked from commit 94bb915)
1 parent 616cb31 commit 1091b37

File tree

2 files changed

+30
-0
lines changed

2 files changed

+30
-0
lines changed

redis/cluster.py

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2110,6 +2110,8 @@ def _send_cluster_commands(
21102110
try:
21112111
connection = get_connection(redis_node, c.args)
21122112
except (ConnectionError, TimeoutError) as e:
2113+
for n in nodes.values():
2114+
n.connection_pool.release(n.connection)
21132115
if self.retry and isinstance(e, self.retry._supported_errors):
21142116
backoff = self.retry._backoff.compute(attempts_count)
21152117
if backoff > 0:

tests/test_cluster.py

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88

99
import pytest
1010

11+
import redis.cluster
1112
from redis import Redis
1213
from redis.backoff import (
1314
ConstantBackoff,
@@ -2970,6 +2971,33 @@ def raise_ask_error():
29702971
assert ask_node.redis_connection.connection.read_response.called
29712972
assert res == ["MOCK_OK"]
29722973

2974+
@pytest.mark.parametrize("error", [ConnectionError, TimeoutError])
2975+
def test_return_previous_acquired_connections(self, r, error):
2976+
# in order to ensure that a pipeline will make use of connections
2977+
# from different nodes
2978+
assert r.keyslot("a") != r.keyslot("b")
2979+
2980+
orig_func = redis.cluster.get_connection
2981+
with patch("redis.cluster.get_connection") as get_connection:
2982+
2983+
def raise_error(target_node, *args, **kwargs):
2984+
if get_connection.call_count == 2:
2985+
raise error("mocked error")
2986+
else:
2987+
return orig_func(target_node, *args, **kwargs)
2988+
2989+
get_connection.side_effect = raise_error
2990+
2991+
r.pipeline().get("a").get("b").execute()
2992+
2993+
# there should have been two get_connections per execution and
2994+
# two executions due to exception raised in the first execution
2995+
assert get_connection.call_count == 4
2996+
for cluster_node in r.nodes_manager.nodes_cache.values():
2997+
connection_pool = cluster_node.redis_connection.connection_pool
2998+
num_of_conns = len(connection_pool._available_connections)
2999+
assert num_of_conns == connection_pool._created_connections
3000+
29733001
def test_empty_stack(self, r):
29743002
"""
29753003
If pipeline is executed with no commands it should

0 commit comments

Comments
 (0)