Message Accumulator: add_message being recursive eventually overflows
@@ -315,6 +315,7 @@ class MessageAccumulator:
If batch is already full this method waits (`timeout` seconds maximum)
until batch is drained by send task
"""
+ while True:
if self._closed:
# this can happen when producer is closing but try to send some
# messages in async task
@@ -330,7 +331,8 @@ class MessageAccumulator:
batch = pending_batches[-1]
future = batch.append(key, value, timestamp_ms, headers=headers)
- if future is None:
+ if future is not None:
+ return future
# Batch is full, can't append data atm,
# waiting until batch per topic-partition is drained
start = self._loop.time()
@@ -338,9 +340,6 @@ class MessageAccumulator:
timeout -= self._loop.time() - start
if timeout <= 0:
raise KafkaTimeoutError()
- return (await self.add_message(
- tp, key, value, timeout, timestamp_ms))
- return future
def data_waiter(self):
""" Return waiter future that will be resolved when accumulator contain
