Fix: Fanout exchange messages mixed across virtual databases in Redis…

… sentinel
celery · May 4, 2024 · 4d28194 · 4d28194
1 parent a56abf6
commit 4d28194
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 0 deletions.
diff --git a/kombu/transport/redis.py b/kombu/transport/redis.py
@@ -1433,6 +1433,8 @@ def _sentinel_managed_pool(self, asynchronous=False):
         ).connection_pool
 
     def _get_pool(self, asynchronous=False):
+        params = self._connparams(asynchronous=asynchronous)
+        self.keyprefix_fanout = self.keyprefix_fanout.format(db=params['db'])
         return self._sentinel_managed_pool(asynchronous)
 
 

diff --git a/t/unit/transport/test_redis.py b/t/unit/transport/test_redis.py
@@ -1634,6 +1634,18 @@ def test_method_called(self):
             connection.channel()
             p.assert_called()
 
+    def test_keyprefix_fanout(self):
+        from kombu.transport.redis import SentinelChannel
+        with patch.object(SentinelChannel, '_sentinel_managed_pool'):
+            connection = Connection(
+                'sentinel://localhost:65532/1',
+                transport_options={
+                    'master_name': 'not_important',
+                },
+            )
+            channel = connection.channel()
+            assert channel.keyprefix_fanout == '/1.'
+
     def test_getting_master_from_sentinel(self):
         with patch('redis.sentinel.Sentinel') as patched:
             connection = Connection(