| |
Methods defined here:
- __bool__(self, /)
- True if self else False
- __init__(self, /, *args, **kwargs)
- Initialize self. See help(type(self)) for accurate signature.
- __len__(self, /)
- Return len(self).
- abort_transaction(...)
- .. py:function:: abort_transaction([timeout])
Aborts the current transaction.
This function should also be used to recover from non-fatal
abortable transaction errors when KafkaError.txn_requires_abort()
is True.
Any outstanding messages will be purged and fail with
_PURGE_INFLIGHT or _PURGE_QUEUE.
Note: This function will block until all outstanding messages
are purged and the transaction abort request has been
successfully handled by the transaction coordinator, or until
the timeout expires, which ever comes first. On timeout the
application may call the function again.
Note: Will automatically call purge() and flush() to ensure
all queued and in-flight messages are purged before attempting
to abort the transaction.
:param float timeout: The maximum amount of time to block
waiting for transaction to abort in seconds.
:raises: KafkaError: Use exc.args[0].retriable() to check if the
operation may be retried.
Treat any other error as a fatal error.
- begin_transaction(...)
- .. py:function:: begin_transaction()
Begin a new transaction.
init_transactions() must have been called successfully (once)
before this function is called.
Any messages produced or offsets sent to a transaction, after
the successful return of this function will be part of the
transaction and committed or aborted atomically.
Complete the transaction by calling commit_transaction() or
Abort the transaction by calling abort_transaction().
:raises: KafkaError: Use exc.args[0].retriable() to check if the
operation may be retried, else treat the
error as a fatal error.
- commit_transaction(...)
- .. py:function:: commit_transaction([timeout])
Commmit the current transaction.
Any outstanding messages will be flushed (delivered) before
actually committing the transaction.
If any of the outstanding messages fail permanently the current
transaction will enter the abortable error state and this
function will return an abortable error, in this case the
application must call abort_transaction() before attempting
a new transaction with begin_transaction().
Note: This function will block until all outstanding messages
are delivered and the transaction commit request has been
successfully handled by the transaction coordinator, or until
the timeout expires, which ever comes first. On timeout the
application may call the function again.
Note: Will automatically call flush() to ensure all queued
messages are delivered before attempting to commit the
transaction. Delivery reports and other callbacks may thus be
triggered from this method.
:param float timeout: The amount of time to block in seconds.
:raises: KafkaError: Use exc.args[0].retriable() to check if the
operation may be retried, or
exc.args[0].txn_requires_abort() if the current
transaction has failed and must be aborted by calling
abort_transaction() and then start a new transaction
with begin_transaction().
Treat any other error as a fatal error.
- flush(...)
- .. py:function:: flush([timeout])
Wait for all messages in the Producer queue to be delivered.
This is a convenience method that calls :py:func:`poll()` until :py:func:`len()` is zero or the optional timeout elapses.
:param: float timeout: Maximum time to block (requires librdkafka >= v0.9.4). (Seconds)
:returns: Number of messages still in queue.
.. note:: See :py:func:`poll()` for a description on what callbacks may be triggered.
- init_transactions(...)
- .. py:function: init_transactions([timeout])
Initializes transactions for the producer instance.
This function ensures any transactions initiated by previous
instances of the producer with the same `transactional.id` are
completed.
If the previous instance failed with a transaction in progress
the previous transaction will be aborted.
This function needs to be called before any other transactional
or produce functions are called when the `transactional.id` is
configured.
If the last transaction had begun completion (following
transaction commit) but not yet finished, this function will
await the previous transaction's completion.
When any previous transactions have been fenced this function
will acquire the internal producer id and epoch, used in all
future transactional messages issued by this producer instance.
Upon successful return from this function the application has to
perform at least one of the following operations within
`transaction.timeout.ms` to avoid timing out the transaction
on the broker:
* produce() (et.al)
* send_offsets_to_transaction()
* commit_transaction()
* abort_transaction()
:param float timeout: Maximum time to block in seconds.
:raises: KafkaError: Use exc.args[0].retriable() to check if the
operation may be retried, else treat the
error as a fatal error.
- list_topics(...)
- .. py:function:: list_topics([topic=None], [timeout=-1])
Request metadata from the cluster.
This method provides the same information as listTopics(), describeTopics() and describeCluster() in the Java Admin client.
:param str topic: If specified, only request information about this topic, else return results for all topics in cluster. Warning: If auto.create.topics.enable is set to true on the broker and an unknown topic is specified, it will be created.
:param float timeout: The maximum response time before timing out, or -1 for infinite timeout.
:rtype: ClusterMetadata
:raises: KafkaException
- poll(...)
- .. py:function:: poll([timeout])
Polls the producer for events and calls the corresponding callbacks (if registered).
Callbacks:
- ``on_delivery`` callbacks from :py:func:`produce()`
- ...
:param float timeout: Maximum time to block waiting for events. (Seconds)
:returns: Number of events processed (callbacks served)
:rtype: int
- produce(...)
- .. py:function:: produce(topic, [value], [key], [partition], [on_delivery], [timestamp], [headers])
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 :py:func:`poll()` when the message has been successfully delivered or permanently fails delivery.
Currently message headers are not supported on the message returned to the callback. The ``msg.headers()`` will return None even if the original message had headers set.
:param str topic: Topic to produce message to
:param str|bytes value: Message payload
:param str|bytes key: Message key
:param int partition: Partition to produce to, else uses the configured built-in partitioner.
:param func on_delivery(err,msg): Delivery report callback to call (from :py:func:`poll()` or :py:func:`flush()`) on successful or failed delivery
:param int timestamp: Message timestamp (CreateTime) in milliseconds since epoch UTC (requires librdkafka >= v0.9.4, api.version.request=true, and broker >= 0.10.0.0). Default value is current time.
:param dict|list headers: Message headers to set on the message. The header key must be a string while the value must be binary, unicode or None. Accepts a list of (key,value) or a dict. (Requires librdkafka >= v0.11.4 and broker version >= 0.11.0.0)
:rtype: None
:raises BufferError: if the internal producer message queue is full (``queue.buffering.max.messages`` exceeded)
:raises KafkaException: for other errors, see exception code
:raises NotImplementedError: if timestamp is specified without underlying library support.
- purge(...)
- .. py:function:: purge([in_queue=True], [in_flight=True], [blocking=True])
Purge messages currently handled by the producer instance.
The application will need to call poll() or flush() afterwards to serve the delivery report callbacks of the purged messages.
:param: bool in_queue: Purge messages from internal queues. By default, true.
:param: bool in_flight: Purge messages in flight to or from the broker. By default, true.
:param: bool blocking: If set to False, will not wait on background thread queue purging to finish. By default, true.
- send_offsets_to_transaction(...)
- .. py:function:: send_offsets_to_transaction(positions, group_metadata, [timeout])
Sends a list of topic partition offsets to the consumer group
coordinator for group_metadata and marks the offsets as part
of the current transaction.
These offsets will be considered committed only if the
transaction is committed successfully.
The offsets should be the next message your application will
consume, i.e., the last processed message's offset + 1 for each
partition.
Either track the offsets manually during processing or use
consumer.position() (on the consumer) to get the current offsets
for the partitions assigned to the consumer.
Use this method at the end of a consume-transform-produce loop
prior to committing the transaction with commit_transaction().
Note: The consumer must disable auto commits
(set `enable.auto.commit` to false on the consumer).
Note: Logical and invalid offsets (e.g., OFFSET_INVALID) in
offsets will be ignored. If there are no valid offsets in
offsets the function will return successfully and no action
will be taken.
:param list(TopicPartition) offsets: current consumer/processing
position(offsets) for the
list of partitions.
:param object group_metadata: consumer group metadata retrieved
from the input consumer's
get_consumer_group_metadata().
:param float timeout: Amount of time to block in seconds.
:raises: KafkaError: Use exc.args[0].retriable() to check if the
operation may be retried, or
exc.args[0].txn_requires_abort() if the current
transaction has failed and must be aborted by calling
abort_transaction() and then start a new transaction
with begin_transaction().
Treat any other error as a fatal error.
- set_sasl_credentials(...)
- .. py:function:: set_sasl_credentials(username, password)
Sets the SASL credentials used for this client.
These credentials will overwrite the old ones, and will be used the next time the client needs to authenticate.
This method will not disconnect existing broker connections that have been established with the old credentials.
This method is applicable only to SASL PLAIN and SCRAM mechanisms.
Static methods defined here:
- __new__(*args, **kwargs) from builtins.type
- Create and return a new object. See help(type) for accurate signature.
|