Bump kafka-python dependency to support 2.3.1

journalpump/senders/kafka_sender.py

@@ -2,6 +2,7 @@
 from kafka import errors, KafkaAdminClient, KafkaProducer
 from kafka.admin import NewTopic
 
+import inspect
 import logging
 import socket
 
@@ -15,7 +16,23 @@
 except ImportError:
     zstd = None
 
-KAFKA_CONN_ERRORS = tuple(errors.RETRY_ERROR_TYPES) + (
+
+def _get_retriable_kafka_errors():
+    """Build the set of retriable Kafka error types.
+
+    kafka-python < 2.1 provides errors.RETRY_ERROR_TYPES.
+    kafka-python >= 2.1 marks individual error classes with retriable = True.
+    """
+    if hasattr(errors, "RETRY_ERROR_TYPES"):
+        return tuple(errors.RETRY_ERROR_TYPES)
+    return tuple(
+        obj
+        for _name, obj in inspect.getmembers(errors)
+        if inspect.isclass(obj) and issubclass(obj, errors.KafkaError) and getattr(obj, "retriable", False)
+    )
+
+
+KAFKA_CONN_ERRORS = _get_retriable_kafka_errors() + (
     errors.UnknownError,
     socket.timeout,
     TimeoutError,

requirements.txt

@@ -1,4 +1,4 @@
-kafka-python<2.0.4
+kafka-python<2.4.0
 requests
 websockets<14.0
 aiohttp-socks

setup.py

@@ -1,7 +1,8 @@
-from journalpump import __version__
+import os
+
 from setuptools import find_packages, setup
 
-import os
+from journalpump import __version__
 
 setup(
     name="journalpump",
@@ -10,10 +11,11 @@
     packages=find_packages(exclude=["test", "systest"]),
     extras_require={},
     install_requires=[
-        "kafka-python",
+        "kafka-python<2.4.0",
         "requests",
-        "websockets",
+        "websockets<14.0",
         "aiohttp-socks",
+        "python-snappy",
         "botocore",
         "google-api-python-client",
         "google-auth",