Update docstring, always force api version to be in sync with actual broker running when building producer

Author: Gabriel Tincu

docs/index.rst

@@ -122,11 +122,11 @@ multiprocessing is recommended.
 Compression
 ***********
 
-kafka-python supports gzip compression/decompression natively. To produce or
-consume lz4 compressed messages, you should install python-lz4 (pip install lz4).
-To enable snappy, install python-snappy (also requires snappy library).
-See `Installation <install.html#optional-snappy-install>`_ for more information.
-
+kafka-python supports multiple compression types. To produce or
+ - gzip : supported natively
+ - lz4 : requires `python-lz4 <https://pypi.org/project/lz4/>`_ installed
+ - snappy : requires the `python-snappy <https://pypi.org/project/python-snappy/>`_  package (which requires the snappy C library)
+ - zstd : requires the `python-zstandard <https://github.com/indygreg/python-zstandard>`_ package installed
 
 Protocol
 ********

test/test_producer.py

@@ -25,7 +25,6 @@ def test_buffer_pool():
 @pytest.mark.skipif(not env_kafka_version(), reason="No KAFKA_VERSION set")
 @pytest.mark.parametrize("compression", [None, 'gzip', 'snappy', 'lz4', 'zstd'])
 def test_end_to_end(kafka_broker, compression):
-
     if compression == 'lz4':
         # LZ4 requires 0.8.2
         if env_kafka_version() < (0, 8, 2):
@@ -36,13 +35,12 @@ def test_end_to_end(kafka_broker, compression):
 
     if compression == 'zstd' and env_kafka_version() < (2, 1, 0):
         return
-    env_version = env_kafka_version()
-    api_version = env_version if env_version >= (2, 1, 0) else None
+
     connect_str = ':'.join([kafka_broker.host, str(kafka_broker.port)])
     producer = KafkaProducer(bootstrap_servers=connect_str,
                              retries=5,
                              max_block_ms=30000,
-                             api_version=api_version,
+                             api_version=env_kafka_version(),
                              compression_type=compression,
                              value_serializer=str.encode)
     consumer = KafkaConsumer(bootstrap_servers=connect_str,
@@ -91,11 +89,9 @@ def test_kafka_producer_proper_record_metadata(kafka_broker, compression):
     if compression == 'zstd' and env_kafka_version() < (2, 1, 0):
         return
     connect_str = ':'.join([kafka_broker.host, str(kafka_broker.port)])
-    env_version = env_kafka_version()
-    api_version = env_version if env_version >= (2, 1, 0) else None
     producer = KafkaProducer(bootstrap_servers=connect_str,
                              retries=5,
-                             api_version=api_version,
+                             api_version=env_kafka_version(),
                              max_block_ms=30000,
                              compression_type=compression)
     magic = producer._max_usable_produce_magic()