kafka-python

Warnings

• breaking Version 2.3.x will be the last release branch to support Python 2. Future major versions will drop Python 2 compatibility entirely.
  Fix: Migrate Python applications to Python 3.8+ (recommended minimum is 3.8+ for full support and testing).
• gotcha The `KafkaConsumer` is not thread-safe. Sharing a single `KafkaConsumer` instance across multiple threads can lead to unpredictable behavior and data corruption. For concurrent consumption, multiprocessing is recommended.
  Fix: Use separate processes for each `KafkaConsumer` instance or ensure thread-local usage.
• gotcha Messages are raw bytes in Kafka. Failing to provide appropriate `key_deserializer` and `value_deserializer` functions to `KafkaConsumer` (and serializers for `KafkaProducer`) can lead to deserialization errors or incorrect data interpretation. JSON, Avro, or Protobuf payloads require explicit (de)serialization.
  Fix: Always configure `value_serializer` (e.g., `json.dumps().encode('utf-8')`) for `KafkaProducer` and `value_deserializer` (e.g., `lambda x: json.loads(x.decode('utf-8'))`) for `KafkaConsumer` if not sending/receiving raw bytes.
• gotcha Not explicitly calling `producer.flush()` or `producer.close()` before application exit may result in lost messages that are still buffered. Similarly, `consumer.close()` is crucial to commit offsets and leave the consumer group cleanly.
  Fix: Always call `producer.flush()` and `producer.close()` for producers, and `consumer.close()` for consumers, ideally within a `finally` block or using a context manager if supported.
• gotcha Misconfiguring consumer `group_id` or `auto_offset_reset`, or not understanding manual offset commits, can lead to message re-processing, data loss, or improper partition rebalances within consumer groups. `group_id=None` disables auto-assignment and offset commits.
  Fix: Ensure `group_id` is consistently set for coordinated consumption. Understand `auto_offset_reset` behavior ('earliest' vs 'latest'). Implement manual offset commits with `enable_auto_commit=False` only when necessary for 'at-least-once' or 'exactly-once' processing guarantees.
• gotcha Setting `request.timeout.ms` (client-side config) too low can exacerbate transient broker issues, leading to premature retries or failures. A shorter timeout might seem intuitive for quicker reactions, but it can worsen performance during temporary network glitches or high broker load.
  Fix: Avoid setting `request.timeout.ms` to very low values without careful consideration of your Kafka cluster's latency and reliability characteristics. The default of 30 seconds is often a good starting point.

Install

• pip install kafka-python Install stable release

Imports

• KafkaProducer

  from kafka import KafkaProducer

• KafkaConsumer

  from kafka import KafkaConsumer

  The PyPI package is `kafka-python`, but the top-level module to import is `kafka`. Importing from `kafka_python` will result in an ImportError.
• KafkaAdminClient

  from kafka.admin import KafkaAdminClient

Quickstart

This quickstart demonstrates basic message production and consumption. It initializes a `KafkaProducer` to send JSON-serialized messages to a topic and a `KafkaConsumer` to read and deserialize those messages, configured to start from the earliest offset within a consumer group. Remember to have a Kafka broker running (e.g., on `localhost:9092`).

import json
import time
from kafka import KafkaProducer, KafkaConsumer

# Ensure Kafka is running, e.g., on localhost:9092
BOOTSTRAP_SERVERS = 'localhost:9092'
TOPIC_NAME = 'my_test_topic'

# --- Producer Example ---
producer = KafkaProducer(
    bootstrap_servers=[BOOTSTRAP_SERVERS],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

try:
    for i in range(5):
        message = {'number': i, 'timestamp': time.time()}
        future = producer.send(TOPIC_NAME, message)
        record_metadata = future.get(timeout=10) # Block for confirmation
        print(f"Produced message: {message} to topic {record_metadata.topic}, partition {record_metadata.partition}, offset {record_metadata.offset}")
        time.sleep(1)
except Exception as e:
    print(f"Producer error: {e}")
finally:
    producer.flush()
    producer.close()
    print("Producer closed.")

# --- Consumer Example ---
consumer = KafkaConsumer(
    TOPIC_NAME,
    bootstrap_servers=[BOOTSTRAP_SERVERS],
    auto_offset_reset='earliest', # Start consuming from the earliest available message
    enable_auto_commit=True,
    group_id='my_python_group',
    value_deserializer=lambda x: json.loads(x.decode('utf-8')),
    consumer_timeout_ms=5000 # Stop waiting for messages after 5 seconds
)

print(f"\nConsuming messages from topic: {TOPIC_NAME}")
try:
    for message in consumer:
        print(f"Consumed message: key={message.key}, value={message.value}, topic={message.topic}, partition={message.partition}, offset={message.offset}")
except Exception as e:
    print(f"Consumer error: {e}")
finally:
    consumer.close()
    print("Consumer closed.")