DOCSEARCHbyAlgolia
Submit Docs

confluent-kafka-python

repository·master·Indexed 19 days ago

https://github.com/confluentinc/confluent-kafka-python

Confluent's high-performance Python client for Apache Kafka, built on librdkafka. Version 2.15.0 features include AsyncIO support via AIOProducer and AIOConsumer, Schema Registry integration with Avro, JSON, and Protobuf serializers, and support for FIPS 140-2 and 140-3 compliance. It also implements KIP-932 with ShareConsumer for queue-like record acknowledgement.

Tokens
33.5K
Snippets
90
Records
143
Agent score
66%

What's inside confluent-kafka-python

  1. Overview of the confluent_kafka API

    master

    The confluent_kafka library is a reliable, performant, and feature-rich Python client for Apache Kafka (v0.8 and above). It provides several high-level APIs for interacting with Kafka, including producers, consumers, and administrative tools.

    Core Client APIs

    • Producer: For sending messages to Kafka topics.
    • Consumer: For reading messages from Kafka topics.
    • ShareConsumer: For using Kafka's shared consumer capabilities (Queues for Kafka).
    • AdminClient: For managing Kafka cluster resources (topics, configs, ACLs, etc.).
    • SchemaRegistryClient: For interacting with the Confluent Schema Registry to manage Avro, JSON Schema, and Protobuf schemas.

    Serialization API

    The library supports various serialization and deserialization formats:

    • Avro
    • JSON Schema
    • Protobuf
    • String
    • Integer
    • Double

    Experimental and Legacy APIs

    • Experimental: SerializingProducer, DeserializingConsumer, and DeserializingShareConsumer. These are subject to breaking changes. It is recommended to use (de)serializers directly instead.
    • Legacy (Deprecated): AvroConsumer and AvroProducer. These will be removed in a future version.

  2. Overview of Confluent's Python Client for Apache Kafka®

    master

    The confluent-kafka-python library provides high-level Producer, Consumer, and AdminClient implementations. It is compatible with Apache Kafka brokers (v0.8+), Confluent Cloud, and Confluent Platform.

    Key advantages over pure Python implementations (like kafka-python) include:

    • Performance: Built on librdkafka (a C library) for high throughput and low latency.
    • AsyncIO Support: Native async/await support via AIOProducer.
    • Enterprise Features: Built-in support for Schema Registry (Avro, Protobuf, JSON Schema), transactions, and exactly-once semantics.
    • Confluent Cloud Optimizations: Automatic zone detection to reduce latency and simplified configuration profiles.

  4. Understand Session Timeout and Fetching in KIP-848

    master

    In the KIP-848 protocol, session timeout is managed by the broker rather than the client:

    • Fetching Behavior: If the Group Coordinator is unreachable, a consumer continues fetching messages but is unable to commit offsets.
    • Fencing: A consumer is only fenced once a heartbeat response is received from the Coordinator.
    • Comparison: In the classic protocol, the client would stop fetching once the session timeout expired.

  5. Decide between AsyncIO and synchronous Producer

    master

    Choosing the right producer depends on your application architecture:

    ScenarioRecommended Producer
    Event-loop based apps (FastAPI, Sanic, aiohttp)AsyncIO Producer (to prevent blocking)
    Scripts & Batch jobsSynchronous Producer (highest throughput)
    High-throughput pipelines (manual thread/process control)Synchronous Producer (call poll()/flush() directly)
    Async servers needing headersSynchronous Producer via run_in_executor

  6. How Share Groups (Queues for Kafka) work

    master

    Share groups (KIP-932) provide queue-like semantics for Apache Kafka. Unlike standard consumer groups where one partition is assigned to exactly one member, a share group allows multiple members to consume from the same partitions cooperatively.

    Key Concepts:

    • Unit of Progress: Progress is measured by individual records rather than committed offsets.
    • Acquisition Lock: When a record is delivered, it is acquired by a member under a time-limited lock (configured on the broker via group.share.record.lock.duration.ms).
    • Lifecycle: A record moves through states: available $\rightarrow$ acquired $\rightarrow$ acknowledged.
    • Redelivery: If a record is not acknowledged before its lock expires, or if it is explicitly RELEASEd, it becomes available again for redelivery.
    • Archiving: If a record is REJECTed or exceeds the broker's delivery-count limit (group.share.delivery.count.limit), it is archived and no longer delivered.
    • Scaling: This model allows you to scale the number of consumers beyond the number of partitions, distributing work like a traditional queue.

  8. Key Features of confluent-kafka-python

    master

    The client includes several advanced features for Kafka integration:

    • High Performance: Leverages librdkafka for stability and speed.
    • Queues for Kafka (Preview): Includes a ShareConsumer (based on [KIP-932]) for queue-like, cooperative consumption with per-record acknowledgement. Note: This is in preview and not recommended for production.
    • AsyncIO Producer: An AIOProducer designed for modern asynchronous Python applications.
    • Schema Registry Integration: Synchronous and asynchronous clients for managing schemas and serialization (Avro, Protobuf, JSON Schema).
    • Confluent Cloud Features: Automatic zone detection and optimized configuration profiles.

  9. Compare AIOProducer to the Synchronous Producer

    master

    When choosing between the standard Producer and the AIOProducer, consider these architectural differences:

    FeatureSynchronous Producer
    Callback MechanismUses polling-based callbacks (requires manual polling)
    ThreadingUser often manages polling/threading logic
    Async SupportNot natively async-first
    FeatureAIOProducer
    Callback MechanismAutomatically schedules callbacks onto the asyncio event loop
    ThreadingAutomatically manages the polling thread in the background
    Async SupportProvides native async/await interfaces for all operations
    LifecycleHandles cleanup and shutdown automatically

  10. Thread Safety for ShareConsumer

    master
    The ShareConsumer is not thread-safe by design. A single instance must not be used concurrently from multiple threads. This follows the KIP-932 design where the consumer is single-threaded, and the application is responsible for its own threading model (e.g., one instance per thread). Concurrent use is detected on a best-effort basis and will raise a ConcurrentModificationException.

  11. Handle errors in ShareConsumer

    master

    The share consumer surfaces errors at three distinct levels. It is critical to check for errors at the appropriate level to ensure correct application behavior.

    1. API-level (call-level) errors: These are raised as exceptions during method calls:

      • ConcurrentModificationException: Raised if the consumer is used concurrently from multiple threads (the consumer is not thread-safe).
      • IllegalStateException: Raised if a method is called in an invalid state (e.g., polling while not subscribed, polling in explicit mode before previous records are acknowledged, or calling APIs from within an acknowledgement-commit callback).
      • KafkaException: For other general call-level failures.

    2. Record-level errors: Reported on the message via Message.error(). The application should check msg.error() on every record before processing data. For these records, librdkafka has already applied internal acknowledgements (e.g., RELEASE for decompression failures, REJECT for corrupt batches), but the application can re-acknowledge them if necessary.

    3. Acknowledgement errors: These are not raised as exceptions but are reported through the commit result:

      • commit_sync(): Returns a mapping of TopicPartition to an optional KafkaException (None on success).
      • commit_async(): Reports a single KafkaException via the acknowledgement-commit callback.

  12. Handle Static Group Membership and Fencing in KIP-848

    master

    When using static group membership via group.instance.id with the KIP-848 protocol, the fencing behavior changes:

    • Fencing: A newly joining member with a duplicate group.instance.id is fenced with UNRELEASED_INSTANCE_ID (fatal). (In the classic protocol, the existing member was fenced instead).
    • Best Practices:
      • Ensure only one active instance exists per group.instance.id.
      • Consumers must shut down cleanly to avoid blocking replacements until the session timeout expires.