Make late acknowledge

Author: andrvb

oslo_messaging/_drivers/amqpdriver.py

@@ -207,20 +207,11 @@ def heartbeat(self):
                 raise MessageDeliveryFailure(
                     "Heartbeat send failed. Missing exchange")
 
-    # NOTE(sileht): Those have already be ack in RpcListener IO thread
-    # We keep them as noop until all drivers do the same
-    def acknowledge(self):
-        pass
-
-    def requeue(self):
-        pass
-
-
-class NotificationAMQPIncomingMessage(AMQPIncomingMessage):
     def acknowledge(self):
         def _do_ack():
             try:
                 self.message.acknowledge()
+                LOG.info("Received message msg_id: %s has been acknowledged", self.msg_id)
             except Exception as exc:
                 # NOTE(kgiusti): this failure is likely due to a loss of the
                 # connection to the broker.  Not much we can do in this case,
@@ -229,6 +220,7 @@ def _do_ack():
                 # (unacked msg is returned to the queue by the broker), but the
                 # driver tries to catch that using the msg_id_cache.
                 LOG.warning("Failed to acknowledge received message: %s", exc)
+
         self._message_operations_handler.do(_do_ack)
         self.listener.msg_id_cache.add(self.unique_id)
 
@@ -242,11 +234,17 @@ def requeue(self):
         def _do_requeue():
             try:
                 self.message.requeue()
+                LOG.info("Received message msg_id: %s has been requeued", self.msg_id)
             except Exception as exc:
                 LOG.warning("Failed to requeue received message: %s", exc)
+
         self._message_operations_handler.do(_do_requeue)
 
 
+class NotificationAMQPIncomingMessage(AMQPIncomingMessage):
+    pass
+
+
 class ObsoleteReplyQueuesCache(object):
     """Cache of reply queue id that doesn't exist anymore.
 
@@ -399,31 +397,10 @@ class RpcAMQPListener(AMQPListener):
     use_cache = True
 
     def __call__(self, message):
-        # NOTE(kgiusti): In the original RPC implementation the RPC server
-        # would acknowledge the request THEN process it.  The goal of this was
-        # to prevent duplication if the ack failed.  Should the ack fail the
-        # request would be discarded since the broker would not remove the
-        # request from the queue since no ack was received.  That would lead to
-        # the request being redelivered at some point. However this approach
-        # meant that the ack was issued from the dispatch thread, not the
-        # consumer thread, which is bad since kombu is not thread safe.  So a
-        # change was made to schedule the ack to be sent on the consumer thread
-        # - breaking the ability to catch ack errors before dispatching the
-        # request.  To fix this we do the actual ack here in the consumer
-        # callback and avoid the upcall if the ack fails.  See
-        # https://bugs.launchpad.net/oslo.messaging/+bug/1695746
-        # for all the gory details...
-        try:
-            message.acknowledge()
-        except Exception as exc:
-            LOG.warning("Discarding RPC request due to failed acknowledge: %s",
-                        exc)
-        else:
-            # NOTE(kgiusti): be aware that even if the acknowledge call
-            # succeeds there is no guarantee the broker actually gets the ACK
-            # since acknowledge() simply writes the ACK to the socket (there is
-            # no ACK confirmation coming back from the broker)
-            super(RpcAMQPListener, self).__call__(message)
+        # NOTE(andrvb): The acknowledgment of the request is deferred until
+        # after the request has been processed. This provides an at-least-once
+        # delivery semantic. See RPCServer._process_incoming.
+        super(RpcAMQPListener, self).__call__(message)
 
 
 class NotificationAMQPListener(AMQPListener):

oslo_messaging/rpc/server.py

@@ -152,10 +152,6 @@ def _create_listener(self):
     def _process_incoming(self, incoming):
         message = incoming[0]
 
-        # TODO(sileht): We should remove that at some point and do
-        # this directly in the driver
-        message.acknowledge()
-
         failure = None
         try:
             res = self.dispatcher.dispatch(message)
@@ -194,6 +190,8 @@ def _process_incoming(self, incoming):
             # exc_info.
             del failure
 
+        message.acknowledge()
+
 
 def get_rpc_server(transport, target, endpoints,
                    executor=None, serializer=None, access_policy=None):