{"id":4367,"library":"kafka-python-ng","title":"Kafka Python Next Generation Client","description":"kafka-python-ng is a pure Python client for Apache Kafka, designed to function similarly to the official Java client, but with Pythonic interfaces. It provides high-level producer and consumer APIs, as well as admin functionality. The library is actively maintained, with frequent releases, and is compatible with Kafka brokers from version 0.8.0 up to 2.6+ (with optimal features for 0.9+). The current version is 2.2.3 and requires Python >=3.8.","status":"active","version":"2.2.3","language":"python","source_language":"en","source_url":"https://github.com/wbarnha/kafka-python-ng","tags":["kafka","messaging","pubsub","streaming","apache-kafka","client"],"install":[{"cmd":"pip install kafka-python-ng","lang":"bash","label":"Install stable version"}],"dependencies":[{"reason":"Optional C-optimized CRC32 validation for improved performance.","package":"crc32c","optional":true},{"reason":"Optional LZ4 compression support.","package":"lz4","optional":true},{"reason":"Optional Snappy compression support.","package":"snappy","optional":true},{"reason":"Optional Zstandard compression support.","package":"zstd","optional":true}],"imports":[{"symbol":"KafkaProducer","correct":"from kafka import KafkaProducer"},{"symbol":"KafkaConsumer","correct":"from kafka import KafkaConsumer"},{"symbol":"KafkaAdminClient","correct":"from kafka import KafkaAdminClient"}],"quickstart":{"code":"import os\nimport json\nimport time\nfrom kafka import KafkaProducer, KafkaConsumer\n\n# Configure Kafka bootstrap servers, use environment variable for production readiness\nBOOTSTRAP_SERVERS = os.environ.get('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092').split(',')\nTOPIC_NAME = 'my_test_topic'\n\ndef produce_messages():\n    producer = KafkaProducer(\n        bootstrap_servers=BOOTSTRAP_SERVERS,\n        value_serializer=lambda v: json.dumps(v).encode('utf-8')\n    )\n    print(f\"Producing messages to topic: {TOPIC_NAME}\")\n    for i in range(5):\n        message = {'number': i, 'timestamp': time.time()}\n        producer.send(TOPIC_NAME, message)\n        print(f\"Sent: {message}\")\n        time.sleep(1)\n    producer.flush()\n    producer.close()\n    print(\"Producer finished.\")\n\ndef consume_messages():\n    consumer = KafkaConsumer(\n        TOPIC_NAME,\n        bootstrap_servers=BOOTSTRAP_SERVERS,\n        auto_offset_reset='earliest',\n        group_id='my_python_group',\n        value_deserializer=lambda m: json.loads(m.decode('utf-8'))\n    )\n    print(f\"Consuming messages from topic: {TOPIC_NAME} (group: my_python_group)\")\n    for message in consumer:\n        print(f\"Received: Topic={message.topic}, Partition={message.partition}, Offset={message.offset}, Value={message.value}\")\n    consumer.close()\n    print(\"Consumer finished.\")\n\nif __name__ == \"__main__\":\n    # In a real application, producer and consumer would likely run in separate processes/threads\n    # For this quickstart, we'll run producer, then consumer sequentially\n    # Ensure a Kafka broker is running at 'localhost:9092' or specify KAFKA_BOOTSTRAP_SERVERS env var\n    produce_messages()\n    print(\"\\nWaiting for a moment before consuming...\")\n    time.sleep(5) # Give Kafka time to process\n    consume_messages()\n","lang":"python","description":"This quickstart demonstrates a basic Kafka producer and consumer. The producer sends JSON-serialized messages to a topic, and the consumer reads and deserializes them. It's configured to connect to `localhost:9092` by default, but can be configured via the `KAFKA_BOOTSTRAP_SERVERS` environment variable for production environments. For the consumer, `auto_offset_reset='earliest'` ensures it starts reading from the beginning of the topic if no offset is committed, and `group_id` enables coordinated consumer group functionality."},"warnings":[{"fix":"Update your `requirements.txt` to `kafka-python-ng` and ensure `from kafka import ...` is used for client classes. If you were explicitly using `kafka_python` in imports, switch to `kafka`.","message":"The project was renamed from `kafka-python` to `kafka-python-ng` in version `2.0.3`. This requires updating `pip install` commands and potentially import paths (`from kafka-python-ng ...` to `from kafka ...` for previous explicit imports).","severity":"breaking","affected_versions":">=2.0.3"},{"fix":"Use a separate `KafkaConsumer` instance per thread, or preferably, use multiprocessing for concurrent consumption.","message":"Unlike `KafkaProducer` which is thread-safe, `KafkaConsumer` is *not thread-safe*. Sharing a single `KafkaConsumer` instance across multiple threads can lead to unexpected behavior or data loss.","severity":"gotcha","affected_versions":"All versions"},{"fix":"Upgrade your Python environment to version 3.8 or newer.","message":"Support for End-of-Life (EOL) Python versions was removed starting from `v2.1.0`. The library now explicitly requires Python >=3.8.","severity":"breaking","affected_versions":">=2.1.0"},{"fix":"Ensure all SSL certificates (CA, client cert, client key) are in PEM format. For JKS, convert them using `keytool` and `openssl`. Verify `ssl_cafile` points to the correct CA certificate chain.","message":"SSL connection issues are common, often related to certificate formats (Java Keystore/JKS vs. PEM) or incorrect `ssl_cafile` paths. Python clients typically require PEM-formatted certificates.","severity":"gotcha","affected_versions":"All versions"},{"fix":"For full consumer group functionality, ensure your Kafka broker version is 0.9 or higher. If using older brokers, you might need to manage partitions and offsets manually.","message":"Fully coordinated consumer groups, dynamic partition assignment, and offset management in `KafkaConsumer` require Kafka brokers version 0.9 or newer. Older brokers (e.g., 0.8.x) may not support these features or require manual partition assignment.","severity":"gotcha","affected_versions":"All versions with Kafka brokers < 0.9"},{"fix":"Upgrade to `kafka-python-ng` version `2.0.3` or newer to resolve this import error.","message":"Versions like `2.0.2` had an import issue (`ModuleNotFoundError: No module named 'kafka.vendor.six.moves'`) on certain Linux distributions (e.g., Rocky Linux 10) with Python 3.12.","severity":"gotcha","affected_versions":"2.0.2"}],"env_vars":null,"search_vec":"'0.8.0':61 '0.9':69 '2.2.3':74 '2.6':64 '3.8':78 'activ':48 'admin':43 'apach':16,84 'apache-kafka':83 'api':39 'broker':58 'client':5,14,26,86 'compat':55 'consum':38 'current':71 'design':18 'featur':67 'frequent':51 'function':20,44 'generat':4 'high':34 'high-level':33 'interfac':30 'java':25 'kafka':1,7,17,57,79,85 'kafka-python-ng':6 'level':35 'librari':46 'maintain':49 'messag':80 'next':3 'ng':9 'offici':24 'optim':66 'produc':36 'provid':32 'pubsub':81 'pure':12 'python':2,8,13,29,77 'releas':52 'requir':76 'similar':21 'stream':82 'version':60,72 'well':41","created_at":"2026-04-12T08:52:44.848918+00:00","updated_at":"2026-04-17T14:24:33.183582+00:00","problems":[{"fix":"Install the package using pip: 'pip install kafka-python-ng'.","cause":"The 'kafka-python-ng' package is not installed in the current Python environment.","error":"ModuleNotFoundError: No module named 'kafka'"},{"fix":"Uninstall any existing 'kafka' packages and install 'kafka-python-ng' from the GitHub repository: 'pip uninstall kafka kafka-python; pip install git+https://github.com/wbarnha/kafka-python-ng.git'.","cause":"The 'kafka-python-ng' package is not installed or is outdated.","error":"ImportError: cannot import name 'IncompatibleBrokerVersion' from 'kafka.errors'"},{"fix":"Ensure 'kafka-python-ng' is installed and rename any local files named 'kafka.py' to avoid conflicts.","cause":"The 'kafka-python-ng' package is not installed or there is a naming conflict with a local file named 'kafka.py'.","error":"ImportError: cannot import name 'KafkaConsumer'"},{"fix":"Use the correct import statement: 'from kafka import KafkaProducer'.","cause":"Incorrect capitalization in the import statement; Python is case-sensitive.","error":"ImportError: cannot import name 'kafkaProducer'"},{"fix":"Install the 'kafka-python-ng' package: 'pip install kafka-python-ng'.","cause":"Compatibility issues with Python 3.12 and the 'kafka-python' package.","error":"ModuleNotFoundError: No module named 'kafka.vendor.six.moves'"}],"ecosystem":"pypi","meta_description":null,"install_score":null,"quickstart_score":null,"quickstart_tag":null,"pypi_latest":"2.2.3","cli_name":"","cli_version":null,"type":"library","homepage":null,"github":"https://github.com/wbarnha/kafka-python-ng","docs":null,"changelog":null,"pypi":"https://pypi.org/project/kafka-python-ng/","npm":null,"openapi_spec":null,"status_page":null,"smithery":null,"categories":["http-networking","database","data"],"base_url":null,"auth_type":null,"provenance":{"verified_status":"passing","verified_at":"2026-06-28","last_verified":"2026-08-29","next_check":"2026-07-28","install_tag":null}}