diff options
author | Matus Valo <matusvalo@users.noreply.github.com> | 2021-10-05 13:09:30 +0200 |
---|---|---|
committer | GitHub <noreply@github.com> | 2021-10-05 14:09:30 +0300 |
commit | 96ca00f05f264947e9256225a83a35ba3297cdfe (patch) | |
tree | b5b9be5755dba010525ee3c798b725f5cafe55ef | |
parent | 5ef5e22638035a6c412e949a2fbc5d44b7b088b2 (diff) | |
download | kombu-96ca00f05f264947e9256225a83a35ba3297cdfe.tar.gz |
Prevent caching of oid in pidbox (#1394)
oid is not cached anymore due race conditions of oid in celery. Caching
oid is causing following exception:
OperationalError("
Cannot route message for exchange 'reply.celery.pidbox': Table empty or key no longer exists.
Probably the key ('_kombu.binding.reply.celery.pidbox') has been removed from the Redis database.
",)
This exception seems to be occuring when multiple celery workers
contains same oid.
-rw-r--r-- | kombu/pidbox.py | 10 |
1 files changed, 2 insertions, 8 deletions
diff --git a/kombu/pidbox.py b/kombu/pidbox.py index da10d1ae..7649736a 100644 --- a/kombu/pidbox.py +++ b/kombu/pidbox.py @@ -6,7 +6,6 @@ from collections import defaultdict, deque from contextlib import contextmanager from copy import copy from itertools import count -from threading import local from time import time from . import Consumer, Exchange, Producer, Queue @@ -187,7 +186,6 @@ class Mailbox: self.clock = LamportClock() if clock is None else clock self.exchange = self._get_exchange(self.namespace, self.type) self.reply_exchange = self._get_reply_exchange(self.namespace) - self._tls = local() self.unclaimed = defaultdict(deque) self.accept = self.accept if accept is None else accept self.serializer = self.serializer if serializer is None else serializer @@ -404,13 +402,9 @@ class Mailbox: durable=False, delivery_mode='transient') - @cached_property + @property def oid(self): - try: - return self._tls.OID - except AttributeError: - oid = self._tls.OID = oid_from(self) - return oid + return oid_from(self) @cached_property def producer_pool(self): |