diff --git a/kombu/transport/virtual/base.py b/kombu/transport/virtual/base.py index 4c3ede5140..221ae670f1 100644 --- a/kombu/transport/virtual/base.py +++ b/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 diff --git a/t/unit/transport/virtual/test_base.py b/t/unit/transport/virtual/test_base.py index 0c3dd1f08e..bf2493dca2 100644 --- a/t/unit/transport/virtual/test_base.py +++ b/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.ack(1) # simulate concurrent ack of the delivered 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_,