Ignore errors when attempting to restore messages if the message was acked

kombu/transport/virtual/base.py

@@ -260,14 +260,18 @@ def restore_unacked(self):
 
         while delivered:
             try:
-                _, message = pop_message()
+                key, message = pop_message()
             except KeyError:  # pragma: no cover
                 break
 
             try:
                 restore(message)
             except BaseException as exc:
-                errors.append((exc, message))
+                if key not in self._dirty:
+                    # Another thread may have acked the message after the earlier '_flush' call.
+                    # This may cause the restore attempt to fail (e.g. in SQS).
+                    # If restore fails, we only care about errors for messages that have not been 'acked'.
+                    errors.append((exc, message))
         delivered.clear()
         return errors
 

t/unit/transport/virtual/test_base.py

@@ -410,6 +410,22 @@ def test_restore_unacked_raises_BaseException(self):
         assert errors[0][1] == 1
         assert not q._delivered
 
+    def test_restore_unacked_ignores_raised_exceptions_when_acked(self):
+        q = self.channel.qos
+        q._flush = Mock()
+        q._delivered = {1: 1}
+
+        def mock_restore_raises_exceptions_due_to_acked_message(*args, **kwargs):
+            q._dirty = {1}  # acked dirty message
+            raise SystemExit(1)
+
+        q.channel._restore = Mock()
+        q.channel._restore.side_effect = mock_restore_raises_exceptions_due_to_acked_message
+
+        errors = q.restore_unacked()
+        assert not errors
+        assert not q._delivered
+
     @patch('kombu.transport.virtual.base.emergency_dump_state')
     @patch(PRINT_FQDN)
     def test_restore_unacked_once_when_unrestored(self, print_,