Fix 1063

kombu/pidbox.py

@@ -14,7 +14,7 @@
 from . import Exchange, Queue, Consumer, Producer
 from .clocks import LamportClock
 from .common import maybe_declare, oid_from
-from .exceptions import InconsistencyError
+from .exceptions import InconsistencyError, OperationalError
 from .five import range, string_t
 from .log import get_logger
 from .utils.functional import maybe_evaluate, reprcall
@@ -151,6 +151,14 @@ def reply(self, data, exchange, routing_key, ticket, **kwargs):
                                     serializer=self.mailbox.serializer)
 
 
+def is_no_route_error_for_reply_celery_pidbox(exc_str):
+    if "Cannot route message for exchange" in exc_str \
+        and "Table empty or key no longer exists" in exc_str \
+        and "reply.celery.pidbox" in exc_str:
+        return True
+    return False
+
+
 class Mailbox(object):
     """Process Mailbox."""
 
@@ -284,6 +292,14 @@ def _publish_reply(self, reply, exchange, routing_key, ticket,
                     }, retry=True,
                     **opts
                 )
+            except OperationalError as exc:
+                # Fixes https://github.com/celery/kombu/issues/1063
+
+                if not exc.args and not is_no_route_error_for_reply_celery_pidbox(exc.args[0]):
+                    raise
+
+                error('NO_ROUTE_ERROR caught: %r', exc, exc_info=1)
+
             except InconsistencyError:
                 # queue probably deleted and no one is expecting a reply.
                 pass

t/unit/test_pidbox.py

@@ -6,9 +6,11 @@
 
 from case import Mock, patch
 
+import kombu
 from kombu import Connection
 from kombu import pidbox
-from kombu.exceptions import ContentDisallowed, InconsistencyError
+from kombu.exceptions import ContentDisallowed, InconsistencyError, OperationalError
+from kombu.transport import redis
 from kombu.utils.uuid import uuid
 
 
@@ -58,6 +60,26 @@ def test_publish_reply_ignores_InconsistencyError(self):
             )
             producer.publish.assert_called()
 
+    def test_publish_reply_handles_redis_OperationalError_wth_no_route_error_msg(self):
+        mailbox = pidbox.Mailbox('reply.celery')(self.connection)
+        exchange = mailbox.reply_exchange.name
+        channel = self.connection.channel()
+        mailbox.reply_queue(channel).declare()
+        ticket = uuid()
+        # Using Channel._lookup as a proxy (in absence of a redis integration test) to mock the actual
+        # redis.get_table call which produces the OperationalError, since _lookup is just a level above
+        # redis.get_table on the stack trace.
+        with patch.object(kombu.transport.virtual.Channel, '_lookup') as simulate_redis_get_table_err:
+            # raise the actual redis error
+            simulate_redis_get_table_err.side_effect = OperationalError(
+                redis.NO_ROUTE_ERROR.format(exchange, mailbox.oid))
+            try:
+                mailbox._publish_reply({'foo': 'bar'}, exchange, mailbox.oid, ticket)
+            except OperationalError as exc:
+                if pidbox.is_no_route_error_for_reply_celery_pidbox(exc.args[0]):
+                    pytest.fail("NO_ROUTE_ERROR with specific message should have been caught")
+
+
     def test_reply__collect(self):
         mailbox = pidbox.Mailbox('test_reply__collect')(self.connection)
         exchange = mailbox.reply_exchange.name