Merge "Make ZeroMQ based RPC consumer threads more robust"

2 files changed, 22 insertions, 0 deletions

diff --git a/openstack/common/rpc/impl_zmq.py b/openstack/common/rpc/impl_zmq.py
index 33d5cb6..a5ddccf 100644
--- a/openstack/common/rpc/impl_zmq.py
+++ b/openstack/common/rpc/impl_zmq.py
@@ -383,6 +383,7 @@ class ZmqBaseReactor(ConsumerBase):
         LOG.info(_("In reactor registered"))
 
     def consume_in_thread(self):
+        @excutils.forever_retry_uncaught_exceptions
         def _consume(sock):
             LOG.info(_("Consuming socket"))
             while True:
diff --git a/tests/unit/rpc/test_zmq.py b/tests/unit/rpc/test_zmq.py
index 11d3fcd..888dbb0 100644
--- a/tests/unit/rpc/test_zmq.py
+++ b/tests/unit/rpc/test_zmq.py
@@ -112,6 +112,27 @@ class _RpcZmqBaseTestCase(common.BaseRpcTestCase):
             finally:
                 self.topic_nested = tmp_topic
 
+    def test_service_consume_thread_unexpected_exceptions(self):
+        def my_TopicConsumer_consume(myself, *args, **kwargs):
+            self.consume_calls += 1
+            # see if it can sustain three failures
+            if self.consume_calls < 3:
+                raise Exception('unexpected exception')
+            else:
+                self.orig_ZmqReactor_consume(myself, *args, **kwargs)
+
+        self.consume_calls = 0
+        self.orig_ZmqReactor_consume = impl_zmq.ZmqReactor.consume
+        self.stubs.Set(impl_zmq.ZmqReactor, 'consume',
+                       my_TopicConsumer_consume)
+
+        value = 42
+        result = self.rpc.call(FLAGS, self.context, self.topic,
+                               {"method": "echo",
+                                "args": {"value": value}})
+        self.assertEqual(value, result)
+        self.stubs.UnsetAll()
+
 
 class RpcZmqBaseTopicTestCase(_RpcZmqBaseTestCase):
     """Base topic RPC ZMQ test case.