Repository: celery/kombu

Issue:
ConsumerMixin no longer send heartbeat if `timeout=None`
:wave: 

First I want to notice that the change in https://github.com/celery/kombu/issues/1997 is backwards incompatible, and still was release on a **minor version** with **no breaking change warning or mention on the Changelog**. This change broke all our consumers.

Second I'd like to question the logic behind the change. Why should `timeout=None` should cause heartbeats not to be sent? `timeout=None` means that we don't want to have any timeout while waiting for new messages. In this case we STILL want to send hearbeats to the broker, otherwise our connections will get killed.

Would you care to explain why it make sense that `timeout=None` should make the client stop sending hearbeats? What should I do if I don't want a timeout (e.g. I want my consumer to run forever, regardless of how long it takes for new messages to arrive) and I still want heartbeats to be sent so my connection is not broken?

Am I not properly understanding the API here?

Thanks!

Hints from issue comments:
@Nusnus: @auvipy @smart-programmer @FrankK-1234 [#2016](https://github.com/celery/kombu/pull/2016)@FrankK-1234: Hello @vitorquintanilha,

After a closer look at the changes (from @smart-programmer) and the stated implied change . I have to agree it is not backwards compatible, perhaps the misunderstanding came from the name heartbeat_check. Actually I think there would be a problem if the parameter safety_interval is changed as that dictates how fast the heartbeat_check is called (it should be called about every second). So the question arise if this is the right place to implement that call.

My advice is to revert the change including the tests I had added, sorry for just assuming a right implementation. @auvipy could you make the revert if you agree? I think there might be another way to reach the original problem and that is to only call heartbeat_check when conn.heartbeat is set, but that is not up to me to decide.

As a workaround one could call the method with a high timeout (e.g. 1000000000) value which would still trigger the heartbeat check but never (well 31 years without a message) trigger the time-out raise.

With kind regards,

Frank



Solution patch:
diff --git a/kombu/mixins.py b/kombu/mixins.py
index 92acee980..14c1c1b9f 100644
--- a/kombu/mixins.py
+++ b/kombu/mixins.py
@@ -195,11 +195,10 @@ def consume(self, limit=None, timeout=None, safety_interval=1, **kwargs):
                 try:
                     conn.drain_events(timeout=safety_interval)
                 except socket.timeout:
-                    if timeout:
-                        conn.heartbeat_check()
-                        elapsed += safety_interval
-                        if elapsed >= timeout:
-                            raise
+                    conn.heartbeat_check()
+                    elapsed += safety_interval
+                    if timeout and 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 a33412eb9..3ef4b9584 100644
--- a/t/unit/test_mixins.py
+++ b/t/unit/test_mixins.py
@@ -91,7 +91,7 @@ def se(*args, **kwargs):
             next(it)
         c.connection.heartbeat_check.assert_called()
 
-    def test_consume_drain_no_heartbeat_check(self):
+    def test_consume_drain_heartbeat_check_no_timeout(self):
         c, Acons, Bcons = self._context()
         c.should_stop = False
         it = c.consume(no_ack=True, timeout=None)
@@ -102,13 +102,13 @@ def se(*args, **kwargs):
         c.connection.drain_events.side_effect = se
         with pytest.raises(StopIteration):
             next(it)
-        c.connection.heartbeat_check.assert_not_called()
+        c.connection.heartbeat_check.assert_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()
+        c.connection.heartbeat_check.assert_called()
 
     def test_Consumer_context(self):
         c, Acons, Bcons = self._context()