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:

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.
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),

pip install -e .
parse_log_pytest
requirements.txt
pytest pytest-cov pytest-catchlog pytest-pylint pytest-sugar pytest-mock mock python-snappy lz4tools xxhash pytest
apt-get update apt-get install -y gcc libsnappy-dev
3.5
docs/requirements.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
alabaster==0.7.13
astroid==2.11.7
attrs==22.2.0
Babel==2.11.0
certifi==2021.5.30
charset-normalizer==2.0.12
coverage==6.2
cramjam==2.5.0
dill==0.3.4
docutils==0.18.1
idna==3.10
imagesize==1.4.1
importlib-metadata==4.8.3
iniconfig==1.1.1
isort==5.10.1
Jinja2==3.0.3
-e git+https://github.com/dpkp/kafka-python.git@bcb4009b935fb74e3ca71206466c68ad74bc7b3c#egg=kafka_python
lazy-object-proxy==1.7.1
lz4tools==1.3.1.2
MarkupSafe==2.0.1
mccabe==0.7.0
mock==5.2.0
packaging==21.3
platformdirs==2.4.0
pluggy==1.0.0
pockets==0.9.1
py==1.11.0
Pygments==2.14.0
pylint==2.13.9
pyparsing==3.1.4
pytest==7.0.1
pytest-catchlog==1.2.2
pytest-cov==4.0.0
pytest-mock==3.6.1
pytest-pylint==0.18.0
pytest-sugar==0.9.6
python-snappy==0.7.3
pytz==2025.2
requests==2.27.1
six==1.17.0
snowballstemmer==2.2.0
Sphinx==5.3.0
sphinx-rtd-theme==2.0.0
sphinxcontrib-applehelp==1.0.2
sphinxcontrib-devhelp==1.0.2
sphinxcontrib-htmlhelp==2.0.0
sphinxcontrib-jquery==4.1
sphinxcontrib-jsmath==1.0.1
sphinxcontrib-napoleon==0.7
sphinxcontrib-qthelp==1.0.3
sphinxcontrib-serializinghtml==1.1.5
termcolor==1.1.0
toml==0.10.2
tomli==1.2.3
typed-ast==1.5.5
typing_extensions==4.1.1
urllib3==1.26.20
wrapt==1.16.0
xxhash==3.2.0
zipp==3.6.0

name: kafka-python
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=21.2.2=py36h06a4308_0
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - alabaster==0.7.13
      - astroid==2.11.7
      - attrs==22.2.0
      - babel==2.11.0
      - charset-normalizer==2.0.12
      - coverage==6.2
      - cramjam==2.5.0
      - dill==0.3.4
      - docutils==0.18.1
      - idna==3.10
      - imagesize==1.4.1
      - importlib-metadata==4.8.3
      - iniconfig==1.1.1
      - isort==5.10.1
      - jinja2==3.0.3
      - lazy-object-proxy==1.7.1
      - lz4tools==1.3.1.2
      - markupsafe==2.0.1
      - mccabe==0.7.0
      - mock==5.2.0
      - packaging==21.3
      - platformdirs==2.4.0
      - pluggy==1.0.0
      - pockets==0.9.1
      - py==1.11.0
      - pygments==2.14.0
      - pylint==2.13.9
      - pyparsing==3.1.4
      - pytest==7.0.1
      - pytest-catchlog==1.2.2
      - pytest-cov==4.0.0
      - pytest-mock==3.6.1
      - pytest-pylint==0.18.0
      - pytest-sugar==0.9.6
      - python-snappy==0.7.3
      - pytz==2025.2
      - requests==2.27.1
      - six==1.17.0
      - snowballstemmer==2.2.0
      - sphinx==5.3.0
      - sphinx-rtd-theme==2.0.0
      - sphinxcontrib-applehelp==1.0.2
      - sphinxcontrib-devhelp==1.0.2
      - sphinxcontrib-htmlhelp==2.0.0
      - sphinxcontrib-jquery==4.1
      - sphinxcontrib-jsmath==1.0.1
      - sphinxcontrib-napoleon==0.7
      - sphinxcontrib-qthelp==1.0.3
      - sphinxcontrib-serializinghtml==1.1.5
      - termcolor==1.1.0
      - toml==0.10.2
      - tomli==1.2.3
      - typed-ast==1.5.5
      - typing-extensions==4.1.1
      - urllib3==1.26.20
      - wrapt==1.16.0
      - xxhash==3.2.0
      - zipp==3.6.0
prefix: /opt/conda/envs/kafka-python

test/test_consumer.py::TestKafkaConsumer::test_fetch_max_wait_larger_than_request_timeout_raises test/test_consumer.py::TestKafkaConsumer::test_session_timeout_larger_than_request_timeout_raises
test/test_consumer.py::TestKafkaConsumer::test_non_integer_partitions test/test_consumer.py::TestMultiProcessConsumer::test_partition_list test/test_consumer.py::TestSimpleConsumer::test_simple_consumer_commit_does_not_raise test/test_consumer.py::TestSimpleConsumer::test_simple_consumer_failed_payloads test/test_consumer.py::TestSimpleConsumer::test_simple_consumer_leader_change test/test_consumer.py::TestSimpleConsumer::test_simple_consumer_reset_partition_offset test/test_consumer.py::TestSimpleConsumer::test_simple_consumer_unknown_topic_partition test/test_util.py::UtilTest::test_group_by_topic_and_partition test/test_util.py::UtilTest::test_read_short_string test/test_util.py::UtilTest::test_relative_unpack2 test/test_util.py::UtilTest::test_relative_unpack3 test/test_util.py::UtilTest::test_write_int_string test/test_util.py::UtilTest::test_write_int_string__empty test/test_util.py::UtilTest::test_write_int_string__null test/test_util.py::UtilTest::test_write_int_string__unicode
kafka/consumer/group.py kafka/producer/buffer.py kafka/client_async.py kafka/producer/kafka.py kafka/util.py kafka/conn.py
kafka/consumer/group.py kafka/producer/buffer.py kafka/client_async.py kafka/producer/kafka.py kafka/util.py kafka/conn.py