Repository: dpkp/kafka-python (python)

Issue:
KafkaProducer hangs in busy loop after delivery timeout
After producing one message and after the delivery timeout expires, the producer thread seems to be hanging in a busy loop. This requires unnecessary CPU cycles. Did I something wrong in the configuration?

```python
import logging
import sys
import time
from datetime import timedelta

import kafka  # kafka-python 2.2.4

logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
logging.getLogger('kafka').setLevel(logging.DEBUG)


def main():
    delivery_timeout = timedelta(seconds=2)

    producer = kafka.KafkaProducer(
        bootstrap_servers="localhost:9092",
        request_timeout_ms=int(delivery_timeout.total_seconds() * 1000 / 2),
        delivery_timeout_ms=int(delivery_timeout.total_seconds() * 1000),
    )

    # at least one message must be produced to trigger the issue
    producer.send("demo",  b"message")

    # simulate time without producing messages
    # after delivery_timeout_ms the poll() hangs in a busy loop
    time.sleep(delivery_timeout.total_seconds() + 1)


if __name__ == '__main__':
    main()
```

[kafka.log](https://github.com/user-attachments/files/20101919/kafka.log)

Pull request description:
Fix producer busy loop with no pending batches
Fix #2615

Patch:
diff --git a/kafka/producer/record_accumulator.py b/kafka/producer/record_accumulator.py
index 1c250ee..77d48d8 100644
--- a/kafka/producer/record_accumulator.py
+++ b/kafka/producer/record_accumulator.py
@@ -328,6 +328,9 @@ class RecordAccumulator(object):
         finally:
             self._appends_in_progress.decrement()
 
+    def reset_next_batch_expiry_time(self):
+        self._next_batch_expiry_time_ms = float('inf')
+
     def maybe_update_next_batch_expiry_time(self, batch):
         self._next_batch_expiry_time_ms = min(self._next_batch_expiry_time_ms, batch.created * 1000 + self.delivery_timeout_ms)
 
diff --git a/kafka/producer/sender.py b/kafka/producer/sender.py
index 4a88b2f..7a4c557 100644
--- a/kafka/producer/sender.py
+++ b/kafka/producer/sender.py
@@ -77,7 +77,7 @@ class Sender(threading.Thread):
         queue.pop()
         heapq.heapify(queue)
 
-    def _get_expired_inflight_batches(self):
+    def _get_expired_inflight_batches(self, now=None):
         """Get the in-flight batches that has reached delivery timeout."""
         expired_batches = []
         to_remove = []
@@ -174,7 +174,7 @@ class Sender(threading.Thread):
     def _send_producer_data(self, now=None):
         now = time.time() if now is None else now
         # get the list of partitions with data ready to send
-        result = self._accumulator.ready(self._metadata)
+        result = self._accumulator.ready(self._metadata, now=now)
         ready_nodes, next_ready_check_delay, unknown_leaders_exist = result
 
         # if there are any partitions whose leaders are not known yet, force
@@ -195,7 +195,7 @@ class Sender(threading.Thread):
 
         # create produce requests
         batches_by_node = self._accumulator.drain(
-            self._metadata, ready_nodes, self.config['max_request_size'])
+            self._metadata, ready_nodes, self.config['max_request_size'], now=now)
 
         for batch_list in six.itervalues(batches_by_node):
             for batch in batch_list:
@@ -209,8 +209,9 @@ class Sender(threading.Thread):
                 for batch in batch_list:
                     self._accumulator.muted.add(batch.topic_partition)
 
-        expired_batches = self._accumulator.expired_batches()
-        expired_batches.extend(self._get_expired_inflight_batches())
+        self._accumulator.reset_next_batch_expiry_time()
+        expired_batches = self._accumulator.expired_batches(now=now)
+        expired_batches.extend(self._get_expired_inflight_batches(now=now))
 
         if expired_batches:
             log.debug("%s: Expired %s batches in accumulator", str(self), len(expired_batches))


Test patch:
diff --git a/test/test_sender.py b/test/test_sender.py
index 0731454..6d29c1e 100644
--- a/test/test_sender.py
+++ b/test/test_sender.py
@@ -240,3 +240,16 @@ def test_maybe_wait_for_producer_id():
 
 def test_run_once():
     pass
+
+
+def test__send_producer_data_expiry_time_reset(sender, accumulator, mocker):
+    now = time.time()
+    tp = TopicPartition('foo', 0)
+    mocker.patch.object(sender, '_failed_produce')
+    result = accumulator.append(tp, 0, b'key', b'value', [], now=now)
+    poll_timeout_ms = sender._send_producer_data(now=now)
+    assert poll_timeout_ms == accumulator.config['delivery_timeout_ms']
+    sender._failed_produce.assert_not_called()
+    now += accumulator.config['delivery_timeout_ms']
+    poll_timeout_ms = sender._send_producer_data(now=now)
+    assert poll_timeout_ms > 0