Ensure consumer recovers from broker connection restart/issues

[GonzaloMFB]

Issue #814 describes how a Kombu consumer fails to properly recover when the message broker experiences issues without explicitly calling consumer.consume(). This happens because, while the queues are still recorded, consumer.revive() will clear the active tags originally recorded in the consumer without setting new ones for the active queues.

A simple solution would be to modify revive to check whether the consumer had any active tags at the time, and call consume if it did. To prevent consumption from previously inactive queues at the time of the connection crash, I've created a new method _consume_previously_active so that revive will not silently activate them.

To avoid duplicating the basic_consume loop between consume and _consume_previously_active, I extracted it into a shared _consume_queues helper. Behavior should be unchanged.

Apart from unit tests, I've tested that case manually by modifying the consumer file provided in #814 to include a new queue that's never consumed from, then logged the active tags in revive at the start and end of the method. Upon RabbitMQ restart, the previously inactive queue remains inactive.

Closes #814

[Copilot]

Pull request overview

This PR fixes a recovery gap in kombu.Consumer where, after a broker reconnect, Consumer.revive() would clear _active_tags and leave the consumer no longer consuming unless user code explicitly called consumer.consume() again. The change re-establishes consumption automatically, but only for queues that were previously active, aligning behavior with the expectations described in Issue #814.

Changes:

• Track queues that were actively consumed before revive(), clear tags, then re-consume only those queues after channel/queue revival.
• Refactor the consumption loop into a shared _consume_queues() helper and add _consume_previously_active() to avoid activating merely-registered queues.
• Add unit tests covering resume vs non-resume behavior and ensuring inactive (never-consumed) queues remain inactive after recovery.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
kombu/messaging.py	Updates Consumer.revive() to resume consumption for previously active queues and refactors consume logic into reusable helpers.
t/unit/test_messaging.py	Adds tests validating revive() resumes consumption only when appropriate and preserves inactivity for never-consumed queues.

Suppressed comments (1)

kombu/messaging.py:702

• To make the revive() resume call able to preserve the effective no_ack value from the last successful (or attempted) consume, _consume_queues() should record the resolved no_ack it uses. Without this, revive() can't distinguish between default self.no_ack and a per-call override.

    def _consume_queues(self, queues, no_ack=None):
        if queues:
            no_ack = self.no_ack if no_ack is None else no_ack

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[auvipy]

thanks for handling this long standing issue! would you mind adding integration tests inspired from the examples or reproducer scripts in the issue? that would provide more way to verify the change

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 83.18%. Comparing base (78b528b) to head (6c646e8).
✅ All tests successful. No failed tests found.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2593      +/-   ##
==========================================
+ Coverage   83.17%   83.18%   +0.01%     
==========================================
  Files          79       79              
  Lines       10636    10643       +7     
  Branches     1243     1243              
==========================================
+ Hits         8846     8853       +7     
  Misses       1584     1584              
  Partials      206      206              

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

[GonzaloMFB]

Thanks again for the quick review!

• Added a new integration test under t/integration/test_py_amqp.py. I'm simulating the connection error/reset by killing the connection socket. The _drain method is the same used in the consumer script listed under Consumer not gracefully recovering from rabbitmq restarts/issues #814 to reproduce the bug.
• Working on the fix for the no_ack overriding bug.

[auvipy]

• Working on the fix for the no_ack overriding bug.

so this part is still remaining to fix? thanks for you effort so far

[GonzaloMFB]

• Working on the fix for the no_ack overriding bug.

so this part is still remaining to fix? thanks for you effort so far

It was; I have just pushed the fix.