Increase producer poll timeout and retries (#771)

Author: ovv

quixstreams/kafka/producer.py

@@ -30,6 +30,9 @@
     KafkaError._DESTROY,  # noqa: SLF001
 )
 
+PRODUCER_POLL_TIMEOUT = 30.0
+PRODUCER_ON_ERROR_RETRIES = 10
+
 
 def _default_error_cb(error: KafkaError):
     error_code = error.code()
@@ -86,8 +89,8 @@ def produce(
         headers: Optional[Headers] = None,
         partition: Optional[int] = None,
         timestamp: Optional[int] = None,
-        poll_timeout: float = 5.0,
-        buffer_error_max_tries: int = 3,
+        poll_timeout: float = PRODUCER_POLL_TIMEOUT,
+        buffer_error_max_tries: int = PRODUCER_ON_ERROR_RETRIES,
         on_delivery: Optional[DeliveryCallback] = None,
     ):
         """

quixstreams/rowproducer.py

@@ -11,7 +11,12 @@
 from .error_callbacks import ProducerErrorCallback, default_on_producer_error
 from .kafka.configuration import ConnectionConfig
 from .kafka.exceptions import KafkaProducerDeliveryError
-from .kafka.producer import Producer, TransactionalProducer
+from .kafka.producer import (
+    PRODUCER_ON_ERROR_RETRIES,
+    PRODUCER_POLL_TIMEOUT,
+    Producer,
+    TransactionalProducer,
+)
 from .models import Headers, Row, Topic
 
 logger = logging.getLogger(__name__)
@@ -180,8 +185,8 @@ def produce(
         headers: Optional[Headers] = None,
         partition: Optional[int] = None,
         timestamp: Optional[int] = None,
-        poll_timeout: float = 5.0,
-        buffer_error_max_tries: int = 3,
+        poll_timeout: float = PRODUCER_POLL_TIMEOUT,
+        buffer_error_max_tries: int = PRODUCER_ON_ERROR_RETRIES,
     ):
         self._raise_for_error()
 

quixstreams/sources/base/source.py

@@ -3,6 +3,7 @@
 from typing import Callable, Optional, Union
 
 from quixstreams.checkpointing.exceptions import CheckpointProducerTimeout
+from quixstreams.kafka.producer import PRODUCER_ON_ERROR_RETRIES, PRODUCER_POLL_TIMEOUT
 from quixstreams.models.messages import KafkaMessage
 from quixstreams.models.topics import Topic
 from quixstreams.models.types import Headers
@@ -322,8 +323,8 @@ def produce(
         headers: Optional[Headers] = None,
         partition: Optional[int] = None,
         timestamp: Optional[int] = None,
-        poll_timeout: float = 5.0,
-        buffer_error_max_tries: int = 3,
+        poll_timeout: float = PRODUCER_POLL_TIMEOUT,
+        buffer_error_max_tries: int = PRODUCER_ON_ERROR_RETRIES,
     ) -> None:
         """
         Produce a message to the configured source topic in Kafka.

tests/test_quixstreams/test_sources/test_core/test_csv.py

@@ -44,11 +44,11 @@ def test_read(self, tmp_path, producer):
         assert producer.produce.called
         assert producer.produce.call_count == 5
         assert producer.produce.call_args.kwargs == {
-            "buffer_error_max_tries": 3,
+            "buffer_error_max_tries": 10,
             "headers": None,
             "key": b"key5",
             "partition": None,
-            "poll_timeout": 5.0,
+            "poll_timeout": 30.0,
             "timestamp": 5,
             "topic": name,
             "value": b'{"key":"key5","field":"value5","timestamp":"5"}',
@@ -79,11 +79,11 @@ def test_read_no_extractors(self, tmp_path, producer):
         assert producer.produce.called
         assert producer.produce.call_count == 5
         assert producer.produce.call_args.kwargs == {
-            "buffer_error_max_tries": 3,
+            "buffer_error_max_tries": 10,
             "headers": None,
             "key": None,
             "partition": None,
-            "poll_timeout": 5.0,
+            "poll_timeout": 30.0,
             "timestamp": None,
             "topic": name,
             "value": b'{"key":"key5","field":"value5","timestamp":"5"}',