confluent-kafka-stubs for Type Hinting

Warnings

• gotcha confluent-kafka-stubs provides only type information for static analysis (e.g., MyPy, IDE auto-completion). It does not contain any executable code or change the runtime behavior of `confluent-kafka`. You must still install and use the `confluent-kafka` library for your application to function.
  Fix: Always install `confluent-kafka` alongside `confluent-kafka-stubs`: `pip install confluent-kafka confluent-kafka-stubs`.
• gotcha Mismatching versions between `confluent-kafka` and `confluent-kafka-stubs` can lead to incorrect type hints or MyPy errors. If the underlying library's API changes significantly, the stubs might become outdated.
  Fix: Keep `confluent-kafka-stubs` updated to a version compatible with your `confluent-kafka` installation. Generally, they are developed together within the same repository, so using the latest available stubs is recommended.
• gotcha Some advanced features or dynamic behaviors of `confluent-kafka` (e.g., highly dynamic configurations or callback signatures) might not be perfectly covered by static stubs, leading to occasional `Any` types or less precise hints.
  Fix: While stubs significantly improve type safety, be aware that complex scenarios might still require some manual type annotation or `type: ignore` comments in your code. Consult `confluent-kafka`'s actual documentation for definitive API behavior.

Install

• pip install confluent-kafka-stubs Install stubs
• pip install confluent-kafka Install core library (required)

Imports

• KafkaProducer, KafkaConsumer, Message

  from confluent_kafka import Producer, Consumer, Message

  confluent-kafka-stubs provides type hints for the 'confluent_kafka' library. You do not import directly from 'confluent_kafka_stubs' at runtime. Your code should import from 'confluent_kafka' as usual, and the stubs will be picked up by type checkers.

Quickstart

This quickstart demonstrates a simple producer and consumer using `confluent-kafka`. To leverage the `confluent-kafka-stubs` for type checking, ensure both `confluent-kafka` and `confluent-kafka-stubs` are installed in your environment. You can then run a static analysis tool like MyPy on your code to benefit from the provided type hints.

import os
from confluent_kafka import Producer, Consumer, KafkaException

def delivery_report(err, msg):
    if err is not None:
        print(f"Message delivery failed: {err}")
    else:
        print(f"Message delivered to {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}")

def run_example():
    # Configuration for Kafka (example with local Kafka or Confluent Cloud)
    # Replace with your actual Kafka broker(s)
    broker = os.environ.get('KAFKA_BROKER', 'localhost:9092')
    conf = {
        'bootstrap.servers': broker,
        # 'sasl.username': os.environ.get('KAFKA_USERNAME'),
        # 'sasl.password': os.environ.get('KAFKA_PASSWORD'),
        # 'security.protocol': 'SASL_SSL',
        # 'sasl.mechanisms': 'PLAIN',
        'acks': 'all'
    }

    topic = 'my_example_topic'

    # Producer
    print("--- Producer ---")
    producer = Producer(conf)
    try:
        producer.produce(topic, key='key1', value='hello world from stubs!', callback=delivery_report)
        producer.flush(10) # Wait for up to 10 seconds for any outstanding messages to be delivered
    except KafkaException as e:
        print(f"Producer error: {e}")

    # Consumer
    print("\n--- Consumer ---")
    consumer_conf = {
        'bootstrap.servers': broker,
        'group.id': 'my_consumer_group',
        'auto.offset.reset': 'earliest'
    }
    consumer = Consumer(consumer_conf)
    try:
        consumer.subscribe([topic])
        # Poll for messages for a short duration
        msg = consumer.poll(timeout=5.0)
        if msg is None:
            print("No message received within timeout.")
        elif msg.error():
            print(f"Consumer error: {msg.error()}")
        else:
            print(f"Received message: key={msg.key().decode('utf-8')}, value={msg.value().decode('utf-8')}")
    except KafkaException as e:
        print(f"Consumer error: {e}")
    finally:
        consumer.close()

if __name__ == '__main__':
    # To run this with type checking, save as e.g., 'example.py'
    # Then run 'mypy example.py'
    # And 'python example.py' (ensure Kafka broker is running)
    run_example()