Repository: celery/kombu

Issue:
ConsumerMixin consume method should not heartbeat if timeout None
<!--
Please fill this template entirely and do not erase parts of it.
We reserve the right to close without a response
enhancement requests which are incomplete.
-->
# Checklist
<!--
To check an item on the list replace [ ] with [x].
-->

- [x] I have checked the [issues list](https://github.com/celery/celery/issues?q=is%3Aissue+label%3A%22Issue+Type%3A+Enhancement%22+-label%3A%22Category%3A+Documentation%22)
  for similar or identical enhancement to an existing feature.
- [x] I have checked the [pull requests list](https://github.com/celery/celery/pulls?q=is%3Apr+label%3A%22Issue+Type%3A+Enhancement%22+-label%3A%22Category%3A+Documentation%22)
  for existing proposed enhancements.
- [x] I have checked the [commit log](https://github.com/celery/celery/commits/main)
  to find out if the same enhancement was already implemented in the
  main branch.
- [x] I have included all related issues and possible duplicate issues in this issue
      (If there are none, check this box anyway).

## Related Issues and Possible Duplicates
<!--
Please make sure to search and mention any related issues
or possible duplicates to this issue as requested by the checklist above.

This may or may not include issues in other repositories that the Celery project
maintains or other repositories that are dependencies of Celery.

If you don't know how to mention issues, please refer to Github's documentation
on the subject: https://help.github.com/en/articles/autolinked-references-and-urls#issues-and-pull-requests
-->

#### Related Issues

- None

#### Possible Duplicates

- None

# Brief Summary

the capture method in celery events Receiver. logs broker heartbeats on socket exceptions. this is a helpful feature sometimes but there's no way to disable it as it mostly clutters the terminal with unhelpful messages most of the time which prevents the user from seeing actual task event logs.
![image](https://github.com/celery/celery/assets/36680382/bf6fdf37-9ca6-4aa4-a6f7-2ca80b82d574)
 

# Design

## Architectural Considerations
<!--
If more components other than Celery are involved,
describe them here and the effect it would have on Celery.
-->
None

## Proposed Behavior

If the timeout argument is None that should disable broker heartbeat_check
![image](https://github.com/celery/celery/assets/36680382/8d633b1b-89d6-414b-9be8-11dad7fafee1)


## Proposed UI/UX
<!--
Please provide your ideas for the API, CLI options,
configuration key names etc. that will be adjusted for this enhancement.
-->

## Diagrams
<!--
Please include any diagrams that might be relevant
to the implementation of this enhancement such as:
* Class Diagrams
* Sequence Diagrams
* Activity Diagrams
You can drag and drop images into the text box to attach them to this issue.
-->
N/A

## Alternatives
<!--
If you have considered any alternative implementations
describe them in detail below.
-->
None


Discussion:

Patch:
diff --git a/kombu/mixins.py b/kombu/mixins.py
index 14c1c1b9..92acee98 100644
--- a/kombu/mixins.py
+++ b/kombu/mixins.py
@@ -195,10 +195,11 @@ class ConsumerMixin:
                 try:
                     conn.drain_events(timeout=safety_interval)
                 except socket.timeout:
-                    conn.heartbeat_check()
-                    elapsed += safety_interval
-                    if timeout and elapsed >= timeout:
-                        raise
+                    if timeout:
+                        conn.heartbeat_check()
+                        elapsed += safety_interval
+                        if elapsed >= timeout:
+                            raise
                 except OSError:
                     if not self.should_stop:
                         raise


Test patch:
diff --git a/t/unit/test_mixins.py b/t/unit/test_mixins.py
index 7a96091e..a33412eb 100644
--- a/t/unit/test_mixins.py
+++ b/t/unit/test_mixins.py
@@ -89,6 +89,26 @@ class test_ConsumerMixin:
         c.connection.drain_events.side_effect = se
         with pytest.raises(socket.error):
             next(it)
+        c.connection.heartbeat_check.assert_called()
+
+    def test_consume_drain_no_heartbeat_check(self):
+        c, Acons, Bcons = self._context()
+        c.should_stop = False
+        it = c.consume(no_ack=True, timeout=None)
+
+        def se(*args, **kwargs):
+            c.should_stop = True
+            raise socket.timeout()
+        c.connection.drain_events.side_effect = se
+        with pytest.raises(StopIteration):
+            next(it)
+        c.connection.heartbeat_check.assert_not_called()
+
+        it = c.consume(no_ack=True, timeout=0)
+        c.connection.drain_events.side_effect = se
+        with pytest.raises(StopIteration):
+            next(it)
+        c.connection.heartbeat_check.assert_not_called()
 
     def test_Consumer_context(self):
         c, Acons, Bcons = self._context()