Fix bug in SimpleBufferPool memory condition waiting / timeout

dpkp · dpkp · commit 799f53f71275 · 2016-02-18T21:49:37.000-08:00
diff --git a/kafka/producer/buffer.py b/kafka/producer/buffer.py
@@ -158,15 +158,16 @@ def allocate(self, size, max_time_to_block_ms):
                 # enough memory to allocate one
                 while buf is None:
                     start_wait = time.time()
-                    if not more_memory.wait(max_time_to_block_ms / 1000.0):
-                        raise Errors.KafkaTimeoutError(
-                            "Failed to allocate memory within the configured"
-                            " max blocking time")
+                    more_memory.wait(max_time_to_block_ms / 1000.0)
                     end_wait = time.time()
                     #this.waitTime.record(endWait - startWait, time.milliseconds());
 
                     if self._free:
                         buf = self._free.popleft()
+                    else:
+                        raise Errors.KafkaTimeoutError(
+                            "Failed to allocate memory within the configured"
+                            " max blocking time")
 
                 # remove the condition for this thread to let the next thread
                 # in line start getting memory
