Welcome to Confluent’s Apache Kafka Python client documentation

Indices and tables

confluent_kafka — Confluent’s Apache Kafka Python client

Consumer

class confluent_kafka.Consumer

High-level Kafka Consumer

Consumer(**kwargs)
Create new Consumer instance using provided configuration dict.
Special configuration properties:
on_commit: Optional callback will be called when a commit request has succeeded or failed.
on_commit(err, partitions)
Parameters:
  • consumer (Consumer) – Consumer instance.
  • err (KafkaError) – Commit error object, or None on success.
  • partitions (list(TopicPartition)) – List of partitions with their committed offsets or per-partition errors.
assign()
assign(partitions)

Set consumer partition assignment to the provided list of TopicPartition and starts consuming.

Parameters:partitions (list(TopicPartition)) – List of topic+partitions and optionally initial offsets to start consuming.
close()

Close down and terminate the Kafka Consumer.

Actions performed:

  • Stops consuming
  • Commits offsets
  • Leave consumer group
Return type:None
commit()
commit([message=None][, offsets=None][, async=True])

Commit a message or a list of offsets.

message and offsets are mutually exclusive, if neither is set the current partition assignment’s offsets are used instead.

Parameters:
  • message (confluent_kafka.Message) – Commit message’s offset+1.
  • offsets (list(TopicPartition)) – List of topic+partitions+offsets to commit.
  • async (bool) – Asynchronous commit, return immediately.
Return type:

None

Raises:

KafkaException

committed()
committed(partitions[, timeout=None])

Retrieve committed offsets for the list of partitions.

Parameters:
  • partitions (list(TopicPartition)) – List of topic+partitions to query for stored offsets.
  • timeout (float) – Request timeout
Returns:

List of topic+partitions with offset and possibly error set.

Return type:

list(TopicPartition)

Raises:

KafkaException

poll()
poll([timeout=None])

Consume messages, calls callbacks and returns events.

The application must check the returned Message object’s Message.error() method to distinguish between proper messages (error() returns None), or an event or error (see error().code() for specifics).

Parameters:timeout (float) – Maximum time to block waiting for message, event or callback.
Returns:A Message object or None on timeout
Return type:Message or None
position()
position(partitions[, timeout=None])

Retrieve current positions (offsets) for the list of partitions.

Parameters:partitions (list(TopicPartition)) – List of topic+partitions to return current offsets for. The current offset is the offset of the last consumed message + 1.
Returns:List of topic+partitions with offset and possibly error set.
Return type:list(TopicPartition)
Raises:KafkaException
subscribe()
subscribe(topics[, listener=None])

Set subscription to supplied list of topics This replaces a previous subscription.

Parameters:
  • topics (list(str)) – List of topics (strings) to subscribe to.
  • on_assign (callable) – callback to provide handling of customized offsets on completion of a successful partition re-assignment.
  • on_revoke (callable) – callback to provide handling of offset commits to a customized store on the start of a rebalance operation.
Raises:

KafkaException

on_assign(consumer, partitions)
on_revoke(consumer, partitions)
Parameters:
  • consumer (Consumer) – Consumer instance.
  • partitions (list(TopicPartition)) – Absolute list of partitions being assigned or revoked.
unassign()

Removes the current partition assignment and stops consuming.

unsubscribe()

Remove current subscription.

Producer

class confluent_kafka.Producer

Asynchronous Kafka Producer

Producer(**kwargs)

Create new Producer instance using provided configuration dict.

len()
Returns:Number of messages and Kafka protocol requests waiting to be delivered to broker.
Return type:int
flush()
Wait for all messages in the Producer queue to be delivered.
This is a convenience method that calls poll() until len() is zero.

Note

See poll() for a description on what callbacks may be triggered.

poll()
poll([timeout])

Polls the producer for events and calls the corresponding callbacks (if registered).

Callbacks:

Parameters:timeout (float) – Maximum time to block waiting for events.
Returns:Number of events processed (callbacks served)
Return type:int
produce()
produce(topic[, value][, key][, partition][, callback])

Produce message to topic. This is an asynchronous operation, an application may use the callback (alias on_delivery) argument to pass a function (or lambda) that will be called from poll() when the message has been successfully delivered or permanently fails delivery.

Parameters:
  • topic (str) – Topic to produce message to
  • value (str|bytes) – Message payload
  • key (str|bytes) – Message key
  • partition (int) – Partition to produce to, elses uses the configured partitioner.
  • on_delivery(err,msg) (func) – Delivery report callback to call (from poll() or flush()) on successful or failed delivery
Return type:

None

Raises:
  • BufferError – if the internal producer message queue is full (queue.buffering.max.messages exceeded)
  • KafkaException – for other errors, see exception code

Message

class confluent_kafka.Message

The Message object represents either a single consumed or produced message, or an event (error() is not None).

An application must check with error() to see if the object is a proper message (error() returns None) or an error/event.

This class is not user-instantiable.

len()
Returns:Message value (payload) size in bytes
Return type:int
error()

The message object is also used to propagate errors and events, an application must check error() to determine if the Message is a proper message (error() returns None) or an error or event (error() returns a KafkaError object)

Return type:None or KafkaError
key()
Returns:message key or None if not available.
Return type:str|bytes or None
offset()
Returns:message offset or None if not available.
Return type:int or None
partition()
Returns:partition number or None if not available.
Return type:int or None
topic()
Returns:topic name or None if not available.
Return type:str or None
value()
Returns:message value (payload) or None if not available.
Return type:str|bytes or None

TopicPartition

class confluent_kafka.TopicPartition

TopicPartition is a generic type to hold a single partition and various information about it.

It is typically used to provide a list of topics or partitions for various operations, such as Consumer.assign().

TopicPartition(topic[, partition][, offset])

Instantiate a TopicPartition object.

Parameters:
  • topic (string) – Topic name
  • partition (int) – Partition id
  • offset (int) – Initial partition offset
Return type:

TopicPartition

error

Indicates an error (with KafkaError) unless None.

offset

Offset

partition

Partition number

topic

Topic name

KafkaError

class confluent_kafka.KafkaError

Kafka error and event object

The KafkaError class serves multiple purposes:

  • Propagation of errors
  • Propagation of events
  • Exceptions

This class is not user-instantiable.

Error and event constants:

Constant Description
_BAD_MSG Local: Bad message format
_BAD_COMPRESSION Local: Invalid compressed data
_DESTROY Local: Broker handle destroyed
_FAIL Local: Communication failure with broker
_TRANSPORT Local: Broker transport failure
_CRIT_SYS_RESOURCE Local: Critical system resource failure
_RESOLVE Local: Host resolution failure
_MSG_TIMED_OUT Local: Message timed out
_PARTITION_EOF Broker: No more messages
_UNKNOWN_PARTITION Local: Unknown partition
_FS Local: File or filesystem error
_UNKNOWN_TOPIC Local: Unknown topic
_ALL_BROKERS_DOWN Local: All broker connections are down
_INVALID_ARG Local: Invalid argument or configuration
_TIMED_OUT Local: Timed out
_QUEUE_FULL Local: Queue full
_ISR_INSUFF Local: ISR count insufficient
_NODE_UPDATE Local: Broker node update
_SSL Local: SSL error
_WAIT_COORD Local: Waiting for coordinator
_UNKNOWN_GROUP Local: Unknown group
_IN_PROGRESS Local: Operation in progress
_PREV_IN_PROGRESS Local: Previous operation in progress
_EXISTING_SUBSCRIPTION Local: Existing subscription
_ASSIGN_PARTITIONS Local: Assign partitions
_REVOKE_PARTITIONS Local: Revoke partitions
_CONFLICT Local: Conflicting use
_STATE Local: Erroneous state
_UNKNOWN_PROTOCOL Local: Unknown protocol
_NOT_IMPLEMENTED Local: Not implemented
_AUTHENTICATION Local: Authentication failure
_NO_OFFSET Local: No offset stored
_OUTDATED Local: Outdated
_TIMED_OUT_QUEUE Local: Timed out in queue
UNKNOWN Unknown broker error
NO_ERROR Success
OFFSET_OUT_OF_RANGE Broker: Offset out of range
INVALID_MSG Broker: Invalid message
UNKNOWN_TOPIC_OR_PART Broker: Unknown topic or partition
INVALID_MSG_SIZE Broker: Invalid message size
LEADER_NOT_AVAILABLE Broker: Leader not available
NOT_LEADER_FOR_PARTITION Broker: Not leader for partition
REQUEST_TIMED_OUT Broker: Request timed out
BROKER_NOT_AVAILABLE Broker: Broker not available
REPLICA_NOT_AVAILABLE Broker: Replica not available
MSG_SIZE_TOO_LARGE Broker: Message size too large
STALE_CTRL_EPOCH Broker: StaleControllerEpochCode
OFFSET_METADATA_TOO_LARGE Broker: Offset metadata string too large
NETWORK_EXCEPTION Broker: Broker disconnected before response received
GROUP_LOAD_IN_PROGRESS Broker: Group coordinator load in progress
GROUP_COORDINATOR_NOT_AVAILABLE Broker: Group coordinator not available
NOT_COORDINATOR_FOR_GROUP Broker: Not coordinator for group
TOPIC_EXCEPTION Broker: Invalid topic
RECORD_LIST_TOO_LARGE Broker: Message batch larger than configured server segment size
NOT_ENOUGH_REPLICAS Broker: Not enough in-sync replicas
NOT_ENOUGH_REPLICAS_AFTER_APPEND Broker: Message(s) written to insufficient number of in-sync replicas
INVALID_REQUIRED_ACKS Broker: Invalid required acks value
ILLEGAL_GENERATION Broker: Specified group generation id is not valid
INCONSISTENT_GROUP_PROTOCOL Broker: Inconsistent group protocol
INVALID_GROUP_ID Broker: Invalid group.id
UNKNOWN_MEMBER_ID Broker: Unknown member
INVALID_SESSION_TIMEOUT Broker: Invalid session timeout
REBALANCE_IN_PROGRESS Broker: Group rebalance in progress
INVALID_COMMIT_OFFSET_SIZE Broker: Commit offset data size is not valid
TOPIC_AUTHORIZATION_FAILED Broker: Topic authorization failed
GROUP_AUTHORIZATION_FAILED Broker: Group authorization failed
CLUSTER_AUTHORIZATION_FAILED Broker: Cluster authorization failed
INVALID_TIMESTAMP Broker: Invalid timestamp
UNSUPPORTED_SASL_MECHANISM Broker: Unsupported SASL mechanism
ILLEGAL_SASL_STATE Broker: Request not valid in current SASL state
UNSUPPORTED_VERSION Broker: API version not supported
code()

Returns the error/event code for comparison toKafkaError.<ERR_CONSTANTS>.

Returns:error/event code
Return type:int
name()

Returns the enum name for error/event.

Returns:error/event enum name string
Return type:str
str()

Returns the human-readable error/event string.

Returns:error/event message string
Return type:str

KafkaException

class confluent_kafka.KafkaException

Kafka exception that wraps the KafkaError class.

Use exception.args[0] to extract the KafkaError object

Configuration

Configuration of producer and consumer instances is performed by providing a dict of configuration properties to the instance constructor, e.g.:

conf = {'bootstrap.servers': 'mybroker.com',
        'group.id': 'mygroup', 'session.timeout.ms': 6000,
        'on_commit': my_commit_callback,
        'default.topic.config': {'auto.offset.reset': 'smallest'}}
consumer = confluent_kafka.Consumer(**conf)

The supported configuration values are dictated by the underlying librdkafka C library. For the full range of configuration properties please consult librdkafka’s documentation: https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md

The Python bindings also provide some additional configuration properties:

  • default.topic.config: value is a dict of topic-level configuration properties that are applied to all used topics for the instance.
  • error_cb(kafka.KafkaError): Callback for generic/global error events. This callback is served by poll().
  • on_delivery(kafka.KafkaError, kafka.Message) (Producer): value is a Python function reference that is called once for each produced message to indicate the final delivery result (success or failure). This property may also be set per-message by passing callback=callable (or on_delivery=callable) to the confluent_kafka.Producer.produce() function.
  • on_commit(kafka.KafkaError, list(kafka.TopicPartition)) (Consumer): Callback used to indicate success or failure of commit requests.

Changelog

  • PR-3 - Add /usr/local/lib to library_dirs in setup
  • PR-4 - Py3: use bytes for Message payload and key
  • PR-5 - Removed hard coded c extentions lib/include paths
  • PR-9 - Use consistent syntax highlighting (e.g. prefix commands with $)
  • PR-17 - Version bump to 0.9.1.2