Repository: dpkp/kafka-python

Issue:
Once KafkaProducer reports "Failed to allocate memory within the configured" exceptions, it never recovers from the failures
We have a situation there is a message burst in our business codes using Kafka 1.3.2. The message rate might reach 100K/s, each message size is less than 16Kb. The producer is shared between threads and configured as default parameters except 'compression_type='gzip', linger_ms=5, retries=3'.
We observe that during message burst, Kafka producer reports:
“KafkaTimeoutError: KafkaTimeoutError: Failed to allocate memory within the configured max blocking time”
And it never comes back.
Traceback listed below:
Traceback (most recent call last):
  File "/root/search/db_publisher.py", line 47, in publish
    self._publish(message)
  File "/root/search/db_publisher.py", line 41, in _publish
    self.producer.send(self.topic, msg)
  File "/usr/lib/python2.7/site-packages/kafka/producer/kafka.py", line 511, in send
    self.config['max_block_ms'])
  File "/usr/lib/python2.7/site-packages/kafka/producer/record_accumulator.py", line 243, in append
    buf = self._free.allocate(size, max_time_to_block_ms)
  File "/usr/lib/python2.7/site-packages/kafka/producer/buffer.py", line 201, in allocate
    "Failed to allocate memory within the configured"
KafkaTimeoutError: KafkaTimeoutError: Failed to allocate memory within the configured max blocking time

Then I check more and find another exception produced:
Traceback (most recent call last):  
File "/root/search/db_publisher.py", line 47, in publish    
self._publish(message)  
File "/root/search/db_publisher.py", line 41, in _publish    
self.producer.send(self.topic, msg)  
File "/usr/lib/python2.7/site-packages/kafka/producer/kafka.py", line 511, in send    self.config['max_block_ms']) 
File "/usr/lib/python2.7/site-packages/kafka/producer/record_accumulator.py", line 243, in append    
buf = self._free.allocate(size, max_time_to_block_ms) 
File "/usr/lib/python2.7/site-packages/kafka/producer/buffer.py", line 207, in allocate
assert removed is more_memory, 'Wrong condition'
AssertionError: Wrong condition

I check the source codes of above exceptions in code buffer.py (class SimpleBufferPool).
There is something wrong in the situation when producer fails to allocate memory from buffer pool within time limit:
                    if self._free:                                                                                                                                      
                        buf = self._free.popleft()                                                                                                                      
                    else:                                                                                                                                               
                        raise Errors.KafkaTimeoutError(                                                                                                                 
                            "Failed to allocate memory within the configured"                                                                                           
                            " max blocking time") 
Should we popleft the semaphore appended to the waiters before raising Errors.KafkaTimeoutError, so next allocation action no need to wait for "max_time_to_block_ms" milliseconds and thus its Condition assertion can pass? Due to each de-allocation notifies the wrong Condtion objects, each allocation action has to waste "max_time_to_block_ms" to wait the semaphore signal. As consequence, our program is slowing down rapidly and not acceptable for business.
The issue has been re-produced 3 times and I will provide debug information about memory allocation and de-allocation logging.
Anyway, thanks your guys  for this wonderful library.

Discussion:

Patch:
diff --git a/kafka/client_async.py b/kafka/client_async.py
index 1513f39..85de90a 100644
--- a/kafka/client_async.py
+++ b/kafka/client_async.py
@@ -105,10 +105,10 @@ class KafkaClient(object):
             providing a file, only the leaf certificate will be checked against
             this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+.
             default: none.
-        api_version (tuple): specify which kafka API version to use. Accepted
-            values are: (0, 8, 0), (0, 8, 1), (0, 8, 2), (0, 9), (0, 10)
-            If None, KafkaClient will attempt to infer the broker
-            version by probing various APIs. Default: None
+        api_version (tuple): Specify which Kafka API version to use. If set
+            to None, KafkaClient will attempt to infer the broker version by
+            probing various APIs. For the full list of supported versions,
+            see KafkaClient.API_VERSIONS. Default: None
         api_version_auto_timeout_ms (int): number of milliseconds to throw a
             timeout exception from the constructor when checking the broker
             api version. Only applies if api_version is None
diff --git a/kafka/conn.py b/kafka/conn.py
index f4fbb93..88013f0 100644
--- a/kafka/conn.py
+++ b/kafka/conn.py
@@ -114,10 +114,9 @@ class BrokerConnection(object):
             providing a file, only the leaf certificate will be checked against
             this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+.
             default: None.
-        api_version (tuple): specify which Kafka API version to use. Accepted
-            values are: (0, 8, 0), (0, 8, 1), (0, 8, 2), (0, 9), (0, 10)
-            If None, KafkaClient will attempt to infer the broker
-            version by probing various APIs. Default: None
+        api_version (tuple): Specify which Kafka API version to use.
+            Accepted values are: (0, 8, 0), (0, 8, 1), (0, 8, 2), (0, 9),
+            (0, 10). Default: (0, 8, 2)
         api_version_auto_timeout_ms (int): number of milliseconds to throw a
             timeout exception from the constructor when checking the broker
             api version. Only applies if api_version is None
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index 47c721f..8c2ab2d 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -6,6 +6,8 @@ import socket
 import sys
 import time
 
+from kafka.errors import KafkaConfigurationError
+
 from kafka.vendor import six
 
 from kafka.client_async import KafkaClient, selectors
@@ -159,9 +161,9 @@ class KafkaConsumer(six.Iterator):
             providing a file, only the leaf certificate will be checked against
             this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+.
             Default: None.
-        api_version (tuple): Specify which kafka API version to use.
-            If set to None, the client will attempt to infer the broker version
-            by probing various APIs. Default: None
+        api_version (tuple): Specify which Kafka API version to use. If set to
+            None, the client will attempt to infer the broker version by probing
+            various APIs. Different versions enable different functionality.
             Examples:
                 (0, 9) enables full group coordination features with automatic
                     partition assignment and rebalancing,
@@ -171,7 +173,8 @@ class KafkaConsumer(six.Iterator):
                     partition assignment only,
                 (0, 8, 0) enables basic functionality but requires manual
                     partition assignment and offset management.
-            For a full list of supported versions, see KafkaClient.API_VERSIONS
+            For the full list of supported versions, see
+            KafkaClient.API_VERSIONS. Default: None
         api_version_auto_timeout_ms (int): number of milliseconds to throw a
             timeout exception from the constructor when checking the broker
             api version. Only applies if api_version set to 'auto'
@@ -267,6 +270,17 @@ class KafkaConsumer(six.Iterator):
                         new_config, self.config['auto_offset_reset'])
             self.config['auto_offset_reset'] = new_config
 
+        request_timeout_ms = self.config['request_timeout_ms']
+        session_timeout_ms = self.config['session_timeout_ms']
+        fetch_max_wait_ms = self.config['fetch_max_wait_ms']
+        if request_timeout_ms <= session_timeout_ms:
+            raise KafkaConfigurationError(
+                "Request timeout (%s) must be larger than session timeout (%s)" %
+                (request_timeout_ms, session_timeout_ms))
+        if request_timeout_ms <= fetch_max_wait_ms:
+            raise KafkaConfigurationError("Request timeout (%s) must be larger than fetch-max-wait-ms (%s)" %
+                                          (request_timeout_ms, fetch_max_wait_ms))
+
         metrics_tags = {'client-id': self.config['client_id']}
         metric_config = MetricConfig(samples=self.config['metrics_num_samples'],
                                      time_window_ms=self.config['metrics_sample_window_ms'],
diff --git a/kafka/producer/buffer.py b/kafka/producer/buffer.py
index 422d47c..d1eeaf1 100644
--- a/kafka/producer/buffer.py
+++ b/kafka/producer/buffer.py
@@ -197,6 +197,7 @@ class SimpleBufferPool(object):
                     if self._free:
                         buf = self._free.popleft()
                     else:
+                        self._waiters.remove(more_memory)
                         raise Errors.KafkaTimeoutError(
                             "Failed to allocate memory within the configured"
                             " max blocking time")
diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py
index 98d4426..5581f63 100644
--- a/kafka/producer/kafka.py
+++ b/kafka/producer/kafka.py
@@ -224,10 +224,10 @@ class KafkaProducer(object):
             providing a file, only the leaf certificate will be checked against
             this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+.
             default: none.
-        api_version (tuple): specify which kafka API version to use.
-            For a full list of supported versions, see KafkaClient.API_VERSIONS
-            If set to None, the client will attempt to infer the broker version
-            by probing various APIs. Default: None
+        api_version (tuple): Specify which Kafka API version to use. If set to
+            None, the client will attempt to infer the broker version by probing
+            various APIs. For a full list of supported versions, see
+            KafkaClient.API_VERSIONS. Default: None
         api_version_auto_timeout_ms (int): number of milliseconds to throw a
             timeout exception from the constructor when checking the broker
             api version. Only applies if api_version set to 'auto'
diff --git a/kafka/util.py b/kafka/util.py
index bc01154..de8f228 100644
--- a/kafka/util.py
+++ b/kafka/util.py
@@ -4,7 +4,6 @@ import atexit
 import binascii
 import collections
 import struct
-import sys
 from threading import Thread, Event
 import weakref
 
@@ -33,19 +32,6 @@ def write_int_string(s):
         return struct.pack('>i%ds' % len(s), len(s), s)
 
 
-def write_short_string(s):
-    if s is not None and not isinstance(s, six.binary_type):
-        raise TypeError('Expected "%s" to be bytes\n'
-                        'data=%s' % (type(s), repr(s)))
-    if s is None:
-        return struct.pack('>h', -1)
-    elif len(s) > 32767 and sys.version_info < (2, 7):
-        # Python 2.6 issues a deprecation warning instead of a struct error
-        raise struct.error(len(s))
-    else:
-        return struct.pack('>h%ds' % len(s), len(s), s)
-
-
 def read_short_string(data, cur):
     if len(data) < cur + 2:
         raise BufferUnderflowError("Not enough data left")
@@ -62,24 +48,6 @@ def read_short_string(data, cur):
     return out, cur + strlen
 
 
-def read_int_string(data, cur):
-    if len(data) < cur + 4:
-        raise BufferUnderflowError(
-            "Not enough data left to read string len (%d < %d)" %
-            (len(data), cur + 4))
-
-    (strlen,) = struct.unpack('>i', data[cur:cur + 4])
-    if strlen == -1:
-        return None, cur + 4
-
-    cur += 4
-    if len(data) < cur + strlen:
-        raise BufferUnderflowError("Not enough data left")
-
-    out = data[cur:cur + strlen]
-    return out, cur + strlen
-
-
 def relative_unpack(fmt, data, cur):
     size = struct.calcsize(fmt)
     if len(data) < cur + size:


Test patch:
diff --git a/test/test_consumer.py b/test/test_consumer.py
index f3dad16..073a3af 100644
--- a/test/test_consumer.py
+++ b/test/test_consumer.py
@@ -16,6 +16,14 @@ class TestKafkaConsumer(unittest.TestCase):
         with self.assertRaises(AssertionError):
             SimpleConsumer(MagicMock(), 'group', 'topic', partitions = [ '0' ])
 
+    def test_session_timeout_larger_than_request_timeout_raises(self):
+        with self.assertRaises(KafkaConfigurationError):
+            KafkaConsumer(bootstrap_servers='localhost:9092', session_timeout_ms=60000, request_timeout_ms=40000)
+
+    def test_fetch_max_wait_larger_than_request_timeout_raises(self):
+        with self.assertRaises(KafkaConfigurationError):
+            KafkaConsumer(bootstrap_servers='localhost:9092', fetch_max_wait_ms=41000, request_timeout_ms=40000)
+
 
 class TestMultiProcessConsumer(unittest.TestCase):
     @unittest.skipIf(sys.platform.startswith('win'), 'test mocking fails on windows')
diff --git a/test/test_util.py b/test/test_util.py
index 5fc3f69..58e5ab8 100644
--- a/test/test_util.py
+++ b/test/test_util.py
@@ -43,57 +43,11 @@ class UtilTest(unittest.TestCase):
             b'\xff\xff\xff\xff'
         )
 
-    def test_read_int_string(self):
-        self.assertEqual(kafka.util.read_int_string(b'\xff\xff\xff\xff', 0), (None, 4))
-        self.assertEqual(kafka.util.read_int_string(b'\x00\x00\x00\x00', 0), (b'', 4))
-        self.assertEqual(kafka.util.read_int_string(b'\x00\x00\x00\x0bsome string', 0), (b'some string', 15))
-
-    def test_read_int_string__insufficient_data(self):
-        with self.assertRaises(kafka.errors.BufferUnderflowError):
-            kafka.util.read_int_string(b'\x00\x00\x00\x021', 0)
-
-    def test_write_short_string(self):
-        self.assertEqual(
-            kafka.util.write_short_string(b'some string'),
-            b'\x00\x0bsome string'
-        )
-
-    def test_write_short_string__unicode(self):
-        with self.assertRaises(TypeError) as cm:
-            kafka.util.write_short_string(u'hello')
-        #: :type: TypeError
-        te = cm.exception
-        if six.PY2:
-            self.assertIn('unicode', str(te))
-        else:
-            self.assertIn('str', str(te))
-        self.assertIn('to be bytes', str(te))
-
-    def test_write_short_string__empty(self):
-        self.assertEqual(
-            kafka.util.write_short_string(b''),
-            b'\x00\x00'
-        )
-
-    def test_write_short_string__null(self):
-        self.assertEqual(
-            kafka.util.write_short_string(None),
-            b'\xff\xff'
-        )
-
-    def test_write_short_string__too_long(self):
-        with self.assertRaises(struct.error):
-            kafka.util.write_short_string(b' ' * 33000)
-
     def test_read_short_string(self):
         self.assertEqual(kafka.util.read_short_string(b'\xff\xff', 0), (None, 2))
         self.assertEqual(kafka.util.read_short_string(b'\x00\x00', 0), (b'', 2))
         self.assertEqual(kafka.util.read_short_string(b'\x00\x0bsome string', 0), (b'some string', 13))
 
-    def test_read_int_string__insufficient_data2(self):
-        with self.assertRaises(kafka.errors.BufferUnderflowError):
-            kafka.util.read_int_string('\x00\x021', 0)
-
     def test_relative_unpack2(self):
         self.assertEqual(
             kafka.util.relative_unpack('>hh', b'\x00\x01\x00\x00\x02', 0),