PyKafka

Common errors

• TypeError: argument of type 'bytes' is not iterable

  cause Attempting to pass a byte string to a `consumer_group` parameter or `client.topics[...]` accessor that now expects a `str`.
  fix Decode the byte string to a UTF-8 string: `consumer_group=b'my_group'.decode('utf-8')` or `client.topics[b'my_topic'].decode('utf-8')`.

• pykafka.exceptions.NoBrokersAvailableError: No brokers available for topic 'my_topic'

  cause The KafkaClient could not connect to any of the specified Kafka brokers, or the topic does not exist/is not discoverable.
  fix Verify that Kafka brokers are running and accessible from the client machine (check network, firewall). Ensure the `hosts` argument to `KafkaClient` is correct (e.g., `'localhost:9092'`). Check Kafka logs for broker issues or topic existence.

• pykafka.exceptions.SocketDisconnectedError: Tried to use a connection that was disconnected

  cause The client's connection to a Kafka broker was unexpectedly lost during an operation.
  fix This can be transient. Often, catching the exception and re-instantiating the PyKafka component (e.g., `KafkaClient`, `Producer`, `Consumer`) or calling `stop()` then `start()` on the component might re-establish connection. Check network stability and Kafka broker health.

• pykafka.exceptions.GroupAuthorizationFailed: [Error 29] Group Authorization Failed: This consumer group is not authorized to access this topic.

  cause The Kafka user associated with the client does not have the necessary permissions to access the specified consumer group or topic.
  fix Check your Kafka ACLs (Access Control Lists) to ensure that the client has read/write permissions for the topic and consumer group, and `READ` permission on the cluster for consumer group management.

Warnings

• breaking In PyKafka 2.8.0, the `consumer_group` keyword argument for consumer components and the parameter for `TopicDict.__getitem__` (when accessing topics) changed their expected type from `bytes` to `str`. If you were passing byte strings, this will now raise a `TypeError`.
  Fix: Ensure `consumer_group` and topic names are passed as `str` instead of `bytes`. Convert existing byte strings using `.decode('utf-8')` if necessary, or ensure they are defined as `str`.
• gotcha When using `Producer` with `delivery_reports=True`, it's critical to regularly drain the delivery report queue. Failing to do so can lead to unbounded memory growth, as reports are stored in memory until consumed.
  Fix: Implement a loop to regularly call `producer.get_delivery_report()` and process the returned reports to prevent memory leaks. This is often done in a separate thread or an asynchronous callback.
• gotcha The behavior of `auto_offset_reset` and `reset_offset_on_start` in consumers can be counter-intuitive. `reset_offset_on_start=True` will *always* reset the offset based on `auto_offset_reset` for the first fetch, even if committed offsets exist. If `False`, it will use committed offsets if available.
  Fix: Carefully review the `Consumer Patterns` documentation regarding `auto_offset_reset` (`OffsetType.EARLIEST` or `OffsetType.LATEST`) and `reset_offset_on_start` to ensure the consumer starts at the intended offset, especially in production environments to avoid message loss or reprocessing.
• gotcha While PyKafka is a pure-Python client, its high-performance C extension (librdkafka) is highly recommended for production use. Without it, performance can be significantly lower, and certain features might be less robust. Installation of `librdkafka` requires system-level dependencies.
  Fix: Install `librdkafka` development packages on your system (e.g., `librdkafka-dev` on Debian/Ubuntu, `librdkafka-devel` on RHEL/CentOS) and then install pykafka using `RDKAFKA_INSTALL=system pip install pykafka`.

Install

• pip install pykafka Install PyKafka
• RDKAFKA_INSTALL=system pip install pykafka Install with librdkafka C extension (recommended)

Imports

• KafkaClient

  from pykafka import KafkaClient

• OffsetType

  from pykafka.common import OffsetType

• Producer
  wrong:

  from pykafka import Producer

  correct:

  from pykafka.producer import Producer

  Producer is typically accessed via topic.get_producer() after initializing KafkaClient and Topic.
• SimpleConsumer
  wrong:

  from pykafka import SimpleConsumer

  correct:

  from pykafka.simpleconsumer import SimpleConsumer

  SimpleConsumer is typically accessed via topic.get_simple_consumer() after initializing KafkaClient and Topic.
• BalancedConsumer
  wrong:

  from pykafka import BalancedConsumer

  correct:

  from pykafka.balancedconsumer import BalancedConsumer

  BalancedConsumer is typically accessed via topic.get_balanced_consumer() after initializing KafkaClient and Topic, often requiring Zookeeper connection.

Quickstart

This quickstart demonstrates how to initialize a PyKafka client, produce a few messages to a topic, and then consume them using a `SimpleConsumer`. It's configured to connect to a local Kafka instance and resets the consumer offset to the earliest available for demonstration purposes. Ensure `KAFKA_HOSTS`, `KAFKA_TOPIC`, and `KAFKA_CONSUMER_GROUP` environment variables are set or default to `localhost:9092` and `test-topic`, `my-consumer-group` respectively.

import os
import time
from pykafka import KafkaClient
from pykafka.common import OffsetType

KAFKA_HOSTS = os.environ.get('KAFKA_HOSTS', 'localhost:9092')
TOPIC_NAME = os.environ.get('KAFKA_TOPIC', 'test-topic').encode('utf-8')
CONSUMER_GROUP = os.environ.get('KAFKA_CONSUMER_GROUP', 'my-consumer-group').encode('utf-8')

# 1. Connect to Kafka
client = KafkaClient(hosts=KAFKA_HOSTS)
topic = client.topics[TOPIC_NAME]

# 2. Produce a message
with topic.get_producer() as producer:
    print(f"Producing message to {TOPIC_NAME.decode()}...")
    for i in range(5):
        message_value = f"test message {i}".encode('utf-8')
        producer.produce(message_value)
        print(f"Sent: {message_value.decode()}")
    print("Finished producing messages.")

# Give Kafka a moment to process
time.sleep(2)

# 3. Consume messages
print(f"\nConsuming messages from {TOPIC_NAME.decode()} (group: {CONSUMER_GROUP.decode()})...")
consumer = topic.get_simple_consumer(
    consumer_group=CONSUMER_GROUP,
    auto_commit_enable=True,
    auto_offset_reset=OffsetType.EARLIEST, # Start from the beginning if no offset is committed
    reset_offset_on_start=True # Force reset on start, useful for quickstart
)

for i, message in enumerate(consumer):
    if message is not None:
        print(f"Received: Offset={message.offset}, Value={message.value.decode()}")
    if i >= 4: # Consume the 5 messages we sent
        break

consumer.stop()
print("Finished consuming messages.")