Kafka Python Next Generation Client

Warnings

• breaking 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).
  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`.
• gotcha 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.
  Fix: Use a separate `KafkaConsumer` instance per thread, or preferably, use multiprocessing for concurrent consumption.
• breaking Support for End-of-Life (EOL) Python versions was removed starting from `v2.1.0`. The library now explicitly requires Python >=3.8.
  Fix: Upgrade your Python environment to version 3.8 or newer.
• gotcha 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.
  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.
• gotcha 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.
  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.
• gotcha 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.
  Fix: Upgrade to `kafka-python-ng` version `2.0.3` or newer to resolve this import error.

Install

• pip install kafka-python-ng Install stable version

Imports

• KafkaProducer

  from kafka import KafkaProducer

• KafkaConsumer

  from kafka import KafkaConsumer

• KafkaAdminClient

  from kafka import KafkaAdminClient

Quickstart

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.

import os
import json
import time
from kafka import KafkaProducer, KafkaConsumer

# Configure Kafka bootstrap servers, use environment variable for production readiness
BOOTSTRAP_SERVERS = os.environ.get('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092').split(',')
TOPIC_NAME = 'my_test_topic'

def produce_messages():
    producer = KafkaProducer(
        bootstrap_servers=BOOTSTRAP_SERVERS,
        value_serializer=lambda v: json.dumps(v).encode('utf-8')
    )
    print(f"Producing messages to topic: {TOPIC_NAME}")
    for i in range(5):
        message = {'number': i, 'timestamp': time.time()}
        producer.send(TOPIC_NAME, message)
        print(f"Sent: {message}")
        time.sleep(1)
    producer.flush()
    producer.close()
    print("Producer finished.")

def consume_messages():
    consumer = KafkaConsumer(
        TOPIC_NAME,
        bootstrap_servers=BOOTSTRAP_SERVERS,
        auto_offset_reset='earliest',
        group_id='my_python_group',
        value_deserializer=lambda m: json.loads(m.decode('utf-8'))
    )
    print(f"Consuming messages from topic: {TOPIC_NAME} (group: my_python_group)")
    for message in consumer:
        print(f"Received: Topic={message.topic}, Partition={message.partition}, Offset={message.offset}, Value={message.value}")
    consumer.close()
    print("Consumer finished.")

if __name__ == "__main__":
    # In a real application, producer and consumer would likely run in separate processes/threads
    # For this quickstart, we'll run producer, then consumer sequentially
    # Ensure a Kafka broker is running at 'localhost:9092' or specify KAFKA_BOOTSTRAP_SERVERS env var
    produce_messages()
    print("\nWaiting for a moment before consuming...")
    time.sleep(5) # Give Kafka time to process
    consume_messages()