Apache Kafka C/C++ consumer and producer client library. More...

Data Structures
struct	rd_kafka_err_desc
	Error code value, name and description. Typically for use with language bindings to automatically expose the full set of librdkafka error codes. More...

struct	rd_kafka_topic_partition_s
	Topic+Partition place holder. More...

struct	rd_kafka_topic_partition_list_s
	A growable list of Topic+Partitions. More...

struct	rd_kafka_vu_s
	VTYPE + argument container for use with rd_kafka_produce_va() More...

struct	rd_kafka_message_s
	A Kafka message as returned by the `rd_kafka_consume*`() family of functions as well as provided to the Producer `dr_msg_cb()`. More...

struct	rd_kafka_metadata_broker
	Broker information. More...

struct	rd_kafka_metadata_partition
	Partition information. More...

struct	rd_kafka_metadata_topic
	Topic information. More...

struct	rd_kafka_metadata
	Metadata container. More...

struct	rd_kafka_group_member_info
	Group member information. More...

struct	rd_kafka_group_info
	Group information. More...

struct	rd_kafka_group_list
	List of groups. More...

Macros
#define	RD_KAFKA_RESP_ERR_GROUP_LOAD_IN_PROGRESS RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS

#define	RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE

#define	RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP RD_KAFKA_RESP_ERR_NOT_COORDINATOR

Typedefs
typedef struct rd_kafka_DeleteAcls_result_response_s	rd_kafka_DeleteAcls_result_response_t

Plugin interface
A plugin interface that allows external runtime-loaded libraries to integrate with a client instance without modifications to the application code. Plugins are loaded when referenced through the `plugin.library.paths` configuration property and operates on the `rd_kafka_conf_t` object prior `rd_kafka_t` instance creation. Warning Plugins require the application to link librdkafka dynamically and not statically. Failure to do so will lead to missing symbols or finding symbols in another librdkafka library than the application was linked with.
typedef rd_kafka_resp_err_t()	rd_kafka_plugin_f_conf_init_t(rd_kafka_conf_t conf, void plug_opaquep, char errstr, size_t errstr_size)
	Plugin's configuration initializer method called each time the library is referenced from configuration (even if previously loaded by another client instance). More...

Functions
RD_EXPORT rd_kafka_AclBinding_t *	rd_kafka_AclBinding_new (rd_kafka_ResourceType_t restype, const char name, rd_kafka_ResourcePatternType_t resource_pattern_type, const char principal, const char host, rd_kafka_AclOperation_t operation, rd_kafka_AclPermissionType_t permission_type, char errstr, size_t errstr_size)
	Create a new AclBinding object. This object is later passed to rd_kafka_CreateAcls(). More...

RD_EXPORT rd_kafka_AclBindingFilter_t *	rd_kafka_AclBindingFilter_new (rd_kafka_ResourceType_t restype, const char name, rd_kafka_ResourcePatternType_t resource_pattern_type, const char principal, const char host, rd_kafka_AclOperation_t operation, rd_kafka_AclPermissionType_t permission_type, char errstr, size_t errstr_size)
	Create a new AclBindingFilter object. This object is later passed to rd_kafka_DescribeAcls() or rd_kafka_DeletesAcls() in order to filter the acls to retrieve or to delete. Use the same rd_kafka_AclBinding functions to query or destroy it. More...

RD_EXPORT rd_kafka_ResourceType_t	rd_kafka_AclBinding_restype (const rd_kafka_AclBinding_t *acl)

const RD_EXPORT char *	rd_kafka_AclBinding_name (const rd_kafka_AclBinding_t *acl)

const RD_EXPORT char *	rd_kafka_AclBinding_principal (const rd_kafka_AclBinding_t *acl)

const RD_EXPORT char *	rd_kafka_AclBinding_host (const rd_kafka_AclBinding_t *acl)

RD_EXPORT rd_kafka_AclOperation_t	rd_kafka_AclBinding_operation (const rd_kafka_AclBinding_t *acl)

RD_EXPORT rd_kafka_AclPermissionType_t	rd_kafka_AclBinding_permission_type (const rd_kafka_AclBinding_t *acl)

RD_EXPORT rd_kafka_ResourcePatternType_t	rd_kafka_AclBinding_resource_pattern_type (const rd_kafka_AclBinding_t *acl)

const RD_EXPORT rd_kafka_error_t *	rd_kafka_AclBinding_error (const rd_kafka_AclBinding_t *acl)

RD_EXPORT void	rd_kafka_AclBinding_destroy (rd_kafka_AclBinding_t *acl_binding)
	Destroy and free an AclBinding object previously created with rd_kafka_AclBinding_new()

RD_EXPORT void	rd_kafka_AclBinding_destroy_array (rd_kafka_AclBinding_t **acl_bindings, size_t acl_bindings_cnt)
	Helper function to destroy all AclBinding objects in the `acl_bindings` array (of `acl_bindings_cnt` elements). The array itself is not freed.

const RD_EXPORT rd_kafka_acl_result_t **	rd_kafka_CreateAcls_result_acls (const rd_kafka_CreateAcls_result_t result, size_t cntp)
	Get an array of acl results from a CreateAcls result. More...

RD_EXPORT void	rd_kafka_CreateAcls (rd_kafka_t rk, rd_kafka_AclBinding_t new_acls, size_t new_acls_cnt, const rd_kafka_AdminOptions_t options, rd_kafka_queue_t *rkqu)
	Create acls as specified by the `new_acls` array of size `new_topic_cnt` elements. More...

const RD_EXPORT rd_kafka_AclBinding_t **	rd_kafka_DescribeAcls_result_acls (const rd_kafka_DescribeAcls_result_t result, size_t cntp)
	Get an array of resource results from a DescribeAcls result. More...

RD_EXPORT void	rd_kafka_DescribeAcls (rd_kafka_t rk, rd_kafka_AclBindingFilter_t acl_filter, const rd_kafka_AdminOptions_t options, rd_kafka_queue_t rkqu)
	Describe acls matching the filter provided in `acl_filter`. More...

const RD_EXPORT rd_kafka_DeleteAcls_result_response_t **	rd_kafka_DeleteAcls_result_responses (const rd_kafka_DeleteAcls_result_t result, size_t cntp)
	Get an array of DeleteAcls result responses from a DeleteAcls result. More...

const RD_EXPORT rd_kafka_error_t *	rd_kafka_DeleteAcls_result_response_error (const rd_kafka_DeleteAcls_result_response_t *result_response)

const RD_EXPORT rd_kafka_AclBinding_t **	rd_kafka_DeleteAcls_result_response_matching_acls (const rd_kafka_DeleteAcls_result_response_t result_response, size_t matching_acls_cntp)

RD_EXPORT void	rd_kafka_DeleteAcls (rd_kafka_t rk, rd_kafka_AclBindingFilter_t del_acls, size_t del_acls_cnt, const rd_kafka_AdminOptions_t options, rd_kafka_queue_t *rkqu)
	Delete acls matching the filteres provided in `del_acls` array of size `del_acls_cnt`. More...

Message headers
Message headers consist of a list of (string key, binary value) pairs. Duplicate keys are supported and the order in which keys were added are retained. Header values are considered binary and may have three types of value: proper value with size > 0 and a valid pointer empty value with size = 0 and any non-NULL pointer null value with size = 0 and a NULL pointer Headers require Apache Kafka broker version v0.11.0.0 or later. Header operations are O(n).
RD_EXPORT rd_kafka_headers_t *	rd_kafka_headers_new (size_t initial_count)
	Create a new headers list. More...

RD_EXPORT void	rd_kafka_headers_destroy (rd_kafka_headers_t *hdrs)
	Destroy the headers list. The object and any returned value pointers are not usable after this call.

RD_EXPORT rd_kafka_headers_t *	rd_kafka_headers_copy (const rd_kafka_headers_t *src)
	Make a copy of headers list `src`.

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_header_add (rd_kafka_headers_t hdrs, const char name, ssize_t name_size, const void *value, ssize_t value_size)
	Add header with name `name` and value `val` (copied) of size `size` (not including null-terminator). More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_header_remove (rd_kafka_headers_t hdrs, const char name)
	Remove all headers for the given key (if any). More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_header_get_last (const rd_kafka_headers_t hdrs, const char name, const void *valuep, size_t sizep)
	Find last header in list `hdrs` matching `name`. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_header_get (const rd_kafka_headers_t hdrs, size_t idx, const char name, const void *valuep, size_t sizep)
	Iterator for headers matching `name`. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_header_get_all (const rd_kafka_headers_t hdrs, size_t idx, const char namep, const void valuep, size_t sizep)
	Iterator for all headers. More...

Topic configuration
Topic configuration property interface
RD_EXPORT rd_kafka_topic_conf_t *	rd_kafka_topic_conf_new (void)
	Create topic configuration object. More...

RD_EXPORT rd_kafka_topic_conf_t *	rd_kafka_topic_conf_dup (const rd_kafka_topic_conf_t *conf)
	Creates a copy/duplicate of topic configuration object `conf`.

RD_EXPORT rd_kafka_topic_conf_t *	rd_kafka_default_topic_conf_dup (rd_kafka_t *rk)
	Creates a copy/duplicate of `rk` 's default topic configuration object.

RD_EXPORT void	rd_kafka_topic_conf_destroy (rd_kafka_topic_conf_t *topic_conf)
	Destroys a topic conf object.

RD_EXPORT rd_kafka_conf_res_t	rd_kafka_topic_conf_set (rd_kafka_topic_conf_t conf, const char name, const char value, char errstr, size_t errstr_size)
	Sets a single rd_kafka_topic_conf_t value by property name. More...

RD_EXPORT void	rd_kafka_topic_conf_set_opaque (rd_kafka_topic_conf_t conf, void rkt_opaque)
	Sets the application's opaque pointer that will be passed to all topic callbacks as the `rkt_opaque` argument. More...

RD_EXPORT void	rd_kafka_topic_conf_set_partitioner_cb (rd_kafka_topic_conf_t topic_conf, int32_t(partitioner)(const rd_kafka_topic_t rkt, const void keydata, size_t keylen, int32_t partition_cnt, void rkt_opaque, void msg_opaque))
	Producer: Set partitioner callback in provided topic conf object. More...

RD_EXPORT void	rd_kafka_topic_conf_set_msg_order_cmp (rd_kafka_topic_conf_t topic_conf, int(msg_order_cmp)(const rd_kafka_message_t a, const rd_kafka_message_t b))
	Producer: Set message queueing order comparator callback. More...

RD_EXPORT int	rd_kafka_topic_partition_available (const rd_kafka_topic_t *rkt, int32_t partition)
	Check if partition is available (has a leader broker). More...

RD_EXPORT int32_t	rd_kafka_msg_partitioner_random (const rd_kafka_topic_t rkt, const void key, size_t keylen, int32_t partition_cnt, void rkt_opaque, void msg_opaque)
	Random partitioner. More...

RD_EXPORT int32_t	rd_kafka_msg_partitioner_consistent (const rd_kafka_topic_t rkt, const void key, size_t keylen, int32_t partition_cnt, void rkt_opaque, void msg_opaque)
	Consistent partitioner. More...

RD_EXPORT int32_t	rd_kafka_msg_partitioner_consistent_random (const rd_kafka_topic_t rkt, const void key, size_t keylen, int32_t partition_cnt, void rkt_opaque, void msg_opaque)
	Consistent-Random partitioner. More...

RD_EXPORT int32_t	rd_kafka_msg_partitioner_murmur2 (const rd_kafka_topic_t rkt, const void key, size_t keylen, int32_t partition_cnt, void rkt_opaque, void msg_opaque)
	Murmur2 partitioner (Java compatible). More...

RD_EXPORT int32_t	rd_kafka_msg_partitioner_murmur2_random (const rd_kafka_topic_t rkt, const void key, size_t keylen, int32_t partition_cnt, void rkt_opaque, void msg_opaque)
	Consistent-Random Murmur2 partitioner (Java compatible). More...

RD_EXPORT int32_t	rd_kafka_msg_partitioner_fnv1a (const rd_kafka_topic_t rkt, const void key, size_t keylen, int32_t partition_cnt, void rkt_opaque, void msg_opaque)
	FNV-1a partitioner. More...

RD_EXPORT int32_t	rd_kafka_msg_partitioner_fnv1a_random (const rd_kafka_topic_t rkt, const void key, size_t keylen, int32_t partition_cnt, void rkt_opaque, void msg_opaque)
	Consistent-Random FNV-1a partitioner. More...

Queue API
Message queues allows the application to re-route consumed messages from multiple topic+partitions into one single queue point. This queue point containing messages from a number of topic+partitions may then be served by a single rd_kafka_consume*_queue() call, rather than one call per topic+partition combination.
RD_EXPORT rd_kafka_queue_t *	rd_kafka_queue_new (rd_kafka_t *rk)
	Create a new message queue. More...

RD_EXPORT void	rd_kafka_queue_destroy (rd_kafka_queue_t *rkqu)

RD_EXPORT rd_kafka_queue_t *	rd_kafka_queue_get_main (rd_kafka_t *rk)

RD_EXPORT rd_kafka_queue_t *	rd_kafka_queue_get_sasl (rd_kafka_t *rk)

RD_EXPORT rd_kafka_error_t *	rd_kafka_sasl_background_callbacks_enable (rd_kafka_t *rk)
	Enable SASL OAUTHBEARER refresh callbacks on the librdkafka background thread. More...

RD_EXPORT rd_kafka_queue_t *	rd_kafka_queue_get_consumer (rd_kafka_t *rk)

RD_EXPORT rd_kafka_queue_t *	rd_kafka_queue_get_partition (rd_kafka_t rk, const char topic, int32_t partition)

RD_EXPORT rd_kafka_queue_t *	rd_kafka_queue_get_background (rd_kafka_t *rk)

RD_EXPORT void	rd_kafka_queue_forward (rd_kafka_queue_t src, rd_kafka_queue_t dst)
	Forward/re-route queue `src` to `dst`. If `dst` is `NULL` the forwarding is removed. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_set_log_queue (rd_kafka_t rk, rd_kafka_queue_t rkqu)
	Forward librdkafka logs (and debug) to the specified queue for serving with one of the ..poll() calls. More...

RD_EXPORT size_t	rd_kafka_queue_length (rd_kafka_queue_t *rkqu)

RD_EXPORT void	rd_kafka_queue_io_event_enable (rd_kafka_queue_t rkqu, int fd, const void payload, size_t size)
	Enable IO event triggering for queue. More...

RD_EXPORT void	rd_kafka_queue_cb_event_enable (rd_kafka_queue_t rkqu, void(event_cb)(rd_kafka_t rk, void qev_opaque), void *qev_opaque)
	Enable callback event triggering for queue. More...

RD_EXPORT void	rd_kafka_queue_yield (rd_kafka_queue_t *rkqu)
	Cancels the current rd_kafka_queue_poll() on `rkqu`. More...

Simple Consumer API (legacy): Queue consumers
The following `..._queue()` functions are analogue to the functions above but reads messages from the provided queue `rkqu` instead. `rkqu` must have been previously created with `rd_kafka_queue_new()` and the topic consumer must have been started with `rd_kafka_consume_start_queue()` utilising the the same queue.
RD_EXPORT rd_kafka_message_t *	rd_kafka_consume_queue (rd_kafka_queue_t *rkqu, int timeout_ms)
	Consume from queue. More...

RD_EXPORT ssize_t	rd_kafka_consume_batch_queue (rd_kafka_queue_t rkqu, int timeout_ms, rd_kafka_message_t *rkmessages, size_t rkmessages_size)
	Consume batch of messages from queue. More...

RD_EXPORT int	rd_kafka_consume_callback_queue (rd_kafka_queue_t rkqu, int timeout_ms, void(consume_cb)(rd_kafka_message_t rkmessage, void commit_opaque), void *commit_opaque)
	Consume multiple messages from queue with callback. More...

Simple Consumer API (legacy): Topic+partition offset store.
If `auto.commit.enable` is true the offset is stored automatically prior to returning of the message(s) in each of the rd_kafka_consume*() functions above.
RD_EXPORT rd_kafka_resp_err_t	rd_kafka_offset_store (rd_kafka_topic_t *rkt, int32_t partition, int64_t offset)
	Store offset `offset` + 1 for topic `rkt` partition `partition`. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_offsets_store (rd_kafka_t rk, rd_kafka_topic_partition_list_t offsets)
	Store offsets for next auto-commit for one or more partitions. More...

KafkaConsumer (C)
High-level KafkaConsumer C API
RD_EXPORT rd_kafka_resp_err_t	rd_kafka_subscribe (rd_kafka_t rk, const rd_kafka_topic_partition_list_t topics)
	Subscribe to topic set using balanced consumer groups. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_unsubscribe (rd_kafka_t *rk)
	Unsubscribe from the current subscription set.

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_subscription (rd_kafka_t rk, rd_kafka_topic_partition_list_t *topics)
	Returns the current topic subscription. More...

RD_EXPORT rd_kafka_message_t *	rd_kafka_consumer_poll (rd_kafka_t *rk, int timeout_ms)
	Poll the consumer for messages or events. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_consumer_close (rd_kafka_t *rk)
	Close the consumer. More...

RD_EXPORT rd_kafka_error_t *	rd_kafka_consumer_close_queue (rd_kafka_t rk, rd_kafka_queue_t rkqu)
	Asynchronously close the consumer. More...

RD_EXPORT int	rd_kafka_consumer_closed (rd_kafka_t *rk)

RD_EXPORT rd_kafka_error_t *	rd_kafka_incremental_assign (rd_kafka_t rk, const rd_kafka_topic_partition_list_t partitions)
	Incrementally add `partitions` to the current assignment. More...

RD_EXPORT rd_kafka_error_t *	rd_kafka_incremental_unassign (rd_kafka_t rk, const rd_kafka_topic_partition_list_t partitions)
	Incrementally remove `partitions` from the current assignment. More...

const RD_EXPORT char *	rd_kafka_rebalance_protocol (rd_kafka_t *rk)
	The rebalance protocol currently in use. This will be "NONE" if the consumer has not (yet) joined a group, else it will match the rebalance protocol ("EAGER", "COOPERATIVE") of the configured and selected assignor(s). All configured assignors must have the same protocol type, meaning online migration of a consumer group from using one protocol to another (in particular upgading from EAGER to COOPERATIVE) without a restart is not currently supported. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_assign (rd_kafka_t rk, const rd_kafka_topic_partition_list_t partitions)
	Atomic assignment of partitions to consume. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_assignment (rd_kafka_t rk, rd_kafka_topic_partition_list_t *partitions)
	Returns the current partition assignment as set by rd_kafka_assign() or rd_kafka_incremental_assign(). More...

RD_EXPORT int	rd_kafka_assignment_lost (rd_kafka_t *rk)
	Check whether the consumer considers the current assignment to have been lost involuntarily. This method is only applicable for use with a high level subscribing consumer. Assignments are revoked immediately when determined to have been lost, so this method is only useful when reacting to a RD_KAFKA_EVENT_REBALANCE event or from within a rebalance_cb. Partitions that have been lost may already be owned by other members in the group and therefore commiting offsets, for example, may fail. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_commit (rd_kafka_t rk, const rd_kafka_topic_partition_list_t offsets, int async)
	Commit offsets on broker for the provided list of partitions. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_commit_message (rd_kafka_t rk, const rd_kafka_message_t rkmessage, int async)
	Commit message's offset on broker for the message's partition. The committed offset is the message's offset + 1. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_commit_queue (rd_kafka_t rk, const rd_kafka_topic_partition_list_t offsets, rd_kafka_queue_t rkqu, void(cb)(rd_kafka_t rk, rd_kafka_resp_err_t err, rd_kafka_topic_partition_list_t offsets, void commit_opaque), void commit_opaque)
	Commit offsets on broker for the provided list of partitions. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_committed (rd_kafka_t rk, rd_kafka_topic_partition_list_t partitions, int timeout_ms)
	Retrieve committed offsets for topics+partitions. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_position (rd_kafka_t rk, rd_kafka_topic_partition_list_t partitions)
	Retrieve current positions (offsets) for topics+partitions. More...

RD_EXPORT rd_kafka_consumer_group_metadata_t *	rd_kafka_consumer_group_metadata (rd_kafka_t *rk)

RD_EXPORT rd_kafka_consumer_group_metadata_t *	rd_kafka_consumer_group_metadata_new (const char *group_id)
	Create a new consumer group metadata object. This is typically only used for writing tests. More...

RD_EXPORT rd_kafka_consumer_group_metadata_t *	rd_kafka_consumer_group_metadata_new_with_genid (const char group_id, int32_t generation_id, const char member_id, const char *group_instance_id)
	Create a new consumer group metadata object. This is typically only used for writing tests. More...

RD_EXPORT void	rd_kafka_consumer_group_metadata_destroy (rd_kafka_consumer_group_metadata_t *)
	Frees the consumer group metadata object as returned by rd_kafka_consumer_group_metadata().

RD_EXPORT rd_kafka_error_t *	rd_kafka_consumer_group_metadata_write (const rd_kafka_consumer_group_metadata_t cgmd, void bufferp, size_t sizep)
	Serialize the consumer group metadata to a binary format. This is mainly for client binding use and not for application use. More...

RD_EXPORT rd_kafka_error_t *	rd_kafka_consumer_group_metadata_read (rd_kafka_consumer_group_metadata_t *cgmdp, const void buffer, size_t size)
	Reads serialized consumer group metadata and returns a consumer group metadata object. This is mainly for client binding use and not for application use. More...

Metadata API
RD_EXPORT rd_kafka_resp_err_t	rd_kafka_metadata (rd_kafka_t rk, int all_topics, rd_kafka_topic_t only_rkt, const struct rd_kafka_metadata **metadatap, int timeout_ms)
	Request Metadata from broker. More...

RD_EXPORT void	rd_kafka_metadata_destroy (const struct rd_kafka_metadata *metadata)
	Release metadata memory.

Client group information
RD_EXPORT rd_kafka_resp_err_t	rd_kafka_list_groups (rd_kafka_t rk, const char group, const struct rd_kafka_group_list **grplistp, int timeout_ms)
	List and describe client groups in cluster. More...

RD_EXPORT void	rd_kafka_group_list_destroy (const struct rd_kafka_group_list *grplist)
	Release list memory.

Experimental APIs
RD_EXPORT rd_kafka_resp_err_t	rd_kafka_poll_set_consumer (rd_kafka_t *rk)
	Redirect the main (rd_kafka_poll()) queue to the KafkaConsumer's queue (rd_kafka_consumer_poll()). More...

Auxiliary types
RD_EXPORT rd_kafka_resp_err_t	rd_kafka_topic_result_error (const rd_kafka_topic_result_t *topicres)
	Topic result provides per-topic operation result information. More...

const RD_EXPORT char *	rd_kafka_topic_result_error_string (const rd_kafka_topic_result_t *topicres)

const RD_EXPORT char *	rd_kafka_topic_result_name (const rd_kafka_topic_result_t *topicres)

const RD_EXPORT rd_kafka_error_t *	rd_kafka_group_result_error (const rd_kafka_group_result_t *groupres)
	Group result provides per-group operation result information. More...

const RD_EXPORT char *	rd_kafka_group_result_name (const rd_kafka_group_result_t *groupres)

const RD_EXPORT rd_kafka_topic_partition_list_t *	rd_kafka_group_result_partitions (const rd_kafka_group_result_t *groupres)

Security APIs
RD_EXPORT rd_kafka_resp_err_t	rd_kafka_oauthbearer_set_token (rd_kafka_t rk, const char token_value, int64_t md_lifetime_ms, const char md_principal_name, const char extensions, size_t extension_size, char errstr, size_t errstr_size)
	Set SASL/OAUTHBEARER token and metadata. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_oauthbearer_set_token_failure (rd_kafka_t rk, const char errstr)
	SASL/OAUTHBEARER token refresh failure indicator. More...

Transactional producer API
The transactional producer operates on top of the idempotent producer, and provides full exactly-once semantics (EOS) for Apache Kafka when used with the transaction aware consumer (`isolation.level=read_committed`). A producer instance is configured for transactions by setting the `transactional.id` to an identifier unique for the application. This id will be used to fence stale transactions from previous instances of the application, typically following an outage or crash. After creating the transactional producer instance using rd_kafka_new() the transactional state must be initialized by calling rd_kafka_init_transactions(). This is a blocking call that will acquire a runtime producer id from the transaction coordinator broker as well as abort any stale transactions and fence any still running producer instances with the same `transactional.id`. Once transactions are initialized the application may begin a new transaction by calling rd_kafka_begin_transaction(). A producer instance may only have one single on-going transaction. Any messages produced after the transaction has been started will belong to the ongoing transaction and will be committed or aborted atomically. It is not permitted to produce messages outside a transaction boundary, e.g., before rd_kafka_begin_transaction() or after rd_kafka_commit_transaction(), rd_kafka_abort_transaction(), or after the current transaction has failed. If consumed messages are used as input to the transaction, the consumer instance must be configured with `enable.auto.commit` set to `false`. To commit the consumed offsets along with the transaction pass the list of consumed partitions and the last offset processed + 1 to rd_kafka_send_offsets_to_transaction() prior to committing the transaction. This allows an aborted transaction to be restarted using the previously committed offsets. To commit the produced messages, and any consumed offsets, to the current transaction, call rd_kafka_commit_transaction(). This call will block until the transaction has been fully committed or failed (typically due to fencing by a newer producer instance). Alternatively, if processing fails, or an abortable transaction error is raised, the transaction needs to be aborted by calling rd_kafka_abort_transaction() which marks any produced messages and offset commits as aborted. After the current transaction has been committed or aborted a new transaction may be started by calling rd_kafka_begin_transaction() again. Retriable errors Some error cases allow the attempted operation to be retried, this is indicated by the error object having the retriable flag set which can be detected by calling rd_kafka_error_is_retriable(). When this flag is set the application may retry the operation immediately or preferably after a shorter grace period (to avoid busy-looping). Retriable errors include timeouts, broker transport failures, etc. Abortable errors An ongoing transaction may fail permanently due to various errors, such as transaction coordinator becoming unavailable, write failures to the Apache Kafka log, under-replicated partitions, etc. At this point the producer application must abort the current transaction using rd_kafka_abort_transaction() and optionally start a new transaction by calling rd_kafka_begin_transaction(). Whether an error is abortable or not is detected by calling rd_kafka_error_txn_requires_abort() on the returned error object. Fatal errors While the underlying idempotent producer will typically only raise fatal errors for unrecoverable cluster errors where the idempotency guarantees can't be maintained, most of these are treated as abortable by the transactional producer since transactions may be aborted and retried in their entirety; The transactional producer on the other hand introduces a set of additional fatal errors which the application needs to handle by shutting down the producer and terminate. There is no way for a producer instance to recover from fatal errors. Whether an error is fatal or not is detected by calling rd_kafka_error_is_fatal() on the returned error object or by checking the global rd_kafka_fatal_error() code. Fatal errors are raised by triggering the `error_cb` (see the Fatal error chapter in INTRODUCTION.md for more information), and any sub-sequent transactional API calls will return RD_KAFKA_RESP_ERR__FATAL or have the fatal flag set (see rd_kafka_error_is_fatal()). The originating fatal error code can be retrieved by calling rd_kafka_fatal_error(). Handling of other errors For errors that have neither retriable, abortable or the fatal flag set it is not always obvious how to handle them. While some of these errors may be indicative of bugs in the application code, such as when an invalid parameter is passed to a method, other errors might originate from the broker and be passed thru as-is to the application. The general recommendation is to treat these errors, that have neither the retriable or abortable flags set, as fatal. Error handling example retry: rd_kafka_error_t error; error = rd_kafka_commit_transaction(producer, 101000); if (!error) return success; else if (rd_kafka_error_txn_requires_abort(error)) { do_abort_transaction_and_reset_inputs(); } else if (rd_kafka_error_is_retriable(error)) { rd_kafka_error_destroy(error); goto retry; } else { // treat all other errors as fatal errors fatal_error(rd_kafka_error_string(error)); } rd_kafka_error_destroy(error);
RD_EXPORT rd_kafka_error_t *	rd_kafka_init_transactions (rd_kafka_t *rk, int timeout_ms)
	Initialize transactions for the producer instance. More...

RD_EXPORT rd_kafka_error_t *	rd_kafka_begin_transaction (rd_kafka_t *rk)
	Begin a new transaction. More...

RD_EXPORT rd_kafka_error_t *	rd_kafka_send_offsets_to_transaction (rd_kafka_t rk, const rd_kafka_topic_partition_list_t offsets, const rd_kafka_consumer_group_metadata_t *cgmetadata, int timeout_ms)
	Sends a list of topic partition offsets to the consumer group coordinator for `cgmetadata`, and marks the offsets as part part of the current transaction. These offsets will be considered committed only if the transaction is committed successfully. More...

RD_EXPORT rd_kafka_error_t *	rd_kafka_commit_transaction (rd_kafka_t *rk, int timeout_ms)
	Commit the current transaction (as started with rd_kafka_begin_transaction()). More...

RD_EXPORT rd_kafka_error_t *	rd_kafka_abort_transaction (rd_kafka_t *rk, int timeout_ms)
	Aborts the ongoing transaction. More...

librdkafka version
#define	RD_KAFKA_VERSION 0x010900ff
	librdkafka version More...

RD_EXPORT int	rd_kafka_version (void)
	Returns the librdkafka version as integer. More...

const RD_EXPORT char *	rd_kafka_version_str (void)
	Returns the librdkafka version as string. More...

Constants, errors, types
#define	RD_KAFKA_DEBUG_CONTEXTS
	Supported debug contexts. (compile time) More...

enum	rd_kafka_type_t { RD_KAFKA_PRODUCER, RD_KAFKA_CONSUMER }
	rd_kafka_t handle type. More...

enum	rd_kafka_timestamp_type_t { RD_KAFKA_TIMESTAMP_NOT_AVAILABLE, RD_KAFKA_TIMESTAMP_CREATE_TIME, RD_KAFKA_TIMESTAMP_LOG_APPEND_TIME }

enum	rd_kafka_resp_err_t { RD_KAFKA_RESP_ERR__BEGIN = -200, RD_KAFKA_RESP_ERR__BAD_MSG = -199, RD_KAFKA_RESP_ERR__BAD_COMPRESSION = -198, RD_KAFKA_RESP_ERR__DESTROY = -197, RD_KAFKA_RESP_ERR__FAIL = -196, RD_KAFKA_RESP_ERR__TRANSPORT = -195, RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE = -194, RD_KAFKA_RESP_ERR__RESOLVE = -193, RD_KAFKA_RESP_ERR__MSG_TIMED_OUT = -192, RD_KAFKA_RESP_ERR__PARTITION_EOF = -191, RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION = -190, RD_KAFKA_RESP_ERR__FS = -189, RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC = -188, RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN = -187, RD_KAFKA_RESP_ERR__INVALID_ARG = -186, RD_KAFKA_RESP_ERR__TIMED_OUT = -185, RD_KAFKA_RESP_ERR__QUEUE_FULL = -184, RD_KAFKA_RESP_ERR__ISR_INSUFF = -183, RD_KAFKA_RESP_ERR__NODE_UPDATE = -182, RD_KAFKA_RESP_ERR__SSL = -181, RD_KAFKA_RESP_ERR__WAIT_COORD = -180, RD_KAFKA_RESP_ERR__UNKNOWN_GROUP = -179, RD_KAFKA_RESP_ERR__IN_PROGRESS = -178, RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS = -177, RD_KAFKA_RESP_ERR__EXISTING_SUBSCRIPTION = -176, RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS = -175, RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS = -174, RD_KAFKA_RESP_ERR__CONFLICT = -173, RD_KAFKA_RESP_ERR__STATE = -172, RD_KAFKA_RESP_ERR__UNKNOWN_PROTOCOL = -171, RD_KAFKA_RESP_ERR__NOT_IMPLEMENTED = -170, RD_KAFKA_RESP_ERR__AUTHENTICATION = -169, RD_KAFKA_RESP_ERR__NO_OFFSET = -168, RD_KAFKA_RESP_ERR__OUTDATED = -167, RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE = -166, RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE = -165, RD_KAFKA_RESP_ERR__WAIT_CACHE = -164, RD_KAFKA_RESP_ERR__INTR = -163, RD_KAFKA_RESP_ERR__KEY_SERIALIZATION = -162, RD_KAFKA_RESP_ERR__VALUE_SERIALIZATION = -161, RD_KAFKA_RESP_ERR__KEY_DESERIALIZATION = -160, RD_KAFKA_RESP_ERR__VALUE_DESERIALIZATION = -159, RD_KAFKA_RESP_ERR__PARTIAL = -158, RD_KAFKA_RESP_ERR__READ_ONLY = -157, RD_KAFKA_RESP_ERR__NOENT = -156, RD_KAFKA_RESP_ERR__UNDERFLOW = -155, RD_KAFKA_RESP_ERR__INVALID_TYPE = -154, RD_KAFKA_RESP_ERR__RETRY = -153, RD_KAFKA_RESP_ERR__PURGE_QUEUE = -152, RD_KAFKA_RESP_ERR__PURGE_INFLIGHT = -151, RD_KAFKA_RESP_ERR__FATAL = -150, RD_KAFKA_RESP_ERR__INCONSISTENT = -149, RD_KAFKA_RESP_ERR__GAPLESS_GUARANTEE = -148, RD_KAFKA_RESP_ERR__MAX_POLL_EXCEEDED = -147, RD_KAFKA_RESP_ERR__UNKNOWN_BROKER = -146, RD_KAFKA_RESP_ERR__NOT_CONFIGURED = -145, RD_KAFKA_RESP_ERR__FENCED = -144, RD_KAFKA_RESP_ERR__APPLICATION = -143, RD_KAFKA_RESP_ERR__ASSIGNMENT_LOST = -142, RD_KAFKA_RESP_ERR__NOOP = -141, RD_KAFKA_RESP_ERR__AUTO_OFFSET_RESET = -140, RD_KAFKA_RESP_ERR__END = -100, RD_KAFKA_RESP_ERR_UNKNOWN = -1, RD_KAFKA_RESP_ERR_NO_ERROR = 0, RD_KAFKA_RESP_ERR_OFFSET_OUT_OF_RANGE = 1, RD_KAFKA_RESP_ERR_INVALID_MSG = 2, RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART = 3, RD_KAFKA_RESP_ERR_INVALID_MSG_SIZE = 4, RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE = 5, RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION = 6, RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT = 7, RD_KAFKA_RESP_ERR_BROKER_NOT_AVAILABLE = 8, RD_KAFKA_RESP_ERR_REPLICA_NOT_AVAILABLE = 9, RD_KAFKA_RESP_ERR_MSG_SIZE_TOO_LARGE = 10, RD_KAFKA_RESP_ERR_STALE_CTRL_EPOCH = 11, RD_KAFKA_RESP_ERR_OFFSET_METADATA_TOO_LARGE = 12, RD_KAFKA_RESP_ERR_NETWORK_EXCEPTION = 13, RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS = 14, RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE = 15, RD_KAFKA_RESP_ERR_NOT_COORDINATOR = 16, RD_KAFKA_RESP_ERR_TOPIC_EXCEPTION = 17, RD_KAFKA_RESP_ERR_RECORD_LIST_TOO_LARGE = 18, RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS = 19, RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS_AFTER_APPEND = 20, RD_KAFKA_RESP_ERR_INVALID_REQUIRED_ACKS = 21, RD_KAFKA_RESP_ERR_ILLEGAL_GENERATION = 22, RD_KAFKA_RESP_ERR_INCONSISTENT_GROUP_PROTOCOL = 23, RD_KAFKA_RESP_ERR_INVALID_GROUP_ID = 24, RD_KAFKA_RESP_ERR_UNKNOWN_MEMBER_ID = 25, RD_KAFKA_RESP_ERR_INVALID_SESSION_TIMEOUT = 26, RD_KAFKA_RESP_ERR_REBALANCE_IN_PROGRESS = 27, RD_KAFKA_RESP_ERR_INVALID_COMMIT_OFFSET_SIZE = 28, RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED = 29, RD_KAFKA_RESP_ERR_GROUP_AUTHORIZATION_FAILED = 30, RD_KAFKA_RESP_ERR_CLUSTER_AUTHORIZATION_FAILED = 31, RD_KAFKA_RESP_ERR_INVALID_TIMESTAMP = 32, RD_KAFKA_RESP_ERR_UNSUPPORTED_SASL_MECHANISM = 33, RD_KAFKA_RESP_ERR_ILLEGAL_SASL_STATE = 34, RD_KAFKA_RESP_ERR_UNSUPPORTED_VERSION = 35, RD_KAFKA_RESP_ERR_TOPIC_ALREADY_EXISTS = 36, RD_KAFKA_RESP_ERR_INVALID_PARTITIONS = 37, RD_KAFKA_RESP_ERR_INVALID_REPLICATION_FACTOR = 38, RD_KAFKA_RESP_ERR_INVALID_REPLICA_ASSIGNMENT = 39, RD_KAFKA_RESP_ERR_INVALID_CONFIG = 40, RD_KAFKA_RESP_ERR_NOT_CONTROLLER = 41, RD_KAFKA_RESP_ERR_INVALID_REQUEST = 42, RD_KAFKA_RESP_ERR_UNSUPPORTED_FOR_MESSAGE_FORMAT = 43, RD_KAFKA_RESP_ERR_POLICY_VIOLATION = 44, RD_KAFKA_RESP_ERR_OUT_OF_ORDER_SEQUENCE_NUMBER = 45, RD_KAFKA_RESP_ERR_DUPLICATE_SEQUENCE_NUMBER = 46, RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH = 47, RD_KAFKA_RESP_ERR_INVALID_TXN_STATE = 48, RD_KAFKA_RESP_ERR_INVALID_PRODUCER_ID_MAPPING = 49, RD_KAFKA_RESP_ERR_INVALID_TRANSACTION_TIMEOUT = 50, RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS = 51, RD_KAFKA_RESP_ERR_TRANSACTION_COORDINATOR_FENCED = 52, RD_KAFKA_RESP_ERR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED = 53, RD_KAFKA_RESP_ERR_SECURITY_DISABLED = 54, RD_KAFKA_RESP_ERR_OPERATION_NOT_ATTEMPTED = 55, RD_KAFKA_RESP_ERR_KAFKA_STORAGE_ERROR = 56, RD_KAFKA_RESP_ERR_LOG_DIR_NOT_FOUND = 57, RD_KAFKA_RESP_ERR_SASL_AUTHENTICATION_FAILED = 58, RD_KAFKA_RESP_ERR_UNKNOWN_PRODUCER_ID = 59, RD_KAFKA_RESP_ERR_REASSIGNMENT_IN_PROGRESS = 60, RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_AUTH_DISABLED = 61, RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_NOT_FOUND = 62, RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_OWNER_MISMATCH = 63, RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_REQUEST_NOT_ALLOWED = 64, RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_AUTHORIZATION_FAILED = 65, RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_EXPIRED = 66, RD_KAFKA_RESP_ERR_INVALID_PRINCIPAL_TYPE = 67, RD_KAFKA_RESP_ERR_NON_EMPTY_GROUP = 68, RD_KAFKA_RESP_ERR_GROUP_ID_NOT_FOUND = 69, RD_KAFKA_RESP_ERR_FETCH_SESSION_ID_NOT_FOUND = 70, RD_KAFKA_RESP_ERR_INVALID_FETCH_SESSION_EPOCH = 71, RD_KAFKA_RESP_ERR_LISTENER_NOT_FOUND = 72, RD_KAFKA_RESP_ERR_TOPIC_DELETION_DISABLED = 73, RD_KAFKA_RESP_ERR_FENCED_LEADER_EPOCH = 74, RD_KAFKA_RESP_ERR_UNKNOWN_LEADER_EPOCH = 75, RD_KAFKA_RESP_ERR_UNSUPPORTED_COMPRESSION_TYPE = 76, RD_KAFKA_RESP_ERR_STALE_BROKER_EPOCH = 77, RD_KAFKA_RESP_ERR_OFFSET_NOT_AVAILABLE = 78, RD_KAFKA_RESP_ERR_MEMBER_ID_REQUIRED = 79, RD_KAFKA_RESP_ERR_PREFERRED_LEADER_NOT_AVAILABLE = 80, RD_KAFKA_RESP_ERR_GROUP_MAX_SIZE_REACHED = 81, RD_KAFKA_RESP_ERR_FENCED_INSTANCE_ID = 82, RD_KAFKA_RESP_ERR_ELIGIBLE_LEADERS_NOT_AVAILABLE = 83, RD_KAFKA_RESP_ERR_ELECTION_NOT_NEEDED = 84, RD_KAFKA_RESP_ERR_NO_REASSIGNMENT_IN_PROGRESS = 85, RD_KAFKA_RESP_ERR_GROUP_SUBSCRIBED_TO_TOPIC = 86, RD_KAFKA_RESP_ERR_INVALID_RECORD = 87, RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT = 88, RD_KAFKA_RESP_ERR_THROTTLING_QUOTA_EXCEEDED = 89, RD_KAFKA_RESP_ERR_PRODUCER_FENCED = 90, RD_KAFKA_RESP_ERR_RESOURCE_NOT_FOUND = 91, RD_KAFKA_RESP_ERR_DUPLICATE_RESOURCE = 92, RD_KAFKA_RESP_ERR_UNACCEPTABLE_CREDENTIAL = 93, RD_KAFKA_RESP_ERR_INCONSISTENT_VOTER_SET = 94, RD_KAFKA_RESP_ERR_INVALID_UPDATE_VERSION = 95, RD_KAFKA_RESP_ERR_FEATURE_UPDATE_FAILED = 96, RD_KAFKA_RESP_ERR_PRINCIPAL_DESERIALIZATION_FAILURE = 97, RD_KAFKA_RESP_ERR_END_ALL }
	Error codes. More...

const RD_EXPORT char *	rd_kafka_get_debug_contexts (void)
	Retrieve supported debug contexts for use with the `"debug"` configuration property. (runtime) More...

RD_EXPORT void	rd_kafka_get_err_descs (const struct rd_kafka_err_desc *errdescs, size_t cntp)
	Returns the full list of error codes.

const RD_EXPORT char *	rd_kafka_err2str (rd_kafka_resp_err_t err)
	Returns a human readable representation of a kafka error. More...

const RD_EXPORT char *	rd_kafka_err2name (rd_kafka_resp_err_t err)
	Returns the error code name (enum name). More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_last_error (void)
	Returns the last error code generated by a legacy API call in the current thread. More...

RD_EXPORT RD_DEPRECATED rd_kafka_resp_err_t	rd_kafka_errno2err (int errnox)
	Converts the system errno value `errnox` to a rd_kafka_resp_err_t error code upon failure from the following functions: More...

RD_EXPORT RD_DEPRECATED int	rd_kafka_errno (void)
	Returns the thread-local system errno. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_fatal_error (rd_kafka_t rk, char errstr, size_t errstr_size)
	Returns the first fatal error set on this client instance, or RD_KAFKA_RESP_ERR_NO_ERROR if no fatal error has occurred. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_test_fatal_error (rd_kafka_t rk, rd_kafka_resp_err_t err, const char reason)
	Trigger a fatal error for testing purposes. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_error_code (const rd_kafka_error_t *error)

const RD_EXPORT char *	rd_kafka_error_name (const rd_kafka_error_t *error)

const RD_EXPORT char *	rd_kafka_error_string (const rd_kafka_error_t *error)

RD_EXPORT int	rd_kafka_error_is_fatal (const rd_kafka_error_t *error)

RD_EXPORT int	rd_kafka_error_is_retriable (const rd_kafka_error_t *error)

RD_EXPORT int	rd_kafka_error_txn_requires_abort (const rd_kafka_error_t *error)

RD_EXPORT void	rd_kafka_error_destroy (rd_kafka_error_t *error)
	Free and destroy an error object. More...

RD_EXPORT rd_kafka_error_t *	rd_kafka_error_new (rd_kafka_resp_err_t code, const char *fmt,...) RD_FORMAT(printf
	Create a new error object with error `code` and optional human readable error string in `fmt`. More...

RD_EXPORT void	rd_kafka_topic_partition_destroy (rd_kafka_topic_partition_t *rktpar)
	Destroy a rd_kafka_topic_partition_t. More...

RD_EXPORT rd_kafka_topic_partition_list_t *	rd_kafka_topic_partition_list_new (int size)
	Create a new list/vector Topic+Partition container. More...

RD_EXPORT void	rd_kafka_topic_partition_list_destroy (rd_kafka_topic_partition_list_t *rkparlist)
	Free all resources used by the list and the list itself.

RD_EXPORT rd_kafka_topic_partition_t *	rd_kafka_topic_partition_list_add (rd_kafka_topic_partition_list_t rktparlist, const char topic, int32_t partition)
	Add topic+partition to list. More...

RD_EXPORT void	rd_kafka_topic_partition_list_add_range (rd_kafka_topic_partition_list_t rktparlist, const char topic, int32_t start, int32_t stop)
	Add range of partitions from `start` to `stop` inclusive. More...

RD_EXPORT int	rd_kafka_topic_partition_list_del (rd_kafka_topic_partition_list_t rktparlist, const char topic, int32_t partition)
	Delete partition from list. More...

RD_EXPORT int	rd_kafka_topic_partition_list_del_by_idx (rd_kafka_topic_partition_list_t *rktparlist, int idx)
	Delete partition from list by elems[] index. More...

RD_EXPORT rd_kafka_topic_partition_list_t *	rd_kafka_topic_partition_list_copy (const rd_kafka_topic_partition_list_t *src)
	Make a copy of an existing list. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_topic_partition_list_set_offset (rd_kafka_topic_partition_list_t rktparlist, const char topic, int32_t partition, int64_t offset)
	Set offset to `offset` for `topic` and `partition`. More...

RD_EXPORT rd_kafka_topic_partition_t *	rd_kafka_topic_partition_list_find (const rd_kafka_topic_partition_list_t rktparlist, const char topic, int32_t partition)
	Find element by `topic` and `partition`. More...

RD_EXPORT void	rd_kafka_topic_partition_list_sort (rd_kafka_topic_partition_list_t rktparlist, int(cmp)(const void a, const void b, void cmp_opaque), void cmp_opaque)
	Sort list using comparator `cmp`. More...

Var-arg tag types
#define	RD_KAFKA_V_END RD_KAFKA_VTYPE_END
	Convenience macros for rd_kafka_vtype_t that takes the correct arguments for each vtype. More...

#define	RD_KAFKA_V_TOPIC(topic)

#define	RD_KAFKA_V_RKT(rkt)

#define	RD_KAFKA_V_PARTITION(partition)

#define	RD_KAFKA_V_VALUE(VALUE, LEN)

#define	RD_KAFKA_V_KEY(KEY, LEN)

#define	RD_KAFKA_V_OPAQUE(msg_opaque)

#define	RD_KAFKA_V_MSGFLAGS(msgflags) _LRK_TYPECHECK(RD_KAFKA_VTYPE_MSGFLAGS, int, msgflags), (int)msgflags

#define	RD_KAFKA_V_TIMESTAMP(timestamp)

#define	RD_KAFKA_V_HEADER(NAME, VALUE, LEN)

#define	RD_KAFKA_V_HEADERS(HDRS)

enum	rd_kafka_vtype_t { RD_KAFKA_VTYPE_END, RD_KAFKA_VTYPE_TOPIC, RD_KAFKA_VTYPE_RKT, RD_KAFKA_VTYPE_PARTITION, RD_KAFKA_VTYPE_VALUE, RD_KAFKA_VTYPE_KEY, RD_KAFKA_VTYPE_OPAQUE, RD_KAFKA_VTYPE_MSGFLAGS, RD_KAFKA_VTYPE_TIMESTAMP, RD_KAFKA_VTYPE_HEADER, RD_KAFKA_VTYPE_HEADERS }
	Var-arg tag types. More...

Kafka messages
enum	rd_kafka_msg_status_t { RD_KAFKA_MSG_STATUS_NOT_PERSISTED = 0, RD_KAFKA_MSG_STATUS_POSSIBLY_PERSISTED = 1, RD_KAFKA_MSG_STATUS_PERSISTED = 2 }
	Message persistence status can be used by the application to find out if a produced message was persisted in the topic log. More...

RD_EXPORT void	rd_kafka_message_destroy (rd_kafka_message_t *rkmessage)
	Frees resources for `rkmessage` and hands ownership back to rdkafka.

const RD_EXPORT char *	rd_kafka_message_errstr (const rd_kafka_message_t *rkmessage)
	Returns the error string for an errored rd_kafka_message_t or NULL if there was no error. More...

RD_EXPORT int64_t	rd_kafka_message_timestamp (const rd_kafka_message_t rkmessage, rd_kafka_timestamp_type_t tstype)
	Returns the message timestamp for a consumed message. More...

RD_EXPORT int64_t	rd_kafka_message_latency (const rd_kafka_message_t *rkmessage)
	Returns the latency for a produced message measured from the produce() call. More...

RD_EXPORT int32_t	rd_kafka_message_broker_id (const rd_kafka_message_t *rkmessage)
	Returns the broker id of the broker the message was produced to or fetched from. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_message_headers (const rd_kafka_message_t rkmessage, rd_kafka_headers_t *hdrsp)
	Get the message header list. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_message_detach_headers (rd_kafka_message_t rkmessage, rd_kafka_headers_t *hdrsp)
	Get the message header list and detach the list from the message making the application the owner of the headers. The application must eventually destroy the headers using rd_kafka_headers_destroy(). The message's headers will be set to NULL. More...

RD_EXPORT void	rd_kafka_message_set_headers (rd_kafka_message_t rkmessage, rd_kafka_headers_t hdrs)
	Replace the message's current headers with a new list. More...

RD_EXPORT size_t	rd_kafka_header_cnt (const rd_kafka_headers_t *hdrs)
	Returns the number of header key/value pairs. More...

RD_EXPORT rd_kafka_msg_status_t	rd_kafka_message_status (const rd_kafka_message_t *rkmessage)
	Returns the message's persistence status in the topic log. More...

Configuration interface
Main/global configuration property interface
enum	rd_kafka_conf_res_t { RD_KAFKA_CONF_UNKNOWN = -2, RD_KAFKA_CONF_INVALID = -1, RD_KAFKA_CONF_OK = 0 }
	Configuration result type. More...

enum	rd_kafka_cert_type_t { RD_KAFKA_CERT_PUBLIC_KEY, RD_KAFKA_CERT_PRIVATE_KEY, RD_KAFKA_CERT_CA, RD_KAFKA_CERT__CNT }
	SSL certificate type. More...

enum	rd_kafka_cert_enc_t { RD_KAFKA_CERT_ENC_PKCS12, RD_KAFKA_CERT_ENC_DER, RD_KAFKA_CERT_ENC_PEM, RD_KAFKA_CERT_ENC__CNT }
	SSL certificate encoding. More...

RD_EXPORT rd_kafka_conf_t *	rd_kafka_conf_new (void)
	Create configuration object. More...

RD_EXPORT void	rd_kafka_conf_destroy (rd_kafka_conf_t *conf)
	Destroys a conf object.

RD_EXPORT rd_kafka_conf_t *	rd_kafka_conf_dup (const rd_kafka_conf_t *conf)
	Creates a copy/duplicate of configuration object `conf`. More...

RD_EXPORT rd_kafka_conf_t *	rd_kafka_conf_dup_filter (const rd_kafka_conf_t conf, size_t filter_cnt, const char *filter)
	Same as rd_kafka_conf_dup() but with an array of property name prefixes to filter out (ignore) when copying.

const RD_EXPORT rd_kafka_conf_t *	rd_kafka_conf (rd_kafka_t *rk)

RD_EXPORT rd_kafka_conf_res_t	rd_kafka_conf_set (rd_kafka_conf_t conf, const char name, const char value, char errstr, size_t errstr_size)
	Sets a configuration property. More...

RD_EXPORT void	rd_kafka_conf_set_events (rd_kafka_conf_t *conf, int events)
	Enable event sourcing. `events` is a bitmask of `RD_KAFKA_EVENT_*` of events to enable for consumption by `rd_kafka_queue_poll()`.

RD_EXPORT void	rd_kafka_conf_set_background_event_cb (rd_kafka_conf_t conf, void(event_cb)(rd_kafka_t rk, rd_kafka_event_t rkev, void *opaque))
	Generic event callback to be used with the event API to trigger callbacks for `rd_kafka_event_t` objects from a background thread serving the background queue. More...

RD_EXPORT void	rd_kafka_conf_set_dr_cb (rd_kafka_conf_t conf, void(dr_cb)(rd_kafka_t rk, void payload, size_t len, rd_kafka_resp_err_t err, void opaque, void msg_opaque))

RD_EXPORT void	rd_kafka_conf_set_dr_msg_cb (rd_kafka_conf_t conf, void(dr_msg_cb)(rd_kafka_t rk, const rd_kafka_message_t rkmessage, void *opaque))
	Producer: Set delivery report callback in provided `conf` object. More...

RD_EXPORT void	rd_kafka_conf_set_consume_cb (rd_kafka_conf_t conf, void(consume_cb)(rd_kafka_message_t rkmessage, void opaque))
	Consumer: Set consume callback for use with rd_kafka_consumer_poll() More...

RD_EXPORT void	rd_kafka_conf_set_rebalance_cb (rd_kafka_conf_t conf, void(rebalance_cb)(rd_kafka_t rk, rd_kafka_resp_err_t err, rd_kafka_topic_partition_list_t partitions, void *opaque))
	Consumer: Set rebalance callback for use with coordinated consumer group balancing. More...

RD_EXPORT void	rd_kafka_conf_set_offset_commit_cb (rd_kafka_conf_t conf, void(offset_commit_cb)(rd_kafka_t rk, rd_kafka_resp_err_t err, rd_kafka_topic_partition_list_t offsets, void *opaque))
	Consumer: Set offset commit callback for use with consumer groups. More...

RD_EXPORT void	rd_kafka_conf_set_error_cb (rd_kafka_conf_t conf, void(error_cb)(rd_kafka_t rk, int err, const char reason, void *opaque))
	Set error callback in provided conf object. More...

RD_EXPORT void	rd_kafka_conf_set_throttle_cb (rd_kafka_conf_t conf, void(throttle_cb)(rd_kafka_t rk, const char broker_name, int32_t broker_id, int throttle_time_ms, void *opaque))
	Set throttle callback. More...

RD_EXPORT void	rd_kafka_conf_set_log_cb (rd_kafka_conf_t conf, void(log_cb)(const rd_kafka_t rk, int level, const char fac, const char *buf))
	Set logger callback. More...

RD_EXPORT void	rd_kafka_conf_set_stats_cb (rd_kafka_conf_t conf, int(stats_cb)(rd_kafka_t rk, char json, size_t json_len, void *opaque))
	Set statistics callback in provided conf object. More...

RD_EXPORT void	rd_kafka_conf_set_oauthbearer_token_refresh_cb (rd_kafka_conf_t conf, void(oauthbearer_token_refresh_cb)(rd_kafka_t rk, const char oauthbearer_config, void *opaque))
	Set SASL/OAUTHBEARER token refresh callback in provided conf object. More...

RD_EXPORT void	rd_kafka_conf_enable_sasl_queue (rd_kafka_conf_t *conf, int enable)
	Enable/disable creation of a queue specific to SASL events and callbacks. More...

RD_EXPORT void	rd_kafka_conf_set_socket_cb (rd_kafka_conf_t conf, int(socket_cb)(int domain, int type, int protocol, void *opaque))
	Set socket callback. More...

RD_EXPORT void	rd_kafka_conf_set_connect_cb (rd_kafka_conf_t conf, int(connect_cb)(int sockfd, const struct sockaddr addr, int addrlen, const char id, void *opaque))
	Set connect callback. More...

RD_EXPORT void	rd_kafka_conf_set_closesocket_cb (rd_kafka_conf_t conf, int(closesocket_cb)(int sockfd, void *opaque))
	Set close socket callback. More...

RD_EXPORT void	rd_kafka_conf_set_open_cb (rd_kafka_conf_t conf, int(open_cb)(const char pathname, int flags, mode_t mode, void opaque))
	Set open callback. More...

RD_EXPORT rd_kafka_conf_res_t	rd_kafka_conf_set_ssl_cert_verify_cb (rd_kafka_conf_t conf, int(ssl_cert_verify_cb)(rd_kafka_t rk, const char broker_name, int32_t broker_id, int x509_error, int depth, const char buf, size_t size, char errstr, size_t errstr_size, void opaque))
	Sets the verification callback of the broker certificate. More...

RD_EXPORT rd_kafka_conf_res_t	rd_kafka_conf_set_ssl_cert (rd_kafka_conf_t conf, rd_kafka_cert_type_t cert_type, rd_kafka_cert_enc_t cert_enc, const void buffer, size_t size, char *errstr, size_t errstr_size)
	Set certificate/key `cert_type` from the `cert_enc` encoded memory at `buffer` of `size` bytes. More...

RD_EXPORT void	rd_kafka_conf_set_engine_callback_data (rd_kafka_conf_t conf, void callback_data)
	Set callback_data for OpenSSL engine. More...

RD_EXPORT void	rd_kafka_conf_set_opaque (rd_kafka_conf_t conf, void opaque)
	Sets the application's opaque pointer that will be passed to callbacks. More...

RD_EXPORT void *	rd_kafka_opaque (const rd_kafka_t *rk)
	Retrieves the opaque pointer previously set with rd_kafka_conf_set_opaque()

RD_EXPORT void	rd_kafka_conf_set_default_topic_conf (rd_kafka_conf_t conf, rd_kafka_topic_conf_t tconf)
	Sets the default topic configuration to use for automatically subscribed topics (e.g., through pattern-matched topics). The topic config object is not usable after this call. More...

RD_EXPORT rd_kafka_topic_conf_t *	rd_kafka_conf_get_default_topic_conf (rd_kafka_conf_t *conf)
	Gets the default topic configuration as previously set with rd_kafka_conf_set_default_topic_conf() or that was implicitly created by configuring a topic-level property on the global `conf` object. More...

RD_EXPORT rd_kafka_conf_res_t	rd_kafka_conf_get (const rd_kafka_conf_t conf, const char name, char dest, size_t dest_size)
	Retrieve configuration value for property `name`. More...

RD_EXPORT rd_kafka_conf_res_t	rd_kafka_topic_conf_get (const rd_kafka_topic_conf_t conf, const char name, char dest, size_t dest_size)
	Retrieve topic configuration value for property `name`. More...

const RD_EXPORT char **	rd_kafka_conf_dump (rd_kafka_conf_t conf, size_t cntp)
	Dump the configuration properties and values of `conf` to an array with "key", "value" pairs. More...

const RD_EXPORT char **	rd_kafka_topic_conf_dump (rd_kafka_topic_conf_t conf, size_t cntp)
	Dump the topic configuration properties and values of `conf` to an array with "key", "value" pairs. More...

RD_EXPORT void	rd_kafka_conf_dump_free (const char **arr, size_t cnt)
	Frees a configuration dump returned from `rd_kafka_conf_dump()` or `rd_kafka_topic_conf_dump().

RD_EXPORT void	rd_kafka_conf_properties_show (FILE *fp)
	Prints a table to `fp` of all supported configuration properties, their default values as well as a description. More...

Main Kafka and Topic object handles
#define	RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE 0x8
	Flags for rd_kafka_destroy_flags() More...

#define	RD_KAFKA_PARTITION_UA ((int32_t)-1)
	Unassigned partition. More...

RD_EXPORT rd_kafka_t *	rd_kafka_new (rd_kafka_type_t type, rd_kafka_conf_t conf, char errstr, size_t errstr_size)
	Creates a new Kafka handle and starts its operation according to the specified `type` (`RD_KAFKA_CONSUMER` or `RD_KAFKA_PRODUCER`). More...

RD_EXPORT void	rd_kafka_destroy (rd_kafka_t *rk)
	Destroy Kafka handle. More...

RD_EXPORT void	rd_kafka_destroy_flags (rd_kafka_t *rk, int flags)
	Destroy Kafka handle according to specified destroy flags. More...

const RD_EXPORT char *	rd_kafka_name (const rd_kafka_t *rk)
	Returns Kafka handle name.

RD_EXPORT rd_kafka_type_t	rd_kafka_type (const rd_kafka_t *rk)
	Returns Kafka handle type.

RD_EXPORT char *	rd_kafka_memberid (const rd_kafka_t *rk)
	Returns this client's broker-assigned group member id. More...

RD_EXPORT char *	rd_kafka_clusterid (rd_kafka_t *rk, int timeout_ms)
	Returns the ClusterId as reported in broker metadata. More...

RD_EXPORT int32_t	rd_kafka_controllerid (rd_kafka_t *rk, int timeout_ms)
	Returns the current ControllerId as reported in broker metadata. More...

RD_EXPORT rd_kafka_topic_t *	rd_kafka_topic_new (rd_kafka_t rk, const char topic, rd_kafka_topic_conf_t *conf)
	Creates a new topic handle for topic named `topic`. More...

RD_EXPORT void	rd_kafka_topic_destroy (rd_kafka_topic_t *rkt)
	Loose application's topic handle refcount as previously created with `rd_kafka_topic_new()`. More...

const RD_EXPORT char *	rd_kafka_topic_name (const rd_kafka_topic_t *rkt)
	Returns the topic name.

RD_EXPORT void *	rd_kafka_topic_opaque (const rd_kafka_topic_t *rkt)
	Get the `rkt_opaque` pointer that was set in the topic configuration with rd_kafka_topic_conf_set_opaque().

RD_EXPORT int	rd_kafka_poll (rd_kafka_t *rk, int timeout_ms)
	Polls the provided kafka handle for events. More...

RD_EXPORT void	rd_kafka_yield (rd_kafka_t *rk)
	Cancels the current callback dispatcher (rd_kafka_poll(), rd_kafka_consume_callback(), etc). More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_pause_partitions (rd_kafka_t rk, rd_kafka_topic_partition_list_t partitions)
	Pause producing or consumption for the provided list of partitions. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_resume_partitions (rd_kafka_t rk, rd_kafka_topic_partition_list_t partitions)
	Resume producing consumption for the provided list of partitions. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_query_watermark_offsets (rd_kafka_t rk, const char topic, int32_t partition, int64_t low, int64_t high, int timeout_ms)
	Query broker for low (oldest/beginning) and high (newest/end) offsets for partition. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_get_watermark_offsets (rd_kafka_t rk, const char topic, int32_t partition, int64_t low, int64_t high)
	Get last known low (oldest/beginning) and high (newest/end) offsets for partition. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_offsets_for_times (rd_kafka_t rk, rd_kafka_topic_partition_list_t offsets, int timeout_ms)
	Look up the offsets for the given partitions by timestamp. More...

RD_EXPORT void *	rd_kafka_mem_calloc (rd_kafka_t *rk, size_t num, size_t size)
	Allocate and zero memory using the same allocator librdkafka uses. More...

RD_EXPORT void *	rd_kafka_mem_malloc (rd_kafka_t *rk, size_t size)
	Allocate memory using the same allocator librdkafka uses. More...

RD_EXPORT void	rd_kafka_mem_free (rd_kafka_t rk, void ptr)
	Free pointer returned by librdkafka. More...

Simple Consumer API (legacy)
#define	RD_KAFKA_OFFSET_BEGINNING -2

#define	RD_KAFKA_OFFSET_END -1

#define	RD_KAFKA_OFFSET_STORED -1000

#define	RD_KAFKA_OFFSET_INVALID -1001

#define	RD_KAFKA_OFFSET_TAIL(CNT) (RD_KAFKA_OFFSET_TAIL_BASE - (CNT))
	Start consuming `CNT` messages from topic's current end offset. More...

RD_EXPORT int	rd_kafka_consume_start (rd_kafka_topic_t *rkt, int32_t partition, int64_t offset)
	Start consuming messages for topic `rkt` and `partition` at offset `offset` which may either be an absolute (0..N) or one of the logical offsets: More...

RD_EXPORT int	rd_kafka_consume_start_queue (rd_kafka_topic_t rkt, int32_t partition, int64_t offset, rd_kafka_queue_t rkqu)
	Same as rd_kafka_consume_start() but re-routes incoming messages to the provided queue `rkqu` (which must have been previously allocated with `rd_kafka_queue_new()`. More...

RD_EXPORT int	rd_kafka_consume_stop (rd_kafka_topic_t *rkt, int32_t partition)
	Stop consuming messages for topic `rkt` and `partition`, purging all messages currently in the local queue. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_seek (rd_kafka_topic_t *rkt, int32_t partition, int64_t offset, int timeout_ms)
	Seek consumer for topic+partition to `offset` which is either an absolute or logical offset. More...

RD_EXPORT rd_kafka_error_t *	rd_kafka_seek_partitions (rd_kafka_t rk, rd_kafka_topic_partition_list_t partitions, int timeout_ms)
	Seek consumer for partitions in `partitions` to the per-partition offset in the .offset field of `partitions`. More...

RD_EXPORT rd_kafka_message_t *	rd_kafka_consume (rd_kafka_topic_t *rkt, int32_t partition, int timeout_ms)
	Consume a single message from topic `rkt` and `partition`. More...

RD_EXPORT ssize_t	rd_kafka_consume_batch (rd_kafka_topic_t rkt, int32_t partition, int timeout_ms, rd_kafka_message_t *rkmessages, size_t rkmessages_size)
	Consume up to `rkmessages_size` from topic `rkt` and `partition` putting a pointer to each message in the application provided array `rkmessages` (of size `rkmessages_size` entries). More...

RD_EXPORT int	rd_kafka_consume_callback (rd_kafka_topic_t rkt, int32_t partition, int timeout_ms, void(consume_cb)(rd_kafka_message_t rkmessage, void commit_opaque), void *commit_opaque)
	Consumes messages from topic `rkt` and `partition`, calling the provided callback for each consumed messsage. More...

Producer API
#define	RD_KAFKA_MSG_F_FREE 0x1
	Producer message flags. More...

#define	RD_KAFKA_MSG_F_COPY 0x2

#define	RD_KAFKA_MSG_F_BLOCK 0x4

#define	RD_KAFKA_MSG_F_PARTITION 0x8

#define	RD_KAFKA_PURGE_F_QUEUE 0x1
	Flags for rd_kafka_purge() More...

#define	RD_KAFKA_PURGE_F_INFLIGHT 0x2

#define	RD_KAFKA_PURGE_F_NON_BLOCKING 0x4

RD_EXPORT int	rd_kafka_produce (rd_kafka_topic_t rkt, int32_t partition, int msgflags, void payload, size_t len, const void key, size_t keylen, void msg_opaque)
	Produce and send a single message to broker. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_producev (rd_kafka_t *rk,...)
	Produce and send a single message to broker. More...

RD_EXPORT rd_kafka_error_t *	rd_kafka_produceva (rd_kafka_t rk, const rd_kafka_vu_t vus, size_t cnt)
	Produce and send a single message to broker. More...

RD_EXPORT int	rd_kafka_produce_batch (rd_kafka_topic_t rkt, int32_t partition, int msgflags, rd_kafka_message_t rkmessages, int message_cnt)
	Produce multiple messages. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_flush (rd_kafka_t *rk, int timeout_ms)
	Wait until all outstanding produce requests, et.al, are completed. This should typically be done prior to destroying a producer instance to make sure all queued and in-flight produce requests are completed before terminating. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_purge (rd_kafka_t *rk, int purge_flags)
	Purge messages currently handled by the producer instance. More...

Miscellaneous APIs
enum	rd_kafka_thread_type_t { RD_KAFKA_THREAD_MAIN, RD_KAFKA_THREAD_BACKGROUND, RD_KAFKA_THREAD_BROKER }
	librdkafka internal thread type. More...

RD_EXPORT int	rd_kafka_brokers_add (rd_kafka_t rk, const char brokerlist)
	Adds one or more brokers to the kafka handle's list of initial bootstrap brokers. More...

RD_EXPORT RD_DEPRECATED void	rd_kafka_set_logger (rd_kafka_t rk, void(func)(const rd_kafka_t rk, int level, const char fac, const char *buf))
	Set logger function. More...

RD_EXPORT void	rd_kafka_set_log_level (rd_kafka_t *rk, int level)
	Specifies the maximum logging level emitted by internal kafka logging and debugging. More...

RD_EXPORT void	rd_kafka_log_print (const rd_kafka_t rk, int level, const char fac, const char *buf)
	Builtin (default) log sink: print to stderr.

RD_EXPORT void	rd_kafka_log_syslog (const rd_kafka_t rk, int level, const char fac, const char *buf)
	Builtin log sink: print to syslog. More...

RD_EXPORT int	rd_kafka_outq_len (rd_kafka_t *rk)
	Returns the current out queue length. More...

RD_EXPORT void	rd_kafka_dump (FILE fp, rd_kafka_t rk)
	Dumps rdkafka's internal state for handle `rk` to stream `fp`. More...

RD_EXPORT int	rd_kafka_thread_cnt (void)
	Retrieve the current number of threads in use by librdkafka. More...

RD_EXPORT int	rd_kafka_wait_destroyed (int timeout_ms)
	Wait for all rd_kafka_t objects to be destroyed. More...

RD_EXPORT int	rd_kafka_unittest (void)
	Run librdkafka's built-in unit-tests. More...

Event interface
The event API provides an alternative pollable non-callback interface to librdkafka's message and event queues.
#define	RD_KAFKA_EVENT_NONE 0x0

#define	RD_KAFKA_EVENT_DR 0x1

#define	RD_KAFKA_EVENT_FETCH 0x2

#define	RD_KAFKA_EVENT_LOG 0x4

#define	RD_KAFKA_EVENT_ERROR 0x8

#define	RD_KAFKA_EVENT_REBALANCE 0x10

#define	RD_KAFKA_EVENT_OFFSET_COMMIT 0x20

#define	RD_KAFKA_EVENT_STATS 0x40

#define	RD_KAFKA_EVENT_CREATETOPICS_RESULT 100

#define	RD_KAFKA_EVENT_DELETETOPICS_RESULT 101

#define	RD_KAFKA_EVENT_CREATEPARTITIONS_RESULT 102

#define	RD_KAFKA_EVENT_ALTERCONFIGS_RESULT 103

#define	RD_KAFKA_EVENT_DESCRIBECONFIGS_RESULT 104

#define	RD_KAFKA_EVENT_DELETERECORDS_RESULT 105

#define	RD_KAFKA_EVENT_DELETEGROUPS_RESULT 106

#define	RD_KAFKA_EVENT_DELETECONSUMERGROUPOFFSETS_RESULT 107

#define	RD_KAFKA_EVENT_OAUTHBEARER_TOKEN_REFRESH 0x100

#define	RD_KAFKA_EVENT_BACKGROUND 0x200

#define	RD_KAFKA_EVENT_CREATEACLS_RESULT 0x400

#define	RD_KAFKA_EVENT_DESCRIBEACLS_RESULT 0x800

#define	RD_KAFKA_EVENT_DELETEACLS_RESULT 0x1000

typedef int	rd_kafka_event_type_t
	Event types.

typedef rd_kafka_event_t	rd_kafka_CreateTopics_result_t

typedef rd_kafka_event_t	rd_kafka_DeleteTopics_result_t

typedef rd_kafka_event_t	rd_kafka_CreateAcls_result_t

typedef rd_kafka_event_t	rd_kafka_DescribeAcls_result_t

typedef rd_kafka_event_t	rd_kafka_DeleteAcls_result_t

typedef rd_kafka_event_t	rd_kafka_CreatePartitions_result_t

typedef rd_kafka_event_t	rd_kafka_AlterConfigs_result_t

typedef rd_kafka_event_t	rd_kafka_DescribeConfigs_result_t

typedef rd_kafka_event_t	rd_kafka_DeleteRecords_result_t

typedef rd_kafka_event_t	rd_kafka_DeleteGroups_result_t

typedef rd_kafka_event_t	rd_kafka_DeleteConsumerGroupOffsets_result_t

RD_EXPORT rd_kafka_event_type_t	rd_kafka_event_type (const rd_kafka_event_t *rkev)

const RD_EXPORT char *	rd_kafka_event_name (const rd_kafka_event_t *rkev)

RD_EXPORT void	rd_kafka_event_destroy (rd_kafka_event_t *rkev)
	Destroy an event. More...

const RD_EXPORT rd_kafka_message_t *	rd_kafka_event_message_next (rd_kafka_event_t *rkev)

RD_EXPORT size_t	rd_kafka_event_message_array (rd_kafka_event_t rkev, const rd_kafka_message_t *rkmessages, size_t size)
	Extacts `size` message(s) from the event into the pre-allocated array `rkmessages`. More...

RD_EXPORT size_t	rd_kafka_event_message_count (rd_kafka_event_t *rkev)

const RD_EXPORT char *	rd_kafka_event_config_string (rd_kafka_event_t *rkev)

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_event_error (rd_kafka_event_t *rkev)

const RD_EXPORT char *	rd_kafka_event_error_string (rd_kafka_event_t *rkev)

RD_EXPORT int	rd_kafka_event_error_is_fatal (rd_kafka_event_t *rkev)

RD_EXPORT void *	rd_kafka_event_opaque (rd_kafka_event_t *rkev)

RD_EXPORT int	rd_kafka_event_log (rd_kafka_event_t rkev, const char fac, const char str, int level)
	Extract log message from the event. More...

RD_EXPORT int	rd_kafka_event_debug_contexts (rd_kafka_event_t rkev, char dst, size_t dstsize)
	Extract log debug context from event. More...

const RD_EXPORT char *	rd_kafka_event_stats (rd_kafka_event_t *rkev)
	Extract stats from the event. More...

RD_EXPORT rd_kafka_topic_partition_list_t *	rd_kafka_event_topic_partition_list (rd_kafka_event_t *rkev)

RD_EXPORT rd_kafka_topic_partition_t *	rd_kafka_event_topic_partition (rd_kafka_event_t *rkev)

const RD_EXPORT rd_kafka_CreateTopics_result_t *	rd_kafka_event_CreateTopics_result (rd_kafka_event_t *rkev)
	Get CreateTopics result. More...

const RD_EXPORT rd_kafka_DeleteTopics_result_t *	rd_kafka_event_DeleteTopics_result (rd_kafka_event_t *rkev)
	Get DeleteTopics result. More...

const RD_EXPORT rd_kafka_CreatePartitions_result_t *	rd_kafka_event_CreatePartitions_result (rd_kafka_event_t *rkev)
	Get CreatePartitions result. More...

const RD_EXPORT rd_kafka_AlterConfigs_result_t *	rd_kafka_event_AlterConfigs_result (rd_kafka_event_t *rkev)
	Get AlterConfigs result. More...

const RD_EXPORT rd_kafka_DescribeConfigs_result_t *	rd_kafka_event_DescribeConfigs_result (rd_kafka_event_t *rkev)
	Get DescribeConfigs result. More...

const RD_EXPORT rd_kafka_DeleteRecords_result_t *	rd_kafka_event_DeleteRecords_result (rd_kafka_event_t *rkev)

const RD_EXPORT rd_kafka_DeleteGroups_result_t *	rd_kafka_event_DeleteGroups_result (rd_kafka_event_t *rkev)
	Get DeleteGroups result. More...

const RD_EXPORT rd_kafka_DeleteConsumerGroupOffsets_result_t *	rd_kafka_event_DeleteConsumerGroupOffsets_result (rd_kafka_event_t *rkev)
	Get DeleteConsumerGroupOffsets result. More...

const RD_EXPORT rd_kafka_CreateAcls_result_t *	rd_kafka_event_CreateAcls_result (rd_kafka_event_t *rkev)

const RD_EXPORT rd_kafka_DescribeAcls_result_t *	rd_kafka_event_DescribeAcls_result (rd_kafka_event_t *rkev)

const RD_EXPORT rd_kafka_DeleteAcls_result_t *	rd_kafka_event_DeleteAcls_result (rd_kafka_event_t *rkev)

RD_EXPORT rd_kafka_event_t *	rd_kafka_queue_poll (rd_kafka_queue_t *rkqu, int timeout_ms)
	Poll a queue for an event for max `timeout_ms`. More...

RD_EXPORT int	rd_kafka_queue_poll_callback (rd_kafka_queue_t *rkqu, int timeout_ms)
	Poll a queue for events served through callbacks for max `timeout_ms`. More...

Interceptors
A callback interface that allows message interception for both producer and consumer data pipelines. Except for the on_new(), on_conf_set(), on_conf_dup() and on_conf_destroy() interceptors, interceptors are added to the newly created rd_kafka_t client instance. These interceptors MUST only be added from on_new() and MUST NOT be added after rd_kafka_new() returns. The on_new(), on_conf_set(), on_conf_dup() and on_conf_destroy() interceptors are added to the configuration object which is later passed to rd_kafka_new() where on_new() is called to allow addition of other interceptors. Each interceptor reference consists of a display name (ic_name), a callback function, and an application-specified opaque value that is passed as-is to the callback. The ic_name must be unique for the interceptor implementation and is used to reject duplicate interceptor methods. Any number of interceptors can be added and they are called in the order they were added, unless otherwise noted. The list of registered interceptor methods are referred to as interceptor chains. Remarks Contrary to the Java client the librdkafka interceptor interface does not support message key and value modification. Message mutability is discouraged in the Java client and the combination of serializers and headers cover most use-cases. Interceptors are NOT copied to the new configuration on rd_kafka_conf_dup() since it would be hard for interceptors to track usage of the interceptor's opaque value. An interceptor should rely on the plugin, which will be copied in rd_kafka_conf_conf_dup(), to set up the initial interceptors. An interceptor should implement the on_conf_dup() method to manually set up its internal configuration on the newly created configuration object that is being copied-to based on the interceptor-specific configuration properties. conf_dup() should thus be treated the same as conf_init(). Interceptors are keyed by the interceptor type (on_..()), the interceptor name (ic_name) and the interceptor method function. Duplicates are not allowed and the .._add_on_..() method will return RD_KAFKA_RESP_ERR__CONFLICT if attempting to add a duplicate method. The only exception is on_conf_destroy() which may be added multiple times by the same interceptor to allow proper cleanup of interceptor configuration state.
typedef rd_kafka_conf_res_t()	rd_kafka_interceptor_f_on_conf_set_t(rd_kafka_conf_t conf, const char name, const char val, char errstr, size_t errstr_size, void *ic_opaque)
	on_conf_set() is called from rd_kafka_*_conf_set() in the order the interceptors were added. More...

typedef rd_kafka_resp_err_t()	rd_kafka_interceptor_f_on_conf_dup_t(rd_kafka_conf_t new_conf, const rd_kafka_conf_t old_conf, size_t filter_cnt, const char *filter, void ic_opaque)
	on_conf_dup() is called from rd_kafka_conf_dup() in the order the interceptors were added and is used to let an interceptor re-register its conf interecptors with a new opaque value. The on_conf_dup() method is called prior to the configuration from `old_conf` being copied to `new_conf`. More...

typedef rd_kafka_resp_err_t()	rd_kafka_interceptor_f_on_conf_destroy_t(void *ic_opaque)
	on_conf_destroy() is called from rd_kafka_*_conf_destroy() in the order the interceptors were added. More...

typedef rd_kafka_resp_err_t()	rd_kafka_interceptor_f_on_new_t(rd_kafka_t rk, const rd_kafka_conf_t conf, void ic_opaque, char errstr, size_t errstr_size)
	on_new() is called from rd_kafka_new() prior toreturning the newly created client instance to the application. More...

typedef rd_kafka_resp_err_t()	rd_kafka_interceptor_f_on_destroy_t(rd_kafka_t rk, void ic_opaque)
	on_destroy() is called from rd_kafka_destroy() or (rd_kafka_new() if rd_kafka_new() fails during initialization). More...

typedef rd_kafka_resp_err_t()	rd_kafka_interceptor_f_on_send_t(rd_kafka_t rk, rd_kafka_message_t rkmessage, void *ic_opaque)
	on_send() is called from rd_kafka_produce*() (et.al) prior to the partitioner being called. More...

typedef rd_kafka_resp_err_t()	rd_kafka_interceptor_f_on_acknowledgement_t(rd_kafka_t rk, rd_kafka_message_t rkmessage, void *ic_opaque)
	on_acknowledgement() is called to inform interceptors that a message was succesfully delivered or permanently failed delivery. The interceptor chain is called from internal librdkafka background threads, or rd_kafka_produce*() if the partitioner failed. More...

typedef rd_kafka_resp_err_t()	rd_kafka_interceptor_f_on_consume_t(rd_kafka_t rk, rd_kafka_message_t rkmessage, void *ic_opaque)
	on_consume() is called just prior to passing the message to the application in rd_kafka_consumer_poll(), rd_kafka_consume*(), the event interface, etc. More...

typedef rd_kafka_resp_err_t()	rd_kafka_interceptor_f_on_commit_t(rd_kafka_t rk, const rd_kafka_topic_partition_list_t offsets, rd_kafka_resp_err_t err, void *ic_opaque)
	on_commit() is called on completed or failed offset commit. It is called from internal librdkafka threads. More...

typedef rd_kafka_resp_err_t()	rd_kafka_interceptor_f_on_request_sent_t(rd_kafka_t rk, int sockfd, const char brokername, int32_t brokerid, int16_t ApiKey, int16_t ApiVersion, int32_t CorrId, size_t size, void *ic_opaque)
	on_request_sent() is called when a request has been fully written to a broker TCP connections socket. More...

typedef rd_kafka_resp_err_t()	rd_kafka_interceptor_f_on_response_received_t(rd_kafka_t rk, int sockfd, const char brokername, int32_t brokerid, int16_t ApiKey, int16_t ApiVersion, int32_t CorrId, size_t size, int64_t rtt, rd_kafka_resp_err_t err, void *ic_opaque)
	on_response_received() is called when a protocol response has been fully received from a broker TCP connection socket but before the response payload is parsed. More...

typedef rd_kafka_resp_err_t()	rd_kafka_interceptor_f_on_thread_start_t(rd_kafka_t rk, rd_kafka_thread_type_t thread_type, const char thread_name, void *ic_opaque)
	on_thread_start() is called from a newly created librdkafka-managed thread. More...

typedef rd_kafka_resp_err_t()	rd_kafka_interceptor_f_on_thread_exit_t(rd_kafka_t rk, rd_kafka_thread_type_t thread_type, const char thread_name, void *ic_opaque)
	on_thread_exit() is called just prior to a librdkafka-managed thread exiting from the exiting thread itself. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_conf_interceptor_add_on_conf_set (rd_kafka_conf_t conf, const char ic_name, rd_kafka_interceptor_f_on_conf_set_t on_conf_set, void ic_opaque)
	Append an on_conf_set() interceptor. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_conf_interceptor_add_on_conf_dup (rd_kafka_conf_t conf, const char ic_name, rd_kafka_interceptor_f_on_conf_dup_t on_conf_dup, void ic_opaque)
	Append an on_conf_dup() interceptor. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_conf_interceptor_add_on_conf_destroy (rd_kafka_conf_t conf, const char ic_name, rd_kafka_interceptor_f_on_conf_destroy_t on_conf_destroy, void ic_opaque)
	Append an on_conf_destroy() interceptor. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_conf_interceptor_add_on_new (rd_kafka_conf_t conf, const char ic_name, rd_kafka_interceptor_f_on_new_t on_new, void ic_opaque)
	Append an on_new() interceptor. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_interceptor_add_on_destroy (rd_kafka_t rk, const char ic_name, rd_kafka_interceptor_f_on_destroy_t on_destroy, void ic_opaque)
	Append an on_destroy() interceptor. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_interceptor_add_on_send (rd_kafka_t rk, const char ic_name, rd_kafka_interceptor_f_on_send_t on_send, void ic_opaque)
	Append an on_send() interceptor. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_interceptor_add_on_acknowledgement (rd_kafka_t rk, const char ic_name, rd_kafka_interceptor_f_on_acknowledgement_t on_acknowledgement, void ic_opaque)
	Append an on_acknowledgement() interceptor. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_interceptor_add_on_consume (rd_kafka_t rk, const char ic_name, rd_kafka_interceptor_f_on_consume_t on_consume, void ic_opaque)
	Append an on_consume() interceptor. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_interceptor_add_on_commit (rd_kafka_t rk, const char ic_name, rd_kafka_interceptor_f_on_commit_t on_commit, void ic_opaque)
	Append an on_commit() interceptor. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_interceptor_add_on_request_sent (rd_kafka_t rk, const char ic_name, rd_kafka_interceptor_f_on_request_sent_t on_request_sent, void ic_opaque)
	Append an on_request_sent() interceptor. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_interceptor_add_on_response_received (rd_kafka_t rk, const char ic_name, rd_kafka_interceptor_f_on_response_received_t on_response_received, void ic_opaque)
	Append an on_response_received() interceptor. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_interceptor_add_on_thread_start (rd_kafka_t rk, const char ic_name, rd_kafka_interceptor_f_on_thread_start_t on_thread_start, void ic_opaque)
	Append an on_thread_start() interceptor. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_interceptor_add_on_thread_exit (rd_kafka_t rk, const char ic_name, rd_kafka_interceptor_f_on_thread_exit_t on_thread_exit, void ic_opaque)
	Append an on_thread_exit() interceptor. More...

Admin API
The Admin API enables applications to perform administrative Apache Kafka tasks, such as creating and deleting topics, altering and reading broker configuration, etc. The Admin API is asynchronous and makes use of librdkafka's standard `rd_kafka_queue_t` queues to propagate the result of an admin operation back to the application. The supplied queue may be any queue, such as a temporary single-call queue, a shared queue used for multiple requests, or even the main queue or consumer queues. Use `rd_kafka_queue_poll()` to collect the result of an admin operation from the queue of your choice, then extract the admin API-specific result type by using the corresponding `rd_kafka_event_CreateTopics_result`, `rd_kafka_event_DescribeConfigs_result`, etc, methods. Use the getter methods on the .._result_t type to extract response information and finally destroy the result and event by calling `rd_kafka_event_destroy()`. Use rd_kafka_event_error() and rd_kafka_event_error_string() to acquire the request-level error/success for an Admin API request. Even if the returned value is `RD_KAFKA_RESP_ERR_NO_ERROR` there may be individual objects (topics, resources, etc) that have failed. Extract per-object error information with the corresponding `rd_kafka_`..._result_topics\|resources\|..() to check per-object errors. Locally triggered errors: `RD_KAFKA_RESP_ERR__TIMED_OUT` - (Controller) broker connection did not become available in the time allowed by AdminOption_set_request_timeout.
enum	rd_kafka_admin_op_t { RD_KAFKA_ADMIN_OP_ANY = 0, RD_KAFKA_ADMIN_OP_CREATETOPICS, RD_KAFKA_ADMIN_OP_DELETETOPICS, RD_KAFKA_ADMIN_OP_CREATEPARTITIONS, RD_KAFKA_ADMIN_OP_ALTERCONFIGS, RD_KAFKA_ADMIN_OP_DESCRIBECONFIGS, RD_KAFKA_ADMIN_OP_DELETERECORDS, RD_KAFKA_ADMIN_OP_DELETEGROUPS, RD_KAFKA_ADMIN_OP_DELETECONSUMERGROUPOFFSETS, RD_KAFKA_ADMIN_OP_CREATEACLS, RD_KAFKA_ADMIN_OP_DESCRIBEACLS, RD_KAFKA_ADMIN_OP_DELETEACLS, RD_KAFKA_ADMIN_OP__CNT }
	Admin operation enum name for use with rd_kafka_AdminOptions_new() More...

enum	rd_kafka_ConfigSource_t { RD_KAFKA_CONFIG_SOURCE_UNKNOWN_CONFIG = 0, RD_KAFKA_CONFIG_SOURCE_DYNAMIC_TOPIC_CONFIG = 1, RD_KAFKA_CONFIG_SOURCE_DYNAMIC_BROKER_CONFIG = 2, RD_KAFKA_CONFIG_SOURCE_DYNAMIC_DEFAULT_BROKER_CONFIG = 3, RD_KAFKA_CONFIG_SOURCE_STATIC_BROKER_CONFIG = 4, RD_KAFKA_CONFIG_SOURCE_DEFAULT_CONFIG = 5, RD_KAFKA_CONFIG_SOURCE__CNT }
	Apache Kafka config sources. More...

enum	rd_kafka_ResourceType_t { RD_KAFKA_RESOURCE_UNKNOWN = 0, RD_KAFKA_RESOURCE_ANY = 1, RD_KAFKA_RESOURCE_TOPIC = 2, RD_KAFKA_RESOURCE_GROUP = 3, RD_KAFKA_RESOURCE_BROKER = 4, RD_KAFKA_RESOURCE__CNT }
	Apache Kafka resource types. More...

enum	rd_kafka_ResourcePatternType_t { RD_KAFKA_RESOURCE_PATTERN_UNKNOWN = 0, RD_KAFKA_RESOURCE_PATTERN_ANY = 1, RD_KAFKA_RESOURCE_PATTERN_MATCH = 2, RD_KAFKA_RESOURCE_PATTERN_LITERAL = 3, RD_KAFKA_RESOURCE_PATTERN_PREFIXED = 4, RD_KAFKA_RESOURCE_PATTERN_TYPE__CNT }
	Apache Kafka pattern types. More...

typedef struct rd_kafka_AdminOptions_s	rd_kafka_AdminOptions_t
	AdminOptions provides a generic mechanism for setting optional parameters for the Admin API requests. More...

typedef struct rd_kafka_NewTopic_s	rd_kafka_NewTopic_t

typedef struct rd_kafka_DeleteTopic_s	rd_kafka_DeleteTopic_t

typedef struct rd_kafka_NewPartitions_s	rd_kafka_NewPartitions_t

typedef struct rd_kafka_ConfigEntry_s	rd_kafka_ConfigEntry_t

typedef struct rd_kafka_ConfigResource_s	rd_kafka_ConfigResource_t

typedef struct rd_kafka_DeleteRecords_s	rd_kafka_DeleteRecords_t

typedef struct rd_kafka_DeleteGroup_s	rd_kafka_DeleteGroup_t

typedef struct rd_kafka_DeleteConsumerGroupOffsets_s	rd_kafka_DeleteConsumerGroupOffsets_t

typedef struct rd_kafka_AclBinding_s	rd_kafka_AclBinding_t
	ACL Binding is used to create access control lists. More...

typedef rd_kafka_AclBinding_t	rd_kafka_AclBindingFilter_t
	ACL Binding filter is used to filter access control lists. More...

RD_EXPORT rd_kafka_AdminOptions_t *	rd_kafka_AdminOptions_new (rd_kafka_t *rk, rd_kafka_admin_op_t for_api)
	Create a new AdminOptions object. More...

RD_EXPORT void	rd_kafka_AdminOptions_destroy (rd_kafka_AdminOptions_t *options)
	Destroy a AdminOptions object.

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_AdminOptions_set_request_timeout (rd_kafka_AdminOptions_t options, int timeout_ms, char errstr, size_t errstr_size)
	Sets the overall request timeout, including broker lookup, request transmission, operation time on broker, and response. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_AdminOptions_set_operation_timeout (rd_kafka_AdminOptions_t options, int timeout_ms, char errstr, size_t errstr_size)
	Sets the broker's operation timeout, such as the timeout for CreateTopics to complete the creation of topics on the controller before returning a result to the application. More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_AdminOptions_set_validate_only (rd_kafka_AdminOptions_t options, int true_or_false, char errstr, size_t errstr_size)
	Tell broker to only validate the request, without performing the requested operation (create topics, etc). More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_AdminOptions_set_broker (rd_kafka_AdminOptions_t options, int32_t broker_id, char errstr, size_t errstr_size)
	Override what broker the Admin request will be sent to. More...

RD_EXPORT void	rd_kafka_AdminOptions_set_opaque (rd_kafka_AdminOptions_t options, void ev_opaque)
	Set application opaque value that can be extracted from the result event using rd_kafka_event_opaque()

RD_EXPORT rd_kafka_NewTopic_t *	rd_kafka_NewTopic_new (const char topic, int num_partitions, int replication_factor, char errstr, size_t errstr_size)
	Create a new NewTopic object. This object is later passed to rd_kafka_CreateTopics(). More...

RD_EXPORT void	rd_kafka_NewTopic_destroy (rd_kafka_NewTopic_t *new_topic)
	Destroy and free a NewTopic object previously created with rd_kafka_NewTopic_new()

RD_EXPORT void	rd_kafka_NewTopic_destroy_array (rd_kafka_NewTopic_t **new_topics, size_t new_topic_cnt)
	Helper function to destroy all NewTopic objects in the `new_topics` array (of `new_topic_cnt` elements). The array itself is not freed.

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_NewTopic_set_replica_assignment (rd_kafka_NewTopic_t new_topic, int32_t partition, int32_t broker_ids, size_t broker_id_cnt, char *errstr, size_t errstr_size)
	Set the replica (broker) assignment for `partition` to the replica set in `broker_ids` (of `broker_id_cnt` elements). More...

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_NewTopic_set_config (rd_kafka_NewTopic_t new_topic, const char name, const char *value)
	Set (broker-side) topic configuration name/value pair. More...

RD_EXPORT void	rd_kafka_CreateTopics (rd_kafka_t rk, rd_kafka_NewTopic_t new_topics, size_t new_topic_cnt, const rd_kafka_AdminOptions_t options, rd_kafka_queue_t *rkqu)
	Create topics in cluster as specified by the `new_topics` array of size `new_topic_cnt` elements. More...

const RD_EXPORT rd_kafka_topic_result_t **	rd_kafka_CreateTopics_result_topics (const rd_kafka_CreateTopics_result_t result, size_t cntp)
	Get an array of topic results from a CreateTopics result. More...

RD_EXPORT rd_kafka_DeleteTopic_t *	rd_kafka_DeleteTopic_new (const char *topic)
	Create a new DeleteTopic object. This object is later passed to rd_kafka_DeleteTopics(). More...

RD_EXPORT void	rd_kafka_DeleteTopic_destroy (rd_kafka_DeleteTopic_t *del_topic)
	Destroy and free a DeleteTopic object previously created with rd_kafka_DeleteTopic_new()

RD_EXPORT void	rd_kafka_DeleteTopic_destroy_array (rd_kafka_DeleteTopic_t **del_topics, size_t del_topic_cnt)
	Helper function to destroy all DeleteTopic objects in the `del_topics` array (of `del_topic_cnt` elements). The array itself is not freed.

RD_EXPORT void	rd_kafka_DeleteTopics (rd_kafka_t rk, rd_kafka_DeleteTopic_t del_topics, size_t del_topic_cnt, const rd_kafka_AdminOptions_t options, rd_kafka_queue_t *rkqu)
	Delete topics from cluster as specified by the `topics` array of size `topic_cnt` elements. More...

const RD_EXPORT rd_kafka_topic_result_t **	rd_kafka_DeleteTopics_result_topics (const rd_kafka_DeleteTopics_result_t result, size_t cntp)
	Get an array of topic results from a DeleteTopics result. More...

RD_EXPORT rd_kafka_NewPartitions_t *	rd_kafka_NewPartitions_new (const char topic, size_t new_total_cnt, char errstr, size_t errstr_size)
	Create a new NewPartitions. This object is later passed to rd_kafka_CreatePartitions() to increase the number of partitions to `new_total_cnt` for an existing topic. More...

RD_EXPORT void	rd_kafka_NewPartitions_destroy (rd_kafka_NewPartitions_t *new_parts)
	Destroy and free a NewPartitions object previously created with rd_kafka_NewPartitions_new()

RD_EXPORT void	rd_kafka_NewPartitions_destroy_array (rd_kafka_NewPartitions_t **new_parts, size_t new_parts_cnt)
	Helper function to destroy all NewPartitions objects in the `new_parts` array (of `new_parts_cnt` elements). The array itself is not freed.

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_NewPartitions_set_replica_assignment (rd_kafka_NewPartitions_t new_parts, int32_t new_partition_idx, int32_t broker_ids, size_t broker_id_cnt, char *errstr, size_t errstr_size)
	Set the replica (broker id) assignment for `new_partition_idx` to the replica set in `broker_ids` (of `broker_id_cnt` elements). More...

RD_EXPORT void	rd_kafka_CreatePartitions (rd_kafka_t rk, rd_kafka_NewPartitions_t new_parts, size_t new_parts_cnt, const rd_kafka_AdminOptions_t options, rd_kafka_queue_t *rkqu)
	Create additional partitions for the given topics, as specified by the `new_parts` array of size `new_parts_cnt` elements. More...

const RD_EXPORT rd_kafka_topic_result_t **	rd_kafka_CreatePartitions_result_topics (const rd_kafka_CreatePartitions_result_t result, size_t cntp)
	Get an array of topic results from a CreatePartitions result. More...

const RD_EXPORT char *	rd_kafka_ConfigSource_name (rd_kafka_ConfigSource_t confsource)

const RD_EXPORT char *	rd_kafka_ConfigEntry_name (const rd_kafka_ConfigEntry_t *entry)

const RD_EXPORT char *	rd_kafka_ConfigEntry_value (const rd_kafka_ConfigEntry_t *entry)

RD_EXPORT rd_kafka_ConfigSource_t	rd_kafka_ConfigEntry_source (const rd_kafka_ConfigEntry_t *entry)

RD_EXPORT int	rd_kafka_ConfigEntry_is_read_only (const rd_kafka_ConfigEntry_t *entry)

RD_EXPORT int	rd_kafka_ConfigEntry_is_default (const rd_kafka_ConfigEntry_t *entry)

RD_EXPORT int	rd_kafka_ConfigEntry_is_sensitive (const rd_kafka_ConfigEntry_t *entry)

RD_EXPORT int	rd_kafka_ConfigEntry_is_synonym (const rd_kafka_ConfigEntry_t *entry)

const RD_EXPORT rd_kafka_ConfigEntry_t **	rd_kafka_ConfigEntry_synonyms (const rd_kafka_ConfigEntry_t entry, size_t cntp)

const RD_EXPORT char *	rd_kafka_ResourcePatternType_name (rd_kafka_ResourcePatternType_t resource_pattern_type)

const RD_EXPORT char *	rd_kafka_ResourceType_name (rd_kafka_ResourceType_t restype)

RD_EXPORT rd_kafka_ConfigResource_t *	rd_kafka_ConfigResource_new (rd_kafka_ResourceType_t restype, const char *resname)
	Create new ConfigResource object. More...

RD_EXPORT void	rd_kafka_ConfigResource_destroy (rd_kafka_ConfigResource_t *config)
	Destroy and free a ConfigResource object previously created with rd_kafka_ConfigResource_new()

RD_EXPORT void	rd_kafka_ConfigResource_destroy_array (rd_kafka_ConfigResource_t **config, size_t config_cnt)
	Helper function to destroy all ConfigResource objects in the `configs` array (of `config_cnt` elements). The array itself is not freed.

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_ConfigResource_set_config (rd_kafka_ConfigResource_t config, const char name, const char *value)
	Set configuration name value pair. More...

const RD_EXPORT rd_kafka_ConfigEntry_t **	rd_kafka_ConfigResource_configs (const rd_kafka_ConfigResource_t config, size_t cntp)
	Get an array of config entries from a ConfigResource object. More...

RD_EXPORT rd_kafka_ResourceType_t	rd_kafka_ConfigResource_type (const rd_kafka_ConfigResource_t *config)

const RD_EXPORT char *	rd_kafka_ConfigResource_name (const rd_kafka_ConfigResource_t *config)

RD_EXPORT rd_kafka_resp_err_t	rd_kafka_ConfigResource_error (const rd_kafka_ConfigResource_t *config)

const RD_EXPORT char *	rd_kafka_ConfigResource_error_string (const rd_kafka_ConfigResource_t *config)

RD_EXPORT void	rd_kafka_AlterConfigs (rd_kafka_t rk, rd_kafka_ConfigResource_t configs, size_t config_cnt, const rd_kafka_AdminOptions_t options, rd_kafka_queue_t *rkqu)
	Update the configuration for the specified resources. Updates are not transactional so they may succeed for a subset of the provided resources while the others fail. The configuration for a particular resource is updated atomically, replacing values using the provided ConfigEntrys and reverting unspecified ConfigEntrys to their default values. More...

const RD_EXPORT rd_kafka_ConfigResource_t **	rd_kafka_AlterConfigs_result_resources (const rd_kafka_AlterConfigs_result_t result, size_t cntp)
	Get an array of resource results from a AlterConfigs result. More...

RD_EXPORT void	rd_kafka_DescribeConfigs (rd_kafka_t rk, rd_kafka_ConfigResource_t configs, size_t config_cnt, const rd_kafka_AdminOptions_t options, rd_kafka_queue_t *rkqu)
	Get configuration for the specified resources in `configs`. More...

const RD_EXPORT rd_kafka_ConfigResource_t **	rd_kafka_DescribeConfigs_result_resources (const rd_kafka_DescribeConfigs_result_t result, size_t cntp)
	Get an array of resource results from a DescribeConfigs result. More...

RD_EXPORT rd_kafka_DeleteRecords_t *	rd_kafka_DeleteRecords_new (const rd_kafka_topic_partition_list_t *before_offsets)
	Create a new DeleteRecords object. This object is later passed to rd_kafka_DeleteRecords(). More...

RD_EXPORT void	rd_kafka_DeleteRecords_destroy (rd_kafka_DeleteRecords_t *del_records)
	Destroy and free a DeleteRecords object previously created with rd_kafka_DeleteRecords_new()

RD_EXPORT void	rd_kafka_DeleteRecords_destroy_array (rd_kafka_DeleteRecords_t **del_records, size_t del_record_cnt)
	Helper function to destroy all DeleteRecords objects in the `del_groups` array (of `del_group_cnt` elements). The array itself is not freed.

RD_EXPORT void	rd_kafka_DeleteRecords (rd_kafka_t rk, rd_kafka_DeleteRecords_t del_records, size_t del_record_cnt, const rd_kafka_AdminOptions_t options, rd_kafka_queue_t *rkqu)
	Delete records (messages) in topic partitions older than the offsets provided. More...

const RD_EXPORT rd_kafka_topic_partition_list_t *	rd_kafka_DeleteRecords_result_offsets (const rd_kafka_DeleteRecords_result_t *result)
	Get a list of topic and partition results from a DeleteRecords result. The returned objects will contain `topic`, `partition`, `offset` and `err`. `offset` will be set to the post-deletion low-watermark (smallest available offset of all live replicas). `err` will be set per-partition if deletion failed. More...

RD_EXPORT rd_kafka_DeleteGroup_t *	rd_kafka_DeleteGroup_new (const char *group)
	Create a new DeleteGroup object. This object is later passed to rd_kafka_DeleteGroups(). More...

RD_EXPORT void	rd_kafka_DeleteGroup_destroy (rd_kafka_DeleteGroup_t *del_group)
	Destroy and free a DeleteGroup object previously created with rd_kafka_DeleteGroup_new()

RD_EXPORT void	rd_kafka_DeleteGroup_destroy_array (rd_kafka_DeleteGroup_t **del_groups, size_t del_group_cnt)
	Helper function to destroy all DeleteGroup objects in the `del_groups` array (of `del_group_cnt` elements). The array itself is not freed.

RD_EXPORT void	rd_kafka_DeleteGroups (rd_kafka_t rk, rd_kafka_DeleteGroup_t del_groups, size_t del_group_cnt, const rd_kafka_AdminOptions_t options, rd_kafka_queue_t *rkqu)
	Delete groups from cluster as specified by the `del_groups` array of size `del_group_cnt` elements. More...

const RD_EXPORT rd_kafka_group_result_t **	rd_kafka_DeleteGroups_result_groups (const rd_kafka_DeleteGroups_result_t result, size_t cntp)
	Get an array of group results from a DeleteGroups result. More...

RD_EXPORT rd_kafka_DeleteConsumerGroupOffsets_t *	rd_kafka_DeleteConsumerGroupOffsets_new (const char group, const rd_kafka_topic_partition_list_t partitions)
	Create a new DeleteConsumerGroupOffsets object. This object is later passed to rd_kafka_DeleteConsumerGroupOffsets(). More...

RD_EXPORT void	rd_kafka_DeleteConsumerGroupOffsets_destroy (rd_kafka_DeleteConsumerGroupOffsets_t *del_grpoffsets)
	Destroy and free a DeleteConsumerGroupOffsets object previously created with rd_kafka_DeleteConsumerGroupOffsets_new()

RD_EXPORT void	rd_kafka_DeleteConsumerGroupOffsets_destroy_array (rd_kafka_DeleteConsumerGroupOffsets_t **del_grpoffsets, size_t del_grpoffset_cnt)
	Helper function to destroy all DeleteConsumerGroupOffsets objects in the `del_grpoffsets` array (of `del_grpoffsets_cnt` elements). The array itself is not freed.

RD_EXPORT void	rd_kafka_DeleteConsumerGroupOffsets (rd_kafka_t rk, rd_kafka_DeleteConsumerGroupOffsets_t del_grpoffsets, size_t del_grpoffsets_cnt, const rd_kafka_AdminOptions_t options, rd_kafka_queue_t *rkqu)
	Delete committed offsets for a set of partitions in a conusmer group. This will succeed at the partition level only if the group is not actively subscribed to the corresponding topic. More...

const RD_EXPORT rd_kafka_group_result_t **	rd_kafka_DeleteConsumerGroupOffsets_result_groups (const rd_kafka_DeleteConsumerGroupOffsets_result_t result, size_t cntp)
	Get an array of results from a DeleteConsumerGroupOffsets result. More...

const RD_EXPORT rd_kafka_error_t *	rd_kafka_acl_result_error (const rd_kafka_acl_result_t *aclres)

AclOperation
enum	rd_kafka_AclOperation_t { RD_KAFKA_ACL_OPERATION_UNKNOWN = 0, RD_KAFKA_ACL_OPERATION_ANY, RD_KAFKA_ACL_OPERATION_ALL = 2, RD_KAFKA_ACL_OPERATION_READ = 3, RD_KAFKA_ACL_OPERATION_WRITE = 4, RD_KAFKA_ACL_OPERATION_CREATE = 5, RD_KAFKA_ACL_OPERATION_DELETE = 6, RD_KAFKA_ACL_OPERATION_ALTER = 7, RD_KAFKA_ACL_OPERATION_DESCRIBE = 8, RD_KAFKA_ACL_OPERATION_CLUSTER_ACTION, RD_KAFKA_ACL_OPERATION_DESCRIBE_CONFIGS, RD_KAFKA_ACL_OPERATION_ALTER_CONFIGS, RD_KAFKA_ACL_OPERATION_IDEMPOTENT_WRITE, RD_KAFKA_ACL_OPERATION__CNT }
	Apache Kafka ACL operation types. More...

const RD_EXPORT char *	rd_kafka_AclOperation_name (rd_kafka_AclOperation_t acl_operation)

AclPermissionType
enum	rd_kafka_AclPermissionType_t { RD_KAFKA_ACL_PERMISSION_TYPE_UNKNOWN = 0, RD_KAFKA_ACL_PERMISSION_TYPE_ANY, RD_KAFKA_ACL_PERMISSION_TYPE_DENY = 2, RD_KAFKA_ACL_PERMISSION_TYPE_ALLOW = 3, RD_KAFKA_ACL_PERMISSION_TYPE__CNT }
	Apache Kafka ACL permission types. More...

const RD_EXPORT char *	rd_kafka_AclPermissionType_name (rd_kafka_AclPermissionType_t acl_permission_type)

Detailed Description

Apache Kafka C/C++ consumer and producer client library.

rdkafka.h contains the public API for librdkafka. The API is documented in this file as comments prefixing the function, type, enum, define, etc.

See also: For the C++ interface see rdkafkacpp.h

Macro Definition Documentation

◆ RD_KAFKA_VERSION

#define RD_KAFKA_VERSION 0x010900ff

librdkafka version

Interpreted as hex MM.mm.rr.xx:

MM = Major
mm = minor
rr = revision
xx = pre-release id (0xff is the final release)

E.g.: 0x000801ff = 0.8.1

Remarks: This value should only be used during compile time, for runtime checks of version use rd_kafka_version()

◆ RD_KAFKA_DEBUG_CONTEXTS

#define RD_KAFKA_DEBUG_CONTEXTS

Value:

        "all,generic,broker,topic,metadata,feature,queue,msg,protocol,cgrp,"   \
        "security,fetch,interceptor,plugin,consumer,admin,eos,mock,assignor,"  \
        "conf"

Supported debug contexts. (compile time)

Deprecated:: This compile time value may be outdated at runtime due to linking another version of the library. Use rd_kafka_get_debug_contexts() instead.

◆ RD_KAFKA_RESP_ERR_GROUP_LOAD_IN_PROGRESS

#define RD_KAFKA_RESP_ERR_GROUP_LOAD_IN_PROGRESS RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS

Group coordinator load in progress

◆ RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE

#define RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE

Group coordinator not available

◆ RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP

#define RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP RD_KAFKA_RESP_ERR_NOT_COORDINATOR

Not coordinator for group

◆ RD_KAFKA_V_END

#define RD_KAFKA_V_END RD_KAFKA_VTYPE_END

Convenience macros for rd_kafka_vtype_t that takes the correct arguments for each vtype.

va-arg end sentinel used to terminate the variable argument list

◆ RD_KAFKA_V_TOPIC

#define RD_KAFKA_V_TOPIC ( topic )

Value:

_LRK_TYPECHECK(RD_KAFKA_VTYPE_TOPIC, const char *, topic), \

(const char *)topic

Topic name (const char *)

rd_kafka_vu_t field: u.cstr

◆ RD_KAFKA_V_RKT

#define RD_KAFKA_V_RKT ( rkt )

Value:

_LRK_TYPECHECK(RD_KAFKA_VTYPE_RKT, rd_kafka_topic_t *, rkt), \

(rd_kafka_topic_t *)rkt

Topic object (rd_kafka_topic_t *)

rd_kafka_vu_t field: u.rkt

◆ RD_KAFKA_V_PARTITION

#define RD_KAFKA_V_PARTITION ( partition )

Value:

_LRK_TYPECHECK(RD_KAFKA_VTYPE_PARTITION, int32_t, partition), \

(int32_t)partition

Partition (int32_t)

rd_kafka_vu_t field: u.i32

◆ RD_KAFKA_V_VALUE

#define RD_KAFKA_V_VALUE	(	VALUE,
		LEN
	)

Value:

_LRK_TYPECHECK2(RD_KAFKA_VTYPE_VALUE, void *, VALUE, size_t, LEN), \

(void *)VALUE, (size_t)LEN

Message value/payload pointer and length (void *, size_t)

rd_kafka_vu_t fields: u.mem.ptr, u.mem.size

◆ RD_KAFKA_V_KEY

#define RD_KAFKA_V_KEY	(	KEY,
		LEN
	)

Value:

_LRK_TYPECHECK2(RD_KAFKA_VTYPE_KEY, const void *, KEY, size_t, LEN), \

(void *)KEY, (size_t)LEN

Message key pointer and length (const void *, size_t)

rd_kafka_vu_t field: u.mem.ptr, rd_kafka_vu.t.u.mem.size

◆ RD_KAFKA_V_OPAQUE

#define RD_KAFKA_V_OPAQUE ( msg_opaque )

Value:

_LRK_TYPECHECK(RD_KAFKA_VTYPE_OPAQUE, void *, msg_opaque), \

(void *)msg_opaque

Message opaque pointer (void *) Same as msg_opaque, produce(.., msg_opaque), and rkmessage->_private .

rd_kafka_vu_t field: u.ptr

◆ RD_KAFKA_V_MSGFLAGS

#define RD_KAFKA_V_MSGFLAGS ( msgflags ) _LRK_TYPECHECK(RD_KAFKA_VTYPE_MSGFLAGS, int, msgflags), (int)msgflags

Message flags (int)

See also: RD_KAFKA_MSG_F_COPY, et.al.

rd_kafka_vu_t field: u.i

◆ RD_KAFKA_V_TIMESTAMP

#define RD_KAFKA_V_TIMESTAMP ( timestamp )

Value:

_LRK_TYPECHECK(RD_KAFKA_VTYPE_TIMESTAMP, int64_t, timestamp), \

(int64_t)timestamp

Timestamp in milliseconds since epoch UTC (int64_t). A value of 0 will use the current wall-clock time.

rd_kafka_vu_t field: u.i64

◆ RD_KAFKA_V_HEADER

#define RD_KAFKA_V_HEADER	(	NAME,
		VALUE,
		LEN
	)

Value:

        _LRK_TYPECHECK3(RD_KAFKA_VTYPE_HEADER, const char *, NAME,             \
                        const void *, VALUE, ssize_t, LEN),                    \
            (const char *)NAME, (const void *)VALUE, (ssize_t)LEN

Add Message Header (const char *NAME, const void *VALUE, ssize_t LEN).

See also: rd_kafka_header_add()

Remarks: RD_KAFKA_V_HEADER() and RD_KAFKA_V_HEADERS() MUST NOT be mixed in the same call to producev().

rd_kafka_vu_t fields: u.header.name, u.header.val, u.header.size

◆ RD_KAFKA_V_HEADERS

#define RD_KAFKA_V_HEADERS ( HDRS )

Value:

_LRK_TYPECHECK(RD_KAFKA_VTYPE_HEADERS, rd_kafka_headers_t *, HDRS), \

(rd_kafka_headers_t *)HDRS

Message Headers list (rd_kafka_headers_t *). The message object will assume ownership of the headers (unless producev() fails). Any existing headers will be replaced.

See also: rd_kafka_message_set_headers()

Remarks: RD_KAFKA_V_HEADER() and RD_KAFKA_V_HEADERS() MUST NOT be mixed in the same call to producev().

rd_kafka_vu_t fields: u.headers

◆ RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE

#define RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE 0x8

Flags for rd_kafka_destroy_flags()

Don't call consumer_close() to leave group and commit final offsets.

This also disables consumer callbacks to be called from rd_kafka_destroy*(), such as rebalance_cb.

The consumer group handler is still closed internally, but from an application perspective none of the functionality from consumer_close() is performed.

◆ RD_KAFKA_PARTITION_UA

#define RD_KAFKA_PARTITION_UA ((int32_t)-1)

Unassigned partition.

The unassigned partition is used by the producer API for messages that should be partitioned using the configured or default partitioner.

◆ RD_KAFKA_OFFSET_BEGINNING

#define RD_KAFKA_OFFSET_BEGINNING -2

Start consuming from beginning of \ kafka partition queue: oldest msg

◆ RD_KAFKA_OFFSET_END

#define RD_KAFKA_OFFSET_END -1

Start consuming from end of kafka \ partition queue: next msg

◆ RD_KAFKA_OFFSET_STORED

#define RD_KAFKA_OFFSET_STORED -1000

Start consuming from offset retrieved \ from offset store

◆ RD_KAFKA_OFFSET_INVALID

#define RD_KAFKA_OFFSET_INVALID -1001

Invalid offset

◆ RD_KAFKA_OFFSET_TAIL

#define RD_KAFKA_OFFSET_TAIL ( CNT ) (RD_KAFKA_OFFSET_TAIL_BASE - (CNT))

Start consuming CNT messages from topic's current end offset.

That is, if current end offset is 12345 and CNT is 200, it will start consuming from offset 12345-200 = 12145.

◆ RD_KAFKA_MSG_F_FREE

#define RD_KAFKA_MSG_F_FREE 0x1

Producer message flags.

Delegate freeing of payload to rdkafka. \

◆ RD_KAFKA_MSG_F_COPY

#define RD_KAFKA_MSG_F_COPY 0x2

rdkafka will make a copy of the payload. \

◆ RD_KAFKA_MSG_F_BLOCK

#define RD_KAFKA_MSG_F_BLOCK 0x4

Block produce*() on message queue full. \ WARNING: If a delivery report callback \ is used the application MUST \ call rd_kafka_poll() (or equiv.) \ to make sure delivered messages \ are drained from the internal \ delivery report queue. \ Failure to do so will result \ in indefinately blocking on \ the produce() call when the \ message queue is full.

◆ RD_KAFKA_MSG_F_PARTITION

#define RD_KAFKA_MSG_F_PARTITION 0x8

produce_batch() will honor \ per-message partition.

◆ RD_KAFKA_PURGE_F_QUEUE

#define RD_KAFKA_PURGE_F_QUEUE 0x1

Flags for rd_kafka_purge()

Purge messages in internal queues.

◆ RD_KAFKA_PURGE_F_INFLIGHT

#define RD_KAFKA_PURGE_F_INFLIGHT 0x2

Purge messages in-flight to or from the broker. Purging these messages will void any future acknowledgements from the broker, making it impossible for the application to know if these messages were successfully delivered or not. Retrying these messages may lead to duplicates.

◆ RD_KAFKA_PURGE_F_NON_BLOCKING

#define RD_KAFKA_PURGE_F_NON_BLOCKING 0x4

Don't wait for background thread queue purging to finish.

◆ RD_KAFKA_EVENT_NONE

#define RD_KAFKA_EVENT_NONE 0x0

Unset value

◆ RD_KAFKA_EVENT_DR

#define RD_KAFKA_EVENT_DR 0x1

Producer Delivery report batch

◆ RD_KAFKA_EVENT_FETCH

#define RD_KAFKA_EVENT_FETCH 0x2

Fetched message (consumer)

◆ RD_KAFKA_EVENT_LOG

#define RD_KAFKA_EVENT_LOG 0x4

Log message

◆ RD_KAFKA_EVENT_ERROR

#define RD_KAFKA_EVENT_ERROR 0x8

Error

◆ RD_KAFKA_EVENT_REBALANCE

#define RD_KAFKA_EVENT_REBALANCE 0x10

Group rebalance (consumer)

◆ RD_KAFKA_EVENT_OFFSET_COMMIT

#define RD_KAFKA_EVENT_OFFSET_COMMIT 0x20

Offset commit result

◆ RD_KAFKA_EVENT_STATS

#define RD_KAFKA_EVENT_STATS 0x40

Stats

◆ RD_KAFKA_EVENT_CREATETOPICS_RESULT

#define RD_KAFKA_EVENT_CREATETOPICS_RESULT 100

CreateTopics_result_t

◆ RD_KAFKA_EVENT_DELETETOPICS_RESULT

#define RD_KAFKA_EVENT_DELETETOPICS_RESULT 101

DeleteTopics_result_t

◆ RD_KAFKA_EVENT_CREATEPARTITIONS_RESULT

#define RD_KAFKA_EVENT_CREATEPARTITIONS_RESULT 102

CreatePartitions_result_t

◆ RD_KAFKA_EVENT_ALTERCONFIGS_RESULT

#define RD_KAFKA_EVENT_ALTERCONFIGS_RESULT 103

AlterConfigs_result_t

◆ RD_KAFKA_EVENT_DESCRIBECONFIGS_RESULT

#define RD_KAFKA_EVENT_DESCRIBECONFIGS_RESULT 104

DescribeConfigs_result_t

◆ RD_KAFKA_EVENT_DELETERECORDS_RESULT

#define RD_KAFKA_EVENT_DELETERECORDS_RESULT 105

DeleteRecords_result_t

◆ RD_KAFKA_EVENT_DELETEGROUPS_RESULT

#define RD_KAFKA_EVENT_DELETEGROUPS_RESULT 106

DeleteGroups_result_t

◆ RD_KAFKA_EVENT_DELETECONSUMERGROUPOFFSETS_RESULT

#define RD_KAFKA_EVENT_DELETECONSUMERGROUPOFFSETS_RESULT 107

DeleteConsumerGroupOffsets_result_t

◆ RD_KAFKA_EVENT_OAUTHBEARER_TOKEN_REFRESH

#define RD_KAFKA_EVENT_OAUTHBEARER_TOKEN_REFRESH 0x100

SASL/OAUTHBEARER token needs to be refreshed

◆ RD_KAFKA_EVENT_BACKGROUND

#define RD_KAFKA_EVENT_BACKGROUND 0x200

Enable background thread.

◆ RD_KAFKA_EVENT_CREATEACLS_RESULT

#define RD_KAFKA_EVENT_CREATEACLS_RESULT 0x400

CreateAcls_result_t

◆ RD_KAFKA_EVENT_DESCRIBEACLS_RESULT

#define RD_KAFKA_EVENT_DESCRIBEACLS_RESULT 0x800

DescribeAcls_result_t

◆ RD_KAFKA_EVENT_DELETEACLS_RESULT

#define RD_KAFKA_EVENT_DELETEACLS_RESULT 0x1000

DeleteAcls_result_t

Typedef Documentation

◆ rd_kafka_CreateTopics_result_t

typedef rd_kafka_event_t rd_kafka_CreateTopics_result_t

CreateTopics result type

◆ rd_kafka_DeleteTopics_result_t

typedef rd_kafka_event_t rd_kafka_DeleteTopics_result_t

DeleteTopics result type

◆ rd_kafka_CreateAcls_result_t

typedef rd_kafka_event_t rd_kafka_CreateAcls_result_t

CreateAcls result type

◆ rd_kafka_DescribeAcls_result_t

typedef rd_kafka_event_t rd_kafka_DescribeAcls_result_t

DescribeAcls result type

◆ rd_kafka_DeleteAcls_result_t

typedef rd_kafka_event_t rd_kafka_DeleteAcls_result_t

DeleteAcls result type

◆ rd_kafka_CreatePartitions_result_t

typedef rd_kafka_event_t rd_kafka_CreatePartitions_result_t

CreatePartitions result type

◆ rd_kafka_AlterConfigs_result_t

typedef rd_kafka_event_t rd_kafka_AlterConfigs_result_t

AlterConfigs result type

◆ rd_kafka_DescribeConfigs_result_t

typedef rd_kafka_event_t rd_kafka_DescribeConfigs_result_t

CreateTopics result type

◆ rd_kafka_DeleteRecords_result_t

typedef rd_kafka_event_t rd_kafka_DeleteRecords_result_t

DeleteRecords result type

◆ rd_kafka_DeleteGroups_result_t

typedef rd_kafka_event_t rd_kafka_DeleteGroups_result_t

DeleteGroups result type

◆ rd_kafka_DeleteConsumerGroupOffsets_result_t

typedef rd_kafka_event_t rd_kafka_DeleteConsumerGroupOffsets_result_t

DeleteConsumerGroupOffsets result type

◆ rd_kafka_plugin_f_conf_init_t

typedef rd_kafka_resp_err_t() rd_kafka_plugin_f_conf_init_t(rd_kafka_conf_t *conf, void **plug_opaquep, char *errstr, size_t errstr_size)

Plugin's configuration initializer method called each time the library is referenced from configuration (even if previously loaded by another client instance).

Remarks: This method MUST be implemented by plugins and have the symbol name conf_init

Parameters

conf	Configuration set up to this point.
plug_opaquep	Plugin can set this pointer to a per-configuration opaque pointer.
errstr	String buffer of size `errstr_size` where plugin must write a human readable error string in the case the initializer fails (returns non-zero).
errstr_size	Maximum space (including \0) in `errstr`.

Remarks: A plugin may add an on_conf_destroy() interceptor to clean up plugin-specific resources created in the plugin's conf_init() method.

Returns: RD_KAFKA_RESP_ERR_NO_ERROR on success or an error code on error.

◆ rd_kafka_interceptor_f_on_conf_set_t

typedef rd_kafka_conf_res_t() rd_kafka_interceptor_f_on_conf_set_t(rd_kafka_conf_t *conf, const char *name, const char *val, char *errstr, size_t errstr_size, void *ic_opaque)

on_conf_set() is called from rd_kafka_*_conf_set() in the order the interceptors were added.

Parameters

conf	Configuration object.
ic_opaque	The interceptor's opaque pointer specified in ..add..().
name	The configuration property to set.
val	The configuration value to set, or NULL for reverting to default in which case the previous value should be freed.
errstr	A human readable error string in case the interceptor fails.
errstr_size	Maximum space (including \0) in `errstr`.

Returns: RD_KAFKA_CONF_OK if the property was known and successfully handled by the interceptor, RD_KAFKA_CONF_INVALID if the property was handled by the interceptor but the value was invalid, or RD_KAFKA_CONF_UNKNOWN if the interceptor did not handle this property, in which case the property is passed on on the interceptor in the chain, finally ending up at the built-in configuration handler.

◆ rd_kafka_interceptor_f_on_conf_dup_t

typedef rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_conf_dup_t(rd_kafka_conf_t *new_conf, const rd_kafka_conf_t *old_conf, size_t filter_cnt, const char **filter, void *ic_opaque)

on_conf_dup() is called from rd_kafka_conf_dup() in the order the interceptors were added and is used to let an interceptor re-register its conf interecptors with a new opaque value. The on_conf_dup() method is called prior to the configuration from old_conf being copied to new_conf.

Parameters

ic_opaque	The interceptor's opaque pointer specified in ..add..().
new_conf	New configuration object.
old_conf	Old configuration object to copy properties from.
filter_cnt	Number of property names to filter in `filter`.
filter	Property names to filter out (ignore) when setting up `new_conf`.

Returns: RD_KAFKA_RESP_ERR_NO_ERROR on success or an error code on failure (which is logged but otherwise ignored).

Remarks: No on_conf_* interceptors are copied to the new configuration object on rd_kafka_conf_dup().

◆ rd_kafka_interceptor_f_on_conf_destroy_t

typedef rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_conf_destroy_t(void *ic_opaque)

on_conf_destroy() is called from rd_kafka_*_conf_destroy() in the order the interceptors were added.

Parameters

ic_opaque The interceptor's opaque pointer specified in ..add..().

◆ rd_kafka_interceptor_f_on_new_t

typedef rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_new_t(rd_kafka_t *rk, const rd_kafka_conf_t *conf, void *ic_opaque, char *errstr, size_t errstr_size)

on_new() is called from rd_kafka_new() prior toreturning the newly created client instance to the application.

Parameters

rk	The client instance.
conf	The client instance's final configuration.
ic_opaque	The interceptor's opaque pointer specified in ..add..().
errstr	A human readable error string in case the interceptor fails.
errstr_size	Maximum space (including \0) in `errstr`.

Returns: an error code on failure, the error is logged but otherwise ignored.

Warning: The rk client instance will not be fully set up when this interceptor is called and the interceptor MUST NOT call any other rk-specific APIs than rd_kafka_interceptor_add..().

◆ rd_kafka_interceptor_f_on_destroy_t

typedef rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_destroy_t(rd_kafka_t *rk, void *ic_opaque)

on_destroy() is called from rd_kafka_destroy() or (rd_kafka_new() if rd_kafka_new() fails during initialization).

Parameters

rk	The client instance.
ic_opaque	The interceptor's opaque pointer specified in ..add..().

◆ rd_kafka_interceptor_f_on_send_t

typedef rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_send_t(rd_kafka_t *rk, rd_kafka_message_t *rkmessage, void *ic_opaque)

on_send() is called from rd_kafka_produce*() (et.al) prior to the partitioner being called.

Parameters

rk	The client instance.
rkmessage	The message being produced. Immutable.
ic_opaque	The interceptor's opaque pointer specified in ..add..().

Remarks: This interceptor is only used by producer instances.; The rkmessage object is NOT mutable and MUST NOT be modified by the interceptor.; If the partitioner fails or an unknown partition was specified, the on_acknowledgement() interceptor chain will be called from within the rd_kafka_produce*() call to maintain send-acknowledgement symmetry.

Returns: an error code on failure, the error is logged but otherwise ignored.

◆ rd_kafka_interceptor_f_on_acknowledgement_t

typedef rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_acknowledgement_t(rd_kafka_t *rk, rd_kafka_message_t *rkmessage, void *ic_opaque)

on_acknowledgement() is called to inform interceptors that a message was succesfully delivered or permanently failed delivery. The interceptor chain is called from internal librdkafka background threads, or rd_kafka_produce*() if the partitioner failed.

Parameters

rk	The client instance.
rkmessage	The message being produced. Immutable.
ic_opaque	The interceptor's opaque pointer specified in ..add..().

Remarks: This interceptor is only used by producer instances.; The rkmessage object is NOT mutable and MUST NOT be modified by the interceptor.

Warning: The on_acknowledgement() method may be called from internal librdkafka threads. An on_acknowledgement() interceptor MUST NOT call any librdkafka API's associated with the rk, or perform any blocking or prolonged work.

Returns: an error code on failure, the error is logged but otherwise ignored.

◆ rd_kafka_interceptor_f_on_consume_t

typedef rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_consume_t(rd_kafka_t *rk, rd_kafka_message_t *rkmessage, void *ic_opaque)

on_consume() is called just prior to passing the message to the application in rd_kafka_consumer_poll(), rd_kafka_consume*(), the event interface, etc.

Parameters

rk	The client instance.
rkmessage	The message being consumed. Immutable.
ic_opaque	The interceptor's opaque pointer specified in ..add..().

Remarks: This interceptor is only used by consumer instances.; The rkmessage object is NOT mutable and MUST NOT be modified by the interceptor.

Returns: an error code on failure, the error is logged but otherwise ignored.

◆ rd_kafka_interceptor_f_on_commit_t

typedef rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_commit_t(rd_kafka_t *rk, const rd_kafka_topic_partition_list_t *offsets, rd_kafka_resp_err_t err, void *ic_opaque)

on_commit() is called on completed or failed offset commit. It is called from internal librdkafka threads.

Parameters

rk	The client instance.
offsets	List of topic+partition+offset+error that were committed. The error message of each partition should be checked for error.
err	The commit error, if any.
ic_opaque	The interceptor's opaque pointer specified in ..add..().

Remarks: This interceptor is only used by consumer instances.

Warning: The on_commit() interceptor is called from internal librdkafka threads. An on_commit() interceptor MUST NOT call any librdkafka API's associated with the rk, or perform any blocking or prolonged work.

Returns: an error code on failure, the error is logged but otherwise ignored.

◆ rd_kafka_interceptor_f_on_request_sent_t

typedef rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_request_sent_t(rd_kafka_t *rk, int sockfd, const char *brokername, int32_t brokerid, int16_t ApiKey, int16_t ApiVersion, int32_t CorrId, size_t size, void *ic_opaque)

on_request_sent() is called when a request has been fully written to a broker TCP connections socket.

Parameters

rk	The client instance.
sockfd	Socket file descriptor.
brokername	Broker request is being sent to.
brokerid	Broker request is being sent to.
ApiKey	Kafka protocol request type.
ApiVersion	Kafka protocol request type version.
CorrId	Kafka protocol request correlation id.
size	Size of request.
ic_opaque	The interceptor's opaque pointer specified in ..add..().

Warning: The on_request_sent() interceptor is called from internal librdkafka broker threads. An on_request_sent() interceptor MUST NOT call any librdkafka API's associated with the rk, or perform any blocking or prolonged work.

Returns: an error code on failure, the error is logged but otherwise ignored.

◆ rd_kafka_interceptor_f_on_response_received_t

typedef rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_response_received_t(rd_kafka_t *rk, int sockfd, const char *brokername, int32_t brokerid, int16_t ApiKey, int16_t ApiVersion, int32_t CorrId, size_t size, int64_t rtt, rd_kafka_resp_err_t err, void *ic_opaque)

on_response_received() is called when a protocol response has been fully received from a broker TCP connection socket but before the response payload is parsed.

Parameters

rk	The client instance.
sockfd	Socket file descriptor (always -1).
brokername	Broker response was received from, possibly empty string on error.
brokerid	Broker response was received from.
ApiKey	Kafka protocol request type or -1 on error.
ApiVersion	Kafka protocol request type version or -1 on error.
CorrId	Kafka protocol request correlation id, possibly -1 on error.
size	Size of response, possibly 0 on error.
rtt	Request round-trip-time in microseconds, possibly -1 on error.
err	Receive error.
ic_opaque	The interceptor's opaque pointer specified in ..add..().

Warning: The on_response_received() interceptor is called from internal librdkafka broker threads. An on_response_received() interceptor MUST NOT call any librdkafka API's associated with the rk, or perform any blocking or prolonged work.

Returns: an error code on failure, the error is logged but otherwise ignored.

◆ rd_kafka_interceptor_f_on_thread_start_t

typedef rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_thread_start_t(rd_kafka_t *rk, rd_kafka_thread_type_t thread_type, const char *thread_name, void *ic_opaque)

on_thread_start() is called from a newly created librdkafka-managed thread.

Parameters

rk	The client instance.
thread_type	Thread type.
thread_name	Human-readable thread name, may not be unique.
ic_opaque	The interceptor's opaque pointer specified in ..add..().

Warning: The on_thread_start() interceptor is called from internal librdkafka threads. An on_thread_start() interceptor MUST NOT call any librdkafka API's associated with the rk, or perform any blocking or prolonged work.

Returns: an error code on failure, the error is logged but otherwise ignored.

◆ rd_kafka_interceptor_f_on_thread_exit_t

typedef rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_thread_exit_t(rd_kafka_t *rk, rd_kafka_thread_type_t thread_type, const char *thread_name, void *ic_opaque)

on_thread_exit() is called just prior to a librdkafka-managed thread exiting from the exiting thread itself.

Parameters

rk	The client instance.
thread_type	Thread type.n
thread_name	Human-readable thread name, may not be unique.
ic_opaque	The interceptor's opaque pointer specified in ..add..().

Remarks: Depending on the thread type, librdkafka may execute additional code on the thread after on_thread_exit() returns.

Warning: The on_thread_exit() interceptor is called from internal librdkafka threads. An on_thread_exit() interceptor MUST NOT call any librdkafka API's associated with the rk, or perform any blocking or prolonged work.

Returns: an error code on failure, the error is logged but otherwise ignored.

◆ rd_kafka_AdminOptions_t

typedef struct rd_kafka_AdminOptions_s rd_kafka_AdminOptions_t

AdminOptions provides a generic mechanism for setting optional parameters for the Admin API requests.

Remarks: Since AdminOptions is decoupled from the actual request type there is no enforcement to prevent setting unrelated properties, e.g. setting validate_only on a DescribeConfigs request is allowed but is silently ignored by DescribeConfigs. Future versions may introduce such enforcement.

◆ rd_kafka_NewTopic_t

typedef struct rd_kafka_NewTopic_s rd_kafka_NewTopic_t

Defines a new topic to be created.

◆ rd_kafka_DeleteTopic_t

typedef struct rd_kafka_DeleteTopic_s rd_kafka_DeleteTopic_t

Represents a topic to be deleted.

◆ rd_kafka_NewPartitions_t

typedef struct rd_kafka_NewPartitions_s rd_kafka_NewPartitions_t

Defines a new partition to be created.

◆ rd_kafka_ConfigEntry_t

typedef struct rd_kafka_ConfigEntry_s rd_kafka_ConfigEntry_t

Apache Kafka configuration entry.

◆ rd_kafka_ConfigResource_t

typedef struct rd_kafka_ConfigResource_s rd_kafka_ConfigResource_t

Apache Kafka configuration resource.

◆ rd_kafka_DeleteRecords_t

typedef struct rd_kafka_DeleteRecords_s rd_kafka_DeleteRecords_t

! Represents records to be deleted

◆ rd_kafka_DeleteGroup_t

typedef struct rd_kafka_DeleteGroup_s rd_kafka_DeleteGroup_t

Represents a group to be deleted.

◆ rd_kafka_DeleteConsumerGroupOffsets_t

typedef struct rd_kafka_DeleteConsumerGroupOffsets_s rd_kafka_DeleteConsumerGroupOffsets_t

Represents consumer group committed offsets to be deleted.

◆ rd_kafka_AclBinding_t

typedef struct rd_kafka_AclBinding_s rd_kafka_AclBinding_t

ACL Binding is used to create access control lists.

◆ rd_kafka_AclBindingFilter_t

typedef rd_kafka_AclBinding_t rd_kafka_AclBindingFilter_t

ACL Binding filter is used to filter access control lists.

◆ rd_kafka_DeleteAcls_result_response_t

typedef struct rd_kafka_DeleteAcls_result_response_s rd_kafka_DeleteAcls_result_response_t

- delete access control lists.

Enumeration Type Documentation

◆ rd_kafka_type_t

enum rd_kafka_type_t

rd_kafka_t handle type.

See also: rd_kafka_new()

Enumerator
RD_KAFKA_PRODUCER	Producer client
RD_KAFKA_CONSUMER	Consumer client

◆ rd_kafka_timestamp_type_t

enum rd_kafka_timestamp_type_t

Timestamp types

See also: rd_kafka_message_timestamp()

Enumerator
RD_KAFKA_TIMESTAMP_NOT_AVAILABLE	Timestamp not available
RD_KAFKA_TIMESTAMP_CREATE_TIME	Message creation time
RD_KAFKA_TIMESTAMP_LOG_APPEND_TIME	Log append time

◆ rd_kafka_resp_err_t

enum rd_kafka_resp_err_t

Error codes.

The negative error codes delimited by two underscores (RD_KAFKA_RESP_ERR__..) denotes errors internal to librdkafka and are displayed as "Local: <error string..>", while the error codes delimited by a single underscore (RD_KAFKA_RESP_ERR_..) denote broker errors and are displayed as "Broker: <error string..>".

See also: Use rd_kafka_err2str() to translate an error code a human readable string

Enumerator
RD_KAFKA_RESP_ERR__BEGIN	Begin internal error codes
RD_KAFKA_RESP_ERR__BAD_MSG	Received message is incorrect
RD_KAFKA_RESP_ERR__BAD_COMPRESSION	Bad/unknown compression
RD_KAFKA_RESP_ERR__DESTROY	Broker is going away
RD_KAFKA_RESP_ERR__FAIL	Generic failure
RD_KAFKA_RESP_ERR__TRANSPORT	Broker transport failure
RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE	Critical system resource
RD_KAFKA_RESP_ERR__RESOLVE	Failed to resolve broker
RD_KAFKA_RESP_ERR__MSG_TIMED_OUT	Produced message timed out
RD_KAFKA_RESP_ERR__PARTITION_EOF	Reached the end of the topic+partition queue on the broker. Not really an error. This event is disabled by default, see the `enable.partition.eof` configuration property.
RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION	Permanent: Partition does not exist in cluster.
RD_KAFKA_RESP_ERR__FS	File or filesystem error
RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC	Permanent: Topic does not exist in cluster.
RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN	All broker connections are down.
RD_KAFKA_RESP_ERR__INVALID_ARG	Invalid argument, or invalid configuration
RD_KAFKA_RESP_ERR__TIMED_OUT	Operation timed out
RD_KAFKA_RESP_ERR__QUEUE_FULL	Queue is full
RD_KAFKA_RESP_ERR__ISR_INSUFF	ISR count < required.acks
RD_KAFKA_RESP_ERR__NODE_UPDATE	Broker node update
RD_KAFKA_RESP_ERR__SSL	SSL error
RD_KAFKA_RESP_ERR__WAIT_COORD	Waiting for coordinator to become available.
RD_KAFKA_RESP_ERR__UNKNOWN_GROUP	Unknown client group
RD_KAFKA_RESP_ERR__IN_PROGRESS	Operation in progress
RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS	Previous operation in progress, wait for it to finish.
RD_KAFKA_RESP_ERR__EXISTING_SUBSCRIPTION	This operation would interfere with an existing subscription
RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS	Assigned partitions (rebalance_cb)
RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS	Revoked partitions (rebalance_cb)
RD_KAFKA_RESP_ERR__CONFLICT	Conflicting use
RD_KAFKA_RESP_ERR__STATE	Wrong state
RD_KAFKA_RESP_ERR__UNKNOWN_PROTOCOL	Unknown protocol
RD_KAFKA_RESP_ERR__NOT_IMPLEMENTED	Not implemented
RD_KAFKA_RESP_ERR__AUTHENTICATION	Authentication failure
RD_KAFKA_RESP_ERR__NO_OFFSET	No stored offset
RD_KAFKA_RESP_ERR__OUTDATED	Outdated
RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE	Timed out in queue
RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE	Feature not supported by broker
RD_KAFKA_RESP_ERR__WAIT_CACHE	Awaiting cache update
RD_KAFKA_RESP_ERR__INTR	Operation interrupted (e.g., due to yield))
RD_KAFKA_RESP_ERR__KEY_SERIALIZATION	Key serialization error
RD_KAFKA_RESP_ERR__VALUE_SERIALIZATION	Value serialization error
RD_KAFKA_RESP_ERR__KEY_DESERIALIZATION	Key deserialization error
RD_KAFKA_RESP_ERR__VALUE_DESERIALIZATION	Value deserialization error
RD_KAFKA_RESP_ERR__PARTIAL	Partial response
RD_KAFKA_RESP_ERR__READ_ONLY	Modification attempted on read-only object
RD_KAFKA_RESP_ERR__NOENT	No such entry / item not found
RD_KAFKA_RESP_ERR__UNDERFLOW	Read underflow
RD_KAFKA_RESP_ERR__INVALID_TYPE	Invalid type
RD_KAFKA_RESP_ERR__RETRY	Retry operation
RD_KAFKA_RESP_ERR__PURGE_QUEUE	Purged in queue
RD_KAFKA_RESP_ERR__PURGE_INFLIGHT	Purged in flight
RD_KAFKA_RESP_ERR__FATAL	Fatal error: see rd_kafka_fatal_error()
RD_KAFKA_RESP_ERR__INCONSISTENT	Inconsistent state
RD_KAFKA_RESP_ERR__GAPLESS_GUARANTEE	Gap-less ordering would not be guaranteed if proceeding
RD_KAFKA_RESP_ERR__MAX_POLL_EXCEEDED	Maximum poll interval exceeded
RD_KAFKA_RESP_ERR__UNKNOWN_BROKER	Unknown broker
RD_KAFKA_RESP_ERR__NOT_CONFIGURED	Functionality not configured
RD_KAFKA_RESP_ERR__FENCED	Instance has been fenced
RD_KAFKA_RESP_ERR__APPLICATION	Application generated error
RD_KAFKA_RESP_ERR__ASSIGNMENT_LOST	Assignment lost
RD_KAFKA_RESP_ERR__NOOP	No operation performed
RD_KAFKA_RESP_ERR__AUTO_OFFSET_RESET	No offset to automatically reset to
RD_KAFKA_RESP_ERR__END	End internal error codes
RD_KAFKA_RESP_ERR_UNKNOWN	Unknown broker error
RD_KAFKA_RESP_ERR_NO_ERROR	Success
RD_KAFKA_RESP_ERR_OFFSET_OUT_OF_RANGE	Offset out of range
RD_KAFKA_RESP_ERR_INVALID_MSG	Invalid message
RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART	Unknown topic or partition
RD_KAFKA_RESP_ERR_INVALID_MSG_SIZE	Invalid message size
RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE	Leader not available
RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION	Not leader for partition
RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT	Request timed out
RD_KAFKA_RESP_ERR_BROKER_NOT_AVAILABLE	Broker not available
RD_KAFKA_RESP_ERR_REPLICA_NOT_AVAILABLE	Replica not available
RD_KAFKA_RESP_ERR_MSG_SIZE_TOO_LARGE	Message size too large
RD_KAFKA_RESP_ERR_STALE_CTRL_EPOCH	StaleControllerEpochCode
RD_KAFKA_RESP_ERR_OFFSET_METADATA_TOO_LARGE	Offset metadata string too large
RD_KAFKA_RESP_ERR_NETWORK_EXCEPTION	Broker disconnected before response received
RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS	Coordinator load in progress
RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE	Coordinator not available
RD_KAFKA_RESP_ERR_NOT_COORDINATOR	Not coordinator
RD_KAFKA_RESP_ERR_TOPIC_EXCEPTION	Invalid topic
RD_KAFKA_RESP_ERR_RECORD_LIST_TOO_LARGE	Message batch larger than configured server segment size
RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS	Not enough in-sync replicas
RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS_AFTER_APPEND	Message(s) written to insufficient number of in-sync replicas
RD_KAFKA_RESP_ERR_INVALID_REQUIRED_ACKS	Invalid required acks value
RD_KAFKA_RESP_ERR_ILLEGAL_GENERATION	Specified group generation id is not valid
RD_KAFKA_RESP_ERR_INCONSISTENT_GROUP_PROTOCOL	Inconsistent group protocol
RD_KAFKA_RESP_ERR_INVALID_GROUP_ID	Invalid group.id
RD_KAFKA_RESP_ERR_UNKNOWN_MEMBER_ID	Unknown member
RD_KAFKA_RESP_ERR_INVALID_SESSION_TIMEOUT	Invalid session timeout
RD_KAFKA_RESP_ERR_REBALANCE_IN_PROGRESS	Group rebalance in progress
RD_KAFKA_RESP_ERR_INVALID_COMMIT_OFFSET_SIZE	Commit offset data size is not valid
RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED	Topic authorization failed
RD_KAFKA_RESP_ERR_GROUP_AUTHORIZATION_FAILED	Group authorization failed
RD_KAFKA_RESP_ERR_CLUSTER_AUTHORIZATION_FAILED	Cluster authorization failed
RD_KAFKA_RESP_ERR_INVALID_TIMESTAMP	Invalid timestamp
RD_KAFKA_RESP_ERR_UNSUPPORTED_SASL_MECHANISM	Unsupported SASL mechanism
RD_KAFKA_RESP_ERR_ILLEGAL_SASL_STATE	Illegal SASL state
RD_KAFKA_RESP_ERR_UNSUPPORTED_VERSION	Unuspported version
RD_KAFKA_RESP_ERR_TOPIC_ALREADY_EXISTS	Topic already exists
RD_KAFKA_RESP_ERR_INVALID_PARTITIONS	Invalid number of partitions
RD_KAFKA_RESP_ERR_INVALID_REPLICATION_FACTOR	Invalid replication factor
RD_KAFKA_RESP_ERR_INVALID_REPLICA_ASSIGNMENT	Invalid replica assignment
RD_KAFKA_RESP_ERR_INVALID_CONFIG	Invalid config
RD_KAFKA_RESP_ERR_NOT_CONTROLLER	Not controller for cluster
RD_KAFKA_RESP_ERR_INVALID_REQUEST	Invalid request
RD_KAFKA_RESP_ERR_UNSUPPORTED_FOR_MESSAGE_FORMAT	Message format on broker does not support request
RD_KAFKA_RESP_ERR_POLICY_VIOLATION	Policy violation
RD_KAFKA_RESP_ERR_OUT_OF_ORDER_SEQUENCE_NUMBER	Broker received an out of order sequence number
RD_KAFKA_RESP_ERR_DUPLICATE_SEQUENCE_NUMBER	Broker received a duplicate sequence number
RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH	Producer attempted an operation with an old epoch
RD_KAFKA_RESP_ERR_INVALID_TXN_STATE	Producer attempted a transactional operation in an invalid state
RD_KAFKA_RESP_ERR_INVALID_PRODUCER_ID_MAPPING	Producer attempted to use a producer id which is not currently assigned to its transactional id
RD_KAFKA_RESP_ERR_INVALID_TRANSACTION_TIMEOUT	Transaction timeout is larger than the maximum value allowed by the broker's max.transaction.timeout.ms
RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS	Producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing
RD_KAFKA_RESP_ERR_TRANSACTION_COORDINATOR_FENCED	Indicates that the transaction coordinator sending a WriteTxnMarker is no longer the current coordinator for a given producer
RD_KAFKA_RESP_ERR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED	Transactional Id authorization failed
RD_KAFKA_RESP_ERR_SECURITY_DISABLED	Security features are disabled
RD_KAFKA_RESP_ERR_OPERATION_NOT_ATTEMPTED	Operation not attempted
RD_KAFKA_RESP_ERR_KAFKA_STORAGE_ERROR	Disk error when trying to access log file on the disk
RD_KAFKA_RESP_ERR_LOG_DIR_NOT_FOUND	The user-specified log directory is not found in the broker config
RD_KAFKA_RESP_ERR_SASL_AUTHENTICATION_FAILED	SASL Authentication failed
RD_KAFKA_RESP_ERR_UNKNOWN_PRODUCER_ID	Unknown Producer Id
RD_KAFKA_RESP_ERR_REASSIGNMENT_IN_PROGRESS	Partition reassignment is in progress
RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_AUTH_DISABLED	Delegation Token feature is not enabled
RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_NOT_FOUND	Delegation Token is not found on server
RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_OWNER_MISMATCH	Specified Principal is not valid Owner/Renewer
RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_REQUEST_NOT_ALLOWED	Delegation Token requests are not allowed on this connection
RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_AUTHORIZATION_FAILED	Delegation Token authorization failed
RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_EXPIRED	Delegation Token is expired
RD_KAFKA_RESP_ERR_INVALID_PRINCIPAL_TYPE	Supplied principalType is not supported
RD_KAFKA_RESP_ERR_NON_EMPTY_GROUP	The group is not empty
RD_KAFKA_RESP_ERR_GROUP_ID_NOT_FOUND	The group id does not exist
RD_KAFKA_RESP_ERR_FETCH_SESSION_ID_NOT_FOUND	The fetch session ID was not found
RD_KAFKA_RESP_ERR_INVALID_FETCH_SESSION_EPOCH	The fetch session epoch is invalid
RD_KAFKA_RESP_ERR_LISTENER_NOT_FOUND	No matching listener
RD_KAFKA_RESP_ERR_TOPIC_DELETION_DISABLED	Topic deletion is disabled
RD_KAFKA_RESP_ERR_FENCED_LEADER_EPOCH	Leader epoch is older than broker epoch
RD_KAFKA_RESP_ERR_UNKNOWN_LEADER_EPOCH	Leader epoch is newer than broker epoch
RD_KAFKA_RESP_ERR_UNSUPPORTED_COMPRESSION_TYPE	Unsupported compression type
RD_KAFKA_RESP_ERR_STALE_BROKER_EPOCH	Broker epoch has changed
RD_KAFKA_RESP_ERR_OFFSET_NOT_AVAILABLE	Leader high watermark is not caught up
RD_KAFKA_RESP_ERR_MEMBER_ID_REQUIRED	Group member needs a valid member ID
RD_KAFKA_RESP_ERR_PREFERRED_LEADER_NOT_AVAILABLE	Preferred leader was not available
RD_KAFKA_RESP_ERR_GROUP_MAX_SIZE_REACHED	Consumer group has reached maximum size
RD_KAFKA_RESP_ERR_FENCED_INSTANCE_ID	Static consumer fenced by other consumer with same group.instance.id.
RD_KAFKA_RESP_ERR_ELIGIBLE_LEADERS_NOT_AVAILABLE	Eligible partition leaders are not available
RD_KAFKA_RESP_ERR_ELECTION_NOT_NEEDED	Leader election not needed for topic partition
RD_KAFKA_RESP_ERR_NO_REASSIGNMENT_IN_PROGRESS	No partition reassignment is in progress
RD_KAFKA_RESP_ERR_GROUP_SUBSCRIBED_TO_TOPIC	Deleting offsets of a topic while the consumer group is subscribed to it
RD_KAFKA_RESP_ERR_INVALID_RECORD	Broker failed to validate record
RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT	There are unstable offsets that need to be cleared
RD_KAFKA_RESP_ERR_THROTTLING_QUOTA_EXCEEDED	Throttling quota has been exceeded
RD_KAFKA_RESP_ERR_PRODUCER_FENCED	There is a newer producer with the same transactionalId which fences the current one
RD_KAFKA_RESP_ERR_RESOURCE_NOT_FOUND	Request illegally referred to resource that does not exist
RD_KAFKA_RESP_ERR_DUPLICATE_RESOURCE	Request illegally referred to the same resource twice
RD_KAFKA_RESP_ERR_UNACCEPTABLE_CREDENTIAL	Requested credential would not meet criteria for acceptability
RD_KAFKA_RESP_ERR_INCONSISTENT_VOTER_SET	Indicates that the either the sender or recipient of a voter-only request is not one of the expected voters
RD_KAFKA_RESP_ERR_INVALID_UPDATE_VERSION	Invalid update version
RD_KAFKA_RESP_ERR_FEATURE_UPDATE_FAILED	Unable to update finalized features due to server error
RD_KAFKA_RESP_ERR_PRINCIPAL_DESERIALIZATION_FAILURE	Request principal deserialization failed during forwarding

◆ rd_kafka_vtype_t

enum rd_kafka_vtype_t

Var-arg tag types.

See also: rd_kafka_producev()

Enumerator
RD_KAFKA_VTYPE_END	va-arg sentinel
RD_KAFKA_VTYPE_TOPIC	(const char *) Topic name
RD_KAFKA_VTYPE_RKT	(rd_kafka_topic_t *) Topic handle
RD_KAFKA_VTYPE_PARTITION	(int32_t) Partition
RD_KAFKA_VTYPE_VALUE	(void *, size_t) Message value (payload)
RD_KAFKA_VTYPE_KEY	(void *, size_t) Message key
RD_KAFKA_VTYPE_OPAQUE	(void *) Per-message application opaque value. This is the same as the _private field in rd_kafka_message_t, also known as the msg_opaque.
RD_KAFKA_VTYPE_MSGFLAGS	(int) RD_KAFKA_MSG_F_.. flags
RD_KAFKA_VTYPE_TIMESTAMP	(int64_t) Milliseconds since epoch UTC
RD_KAFKA_VTYPE_HEADER	(const char , const void , ssize_t) Message Header
RD_KAFKA_VTYPE_HEADERS	(rd_kafka_headers_t *) Headers list

◆ rd_kafka_msg_status_t

enum rd_kafka_msg_status_t

Message persistence status can be used by the application to find out if a produced message was persisted in the topic log.

Enumerator
RD_KAFKA_MSG_STATUS_NOT_PERSISTED	Message was never transmitted to the broker, or failed with an error indicating it was not written to the log. Application retry risks ordering, but not duplication.
RD_KAFKA_MSG_STATUS_POSSIBLY_PERSISTED	Message was transmitted to broker, but no acknowledgement was received. Application retry risks ordering and duplication.
RD_KAFKA_MSG_STATUS_PERSISTED	Message was written to the log and acknowledged by the broker. No reason for application to retry. Note: this value should only be trusted with `acks=all`.

◆ rd_kafka_conf_res_t

enum rd_kafka_conf_res_t

Configuration result type.

Enumerator
RD_KAFKA_CONF_UNKNOWN	Unknown configuration name.
RD_KAFKA_CONF_INVALID	Invalid configuration value or property or value not supported in this build.
RD_KAFKA_CONF_OK	Configuration okay

◆ rd_kafka_cert_type_t

enum rd_kafka_cert_type_t

SSL certificate type.

See also: rd_kafka_conf_set_ssl_cert

Enumerator
RD_KAFKA_CERT_PUBLIC_KEY	Client's public key
RD_KAFKA_CERT_PRIVATE_KEY	Client's private key
RD_KAFKA_CERT_CA	CA certificate

◆ rd_kafka_cert_enc_t

enum rd_kafka_cert_enc_t

SSL certificate encoding.

See also: rd_kafka_conf_set_ssl_cert

Enumerator
RD_KAFKA_CERT_ENC_PKCS12	PKCS#12
RD_KAFKA_CERT_ENC_DER	DER / binary X.509 ASN1
RD_KAFKA_CERT_ENC_PEM	PEM

◆ rd_kafka_thread_type_t

enum rd_kafka_thread_type_t

librdkafka internal thread type.

See also: rd_kafka_interceptor_add_on_thread_start()

Enumerator
RD_KAFKA_THREAD_MAIN	librdkafka's internal main thread
RD_KAFKA_THREAD_BACKGROUND	Background thread (if enabled)
RD_KAFKA_THREAD_BROKER	Per-broker thread

◆ rd_kafka_admin_op_t

enum rd_kafka_admin_op_t

Admin operation enum name for use with rd_kafka_AdminOptions_new()

See also: rd_kafka_AdminOptions_new()

Enumerator
RD_KAFKA_ADMIN_OP_ANY	Default value
RD_KAFKA_ADMIN_OP_CREATETOPICS	CreateTopics
RD_KAFKA_ADMIN_OP_DELETETOPICS	DeleteTopics
RD_KAFKA_ADMIN_OP_CREATEPARTITIONS	CreatePartitions
RD_KAFKA_ADMIN_OP_ALTERCONFIGS	AlterConfigs
RD_KAFKA_ADMIN_OP_DESCRIBECONFIGS	DescribeConfigs
RD_KAFKA_ADMIN_OP_DELETERECORDS	DeleteRecords
RD_KAFKA_ADMIN_OP_DELETEGROUPS	DeleteGroups
RD_KAFKA_ADMIN_OP_DELETECONSUMERGROUPOFFSETS	DeleteConsumerGroupOffsets
RD_KAFKA_ADMIN_OP_CREATEACLS	CreateAcls
RD_KAFKA_ADMIN_OP_DESCRIBEACLS	DescribeAcls
RD_KAFKA_ADMIN_OP_DELETEACLS	DeleteAcls
RD_KAFKA_ADMIN_OP__CNT	Number of ops defined

◆ rd_kafka_ConfigSource_t

enum rd_kafka_ConfigSource_t

Apache Kafka config sources.

Remarks: These entities relate to the cluster, not the local client.

See also: rd_kafka_conf_set(), et.al. for local client configuration.

Enumerator
RD_KAFKA_CONFIG_SOURCE_UNKNOWN_CONFIG	Source unknown, e.g., in the ConfigEntry used for alter requests where source is not set
RD_KAFKA_CONFIG_SOURCE_DYNAMIC_TOPIC_CONFIG	Dynamic topic config that is configured for a specific topic
RD_KAFKA_CONFIG_SOURCE_DYNAMIC_BROKER_CONFIG	Dynamic broker config that is configured for a specific broker
RD_KAFKA_CONFIG_SOURCE_DYNAMIC_DEFAULT_BROKER_CONFIG	Dynamic broker config that is configured as default for all brokers in the cluster
RD_KAFKA_CONFIG_SOURCE_STATIC_BROKER_CONFIG	Static broker config provided as broker properties at startup (e.g. from server.properties file)
RD_KAFKA_CONFIG_SOURCE_DEFAULT_CONFIG	Built-in default configuration for configs that have a default value
RD_KAFKA_CONFIG_SOURCE__CNT	Number of source types defined

◆ rd_kafka_ResourceType_t

enum rd_kafka_ResourceType_t

Apache Kafka resource types.

Enumerator
RD_KAFKA_RESOURCE_UNKNOWN	Unknown
RD_KAFKA_RESOURCE_ANY	Any (used for lookups)
RD_KAFKA_RESOURCE_TOPIC	Topic
RD_KAFKA_RESOURCE_GROUP	Group
RD_KAFKA_RESOURCE_BROKER	Broker
RD_KAFKA_RESOURCE__CNT	Number of resource types defined

◆ rd_kafka_ResourcePatternType_t

enum rd_kafka_ResourcePatternType_t

Apache Kafka pattern types.

Enumerator
RD_KAFKA_RESOURCE_PATTERN_UNKNOWN	Unknown
RD_KAFKA_RESOURCE_PATTERN_ANY	Any (used for lookups)
RD_KAFKA_RESOURCE_PATTERN_MATCH	Match: will perform pattern matching
RD_KAFKA_RESOURCE_PATTERN_LITERAL	Literal: A literal resource name
RD_KAFKA_RESOURCE_PATTERN_PREFIXED	Prefixed: A prefixed resource name

◆ rd_kafka_AclOperation_t

enum rd_kafka_AclOperation_t

Apache Kafka ACL operation types.

Enumerator
RD_KAFKA_ACL_OPERATION_UNKNOWN	Unknown
RD_KAFKA_ACL_OPERATION_ANY	In a filter, matches any AclOperation
RD_KAFKA_ACL_OPERATION_ALL	ALL operation
RD_KAFKA_ACL_OPERATION_READ	READ operation
RD_KAFKA_ACL_OPERATION_WRITE	WRITE operation
RD_KAFKA_ACL_OPERATION_CREATE	CREATE operation
RD_KAFKA_ACL_OPERATION_DELETE	DELETE operation
RD_KAFKA_ACL_OPERATION_ALTER	ALTER operation
RD_KAFKA_ACL_OPERATION_DESCRIBE	DESCRIBE operation
RD_KAFKA_ACL_OPERATION_CLUSTER_ACTION	CLUSTER_ACTION operation
RD_KAFKA_ACL_OPERATION_DESCRIBE_CONFIGS	DESCRIBE_CONFIGS operation
RD_KAFKA_ACL_OPERATION_ALTER_CONFIGS	ALTER_CONFIGS operation
RD_KAFKA_ACL_OPERATION_IDEMPOTENT_WRITE	IDEMPOTENT_WRITE operation

◆ rd_kafka_AclPermissionType_t

enum rd_kafka_AclPermissionType_t

Apache Kafka ACL permission types.

Enumerator
RD_KAFKA_ACL_PERMISSION_TYPE_UNKNOWN	Unknown
RD_KAFKA_ACL_PERMISSION_TYPE_ANY	In a filter, matches any AclPermissionType
RD_KAFKA_ACL_PERMISSION_TYPE_DENY	Disallows access
RD_KAFKA_ACL_PERMISSION_TYPE_ALLOW	Grants access.

Function Documentation

◆ rd_kafka_version()

RD_EXPORT int rd_kafka_version ( void )

Returns the librdkafka version as integer.

Returns: Version integer.

See also: See RD_KAFKA_VERSION for how to parse the integer format.; Use rd_kafka_version_str() to retreive the version as a string.

◆ rd_kafka_version_str()

const RD_EXPORT char* rd_kafka_version_str ( void )

Returns the librdkafka version as string.

Returns: Version string

◆ rd_kafka_get_debug_contexts()

const RD_EXPORT char* rd_kafka_get_debug_contexts ( void )

Retrieve supported debug contexts for use with the "debug" configuration property. (runtime)

Returns: Comma-separated list of available debugging contexts.

◆ rd_kafka_err2str()

const RD_EXPORT char* rd_kafka_err2str ( rd_kafka_resp_err_t err )

Returns a human readable representation of a kafka error.

Parameters

err	Error code to translate

◆ rd_kafka_err2name()

const RD_EXPORT char* rd_kafka_err2name ( rd_kafka_resp_err_t err )

Returns the error code name (enum name).

Parameters

err	Error code to translate

◆ rd_kafka_last_error()

RD_EXPORT rd_kafka_resp_err_t rd_kafka_last_error ( void )

Returns the last error code generated by a legacy API call in the current thread.

The legacy APIs are the ones using errno to propagate error value, namely:

The main use for this function is to avoid converting system errno values to rd_kafka_resp_err_t codes for legacy APIs.

Remarks: The last error is stored per-thread, if multiple rd_kafka_t handles are used in the same application thread the developer needs to make sure rd_kafka_last_error() is called immediately after a failed API call.; errno propagation from librdkafka is not safe on Windows and should not be used, use rd_kafka_last_error() instead.

◆ rd_kafka_errno2err()

RD_EXPORT RD_DEPRECATED rd_kafka_resp_err_t rd_kafka_errno2err ( int errnox )

Converts the system errno value errnox to a rd_kafka_resp_err_t error code upon failure from the following functions:

Parameters

errnox System errno value to convert

Returns: Appropriate error code for errnox

Remarks: A better alternative is to call rd_kafka_last_error() immediately after any of the above functions return -1 or NULL.

Deprecated:: Use rd_kafka_last_error() to retrieve the last error code set by the legacy librdkafka APIs.

See also: rd_kafka_last_error()

◆ rd_kafka_errno()

RD_EXPORT RD_DEPRECATED int rd_kafka_errno ( void )

Returns the thread-local system errno.

On most platforms this is the same as errno but in case of different runtimes between library and application (e.g., Windows static DLLs) this provides a means for exposing the errno librdkafka uses.

Remarks: The value is local to the current calling thread.

Deprecated:: Use rd_kafka_last_error() to retrieve the last error code set by the legacy librdkafka APIs.

◆ rd_kafka_fatal_error()

RD_EXPORT rd_kafka_resp_err_t rd_kafka_fatal_error	(	rd_kafka_t *	rk,
		char *	errstr,
		size_t	errstr_size
	)

Returns the first fatal error set on this client instance, or RD_KAFKA_RESP_ERR_NO_ERROR if no fatal error has occurred.

This function is to be used with the Idempotent Producer and error_cb to detect fatal errors.

Generally all errors raised by error_cb are to be considered informational and temporary, the client will try to recover from all errors in a graceful fashion (by retrying, etc).

However, some errors should logically be considered fatal to retain consistency; in particular a set of errors that may occur when using the Idempotent Producer and the in-order or exactly-once producer guarantees can't be satisfied.

Parameters

rk	Client instance.
errstr	A human readable error string (nul-terminated) is written to this location that must be of at least `errstr_size` bytes. The `errstr` is only written to if there is a fatal error.
errstr_size	Writable size in `errstr`.

Returns: RD_KAFKA_RESP_ERR_NO_ERROR if no fatal error has been raised, else any other error code.

◆ rd_kafka_test_fatal_error()

RD_EXPORT rd_kafka_resp_err_t rd_kafka_test_fatal_error	(	rd_kafka_t *	rk,
		rd_kafka_resp_err_t	err,
		const char *	reason
	)

Trigger a fatal error for testing purposes.

Since there is no practical way to trigger real fatal errors in the idempotent producer, this method allows an application to trigger fabricated fatal errors in tests to check its error handling code.

Parameters

rk	Client instance.
err	The underlying error code.
reason	A human readable error reason. Will be prefixed with "test_fatal_error: " to differentiate from real fatal errors.

Returns: RD_KAFKA_RESP_ERR_NO_ERROR if a fatal error was triggered, or RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS if a previous fatal error has already been triggered.

◆ rd_kafka_error_code()

RD_EXPORT rd_kafka_resp_err_t rd_kafka_error_code ( const rd_kafka_error_t * error )

Returns: the error code for error or RD_KAFKA_RESP_ERR_NO_ERROR if error is NULL.

◆ rd_kafka_error_name()

const RD_EXPORT char* rd_kafka_error_name ( const rd_kafka_error_t * error )

Returns: the error code name for error, e.g, "ERR_UNKNOWN_MEMBER_ID", or an empty string if error is NULL.

Remarks: The lifetime of the returned pointer is the same as the error object.

See also: rd_kafka_err2name()

◆ rd_kafka_error_string()

const RD_EXPORT char* rd_kafka_error_string ( const rd_kafka_error_t * error )

Returns: a human readable error string for error, or an empty string if error is NULL.

Remarks: The lifetime of the returned pointer is the same as the error object.

◆ rd_kafka_error_is_fatal()

RD_EXPORT int rd_kafka_error_is_fatal ( const rd_kafka_error_t * error )

Returns: 1 if the error is a fatal error, indicating that the client instance is no longer usable, else 0 (also if error is NULL).

◆ rd_kafka_error_is_retriable()

RD_EXPORT int rd_kafka_error_is_retriable ( const rd_kafka_error_t * error )

Returns: 1 if the operation may be retried, else 0 (also if error is NULL).

◆ rd_kafka_error_txn_requires_abort()

RD_EXPORT int rd_kafka_error_txn_requires_abort ( const rd_kafka_error_t * error )

Returns: 1 if the error is an abortable transaction error in which case the application must call rd_kafka_abort_transaction() and start a new transaction with rd_kafka_begin_transaction() if it wishes to proceed with transactions. Else returns 0 (also if error is NULL).

Remarks: The return value of this method is only valid for errors returned by the transactional API.

◆ rd_kafka_error_destroy()

RD_EXPORT void rd_kafka_error_destroy ( rd_kafka_error_t * error )

Free and destroy an error object.

Remarks: As a conveniance it is permitted to pass a NULL error.

◆ rd_kafka_error_new()

RD_EXPORT rd_kafka_error_t* rd_kafka_error_new	(	rd_kafka_resp_err_t	code,
		const char *	fmt,
			...
	)

Create a new error object with error code and optional human readable error string in fmt.

This method is mainly to be used for mocking errors in application test code.

The returned object must be destroyed with rd_kafka_error_destroy().

◆ rd_kafka_topic_partition_destroy()

RD_EXPORT void rd_kafka_topic_partition_destroy ( rd_kafka_topic_partition_t * rktpar )

Destroy a rd_kafka_topic_partition_t.

Remarks: This must not be called for elements in a topic partition list.

◆ rd_kafka_topic_partition_list_new()

RD_EXPORT rd_kafka_topic_partition_list_t* rd_kafka_topic_partition_list_new ( int size )

Create a new list/vector Topic+Partition container.

Parameters

size	Initial allocated size used when the expected number of elements is known or can be estimated. Avoids reallocation and possibly relocation of the elems array.

Returns: A newly allocated Topic+Partition list.

Remarks: Use rd_kafka_topic_partition_list_destroy() to free all resources in use by a list and the list itself.

See also: rd_kafka_topic_partition_list_add()

◆ rd_kafka_topic_partition_list_add()

RD_EXPORT rd_kafka_topic_partition_t* rd_kafka_topic_partition_list_add	(	rd_kafka_topic_partition_list_t *	rktparlist,
		const char *	topic,
		int32_t	partition
	)

Add topic+partition to list.

Parameters

rktparlist	List to extend
topic	Topic name (copied)
partition	Partition id

Returns: The object which can be used to fill in additionals fields.

◆ rd_kafka_topic_partition_list_add_range()

RD_EXPORT void rd_kafka_topic_partition_list_add_range	(	rd_kafka_topic_partition_list_t *	rktparlist,
		const char *	topic,
		int32_t	start,
		int32_t	stop
	)

Add range of partitions from start to stop inclusive.

Parameters

rktparlist	List to extend
topic	Topic name (copied)
start	Start partition of range
stop	Last partition of range (inclusive)

◆ rd_kafka_topic_partition_list_del()

RD_EXPORT int rd_kafka_topic_partition_list_del	(	rd_kafka_topic_partition_list_t *	rktparlist,
		const char *	topic,
		int32_t	partition
	)

Delete partition from list.

Parameters

rktparlist	List to modify
topic	Topic name to match
partition	Partition to match

Returns: 1 if partition was found (and removed), else 0.

Remarks: Any held indices to elems[] are unusable after this call returns 1.

◆ rd_kafka_topic_partition_list_del_by_idx()

RD_EXPORT int rd_kafka_topic_partition_list_del_by_idx	(	rd_kafka_topic_partition_list_t *	rktparlist,
		int	idx
	)

Delete partition from list by elems[] index.

Returns: 1 if partition was found (and removed), else 0.

See also: rd_kafka_topic_partition_list_del()

◆ rd_kafka_topic_partition_list_copy()

RD_EXPORT rd_kafka_topic_partition_list_t* rd_kafka_topic_partition_list_copy ( const rd_kafka_topic_partition_list_t * src )

Make a copy of an existing list.

Parameters

src	The existing list to copy.

Returns: A new list fully populated to be identical to src

◆ rd_kafka_topic_partition_list_set_offset()

RD_EXPORT rd_kafka_resp_err_t rd_kafka_topic_partition_list_set_offset	(	rd_kafka_topic_partition_list_t *	rktparlist,
		const char *	topic,
		int32_t	partition,
		int64_t	offset
	)

Set offset to offset for topic and partition.

Returns: RD_KAFKA_RESP_ERR_NO_ERROR on success or RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION if partition was not found in the list.

◆ rd_kafka_topic_partition_list_find()

RD_EXPORT rd_kafka_topic_partition_t* rd_kafka_topic_partition_list_find	(	const rd_kafka_topic_partition_list_t *	rktparlist,
		const char *	topic,
		int32_t	partition
	)

Find element by topic and partition.

Returns: a pointer to the first matching element, or NULL if not found.

◆ rd_kafka_topic_partition_list_sort()

RD_EXPORT void rd_kafka_topic_partition_list_sort	(	rd_kafka_topic_partition_list_t *	rktparlist,
		int()(const void a, const void b, void cmp_opaque)	cmp,
		void *	cmp_opaque
	)

Sort list using comparator cmp.

If cmp is NULL the default comparator will be used that sorts by ascending topic name and partition.

cmp_opaque is provided as the cmp_opaque argument to cmp.

◆ rd_kafka_headers_new()

RD_EXPORT rd_kafka_headers_t* rd_kafka_headers_new ( size_t initial_count )

Create a new headers list.

Parameters

initial_count Preallocate space for this number of headers. Any number of headers may be added, updated and removed regardless of the initial count.

◆ rd_kafka_header_add()

RD_EXPORT rd_kafka_resp_err_t rd_kafka_header_add	(	rd_kafka_headers_t *	hdrs,
		const char *	name,
		ssize_t	name_size,
		const void *	value,
		ssize_t	value_size
	)

Add header with name name and value val (copied) of size size (not including null-terminator).

Parameters

hdrs	Headers list.
name	Header name.
name_size	Header name size (not including the null-terminator). If -1 the `name` length is automatically acquired using strlen().
value	Pointer to header value, or NULL (set size to 0 or -1).
value_size	Size of header value. If -1 the `value` is assumed to be a null-terminated string and the length is automatically acquired using strlen().

Returns: RD_KAFKA_RESP_ERR__READ_ONLY if the headers are read-only, else RD_KAFKA_RESP_ERR_NO_ERROR.

◆ rd_kafka_header_remove()

RD_EXPORT rd_kafka_resp_err_t rd_kafka_header_remove	(	rd_kafka_headers_t *	hdrs,
		const char *	name
	)

Remove all headers for the given key (if any).

Returns: RD_KAFKA_RESP_ERR__READ_ONLY if the headers are read-only, RD_KAFKA_RESP_ERR__NOENT if no matching headers were found, else RD_KAFKA_RESP_ERR_NO_ERROR if headers were removed.

◆ rd_kafka_header_get_last()

RD_EXPORT rd_kafka_resp_err_t rd_kafka_header_get_last	(	const rd_kafka_headers_t *	hdrs,
		const char *	name,
		const void **	valuep,
		size_t *	sizep
	)

Find last header in list hdrs matching name.

Parameters

hdrs	Headers list.
name	Header to find (last match).
valuep	(out) Set to a (null-terminated) const pointer to the value (may be NULL).
sizep	(out) Set to the value's size (not including null-terminator).

Returns: RD_KAFKA_RESP_ERR_NO_ERROR if an entry was found, else RD_KAFKA_RESP_ERR__NOENT.

Remarks: The returned pointer in valuep includes a trailing null-terminator that is not accounted for in sizep.; The returned pointer is only valid as long as the headers list and the header item is valid.

◆ rd_kafka_header_get()

RD_EXPORT rd_kafka_resp_err_t rd_kafka_header_get	(	const rd_kafka_headers_t *	hdrs,
		size_t	idx,
		const char *	name,
		const void **	valuep,
		size_t *	sizep
	)

Iterator for headers matching name.

   Same semantics as rd_kafka_header_get_last()

Parameters

hdrs	Headers to iterate.
idx	Iterator index, start at 0 and increment by one for each call as long as RD_KAFKA_RESP_ERR_NO_ERROR is returned.
name	Header name to match.
valuep	(out) Set to a (null-terminated) const pointer to the value (may be NULL).
sizep	(out) Set to the value's size (not including null-terminator).

◆ rd_kafka_header_get_all()

RD_EXPORT rd_kafka_resp_err_t rd_kafka_header_get_all	(	const rd_kafka_headers_t *	hdrs,
		size_t	idx,
		const char **	namep,
		const void **	valuep,
		size_t *	sizep
	)

Iterator for all headers.

   Same semantics as rd_kafka_header_get()

See also: rd_kafka_header_get()

◆ rd_kafka_message_errstr()

const RD_EXPORT char* rd_kafka_message_errstr ( const rd_kafka_message_t * rkmessage )

Returns the error string for an errored rd_kafka_message_t or NULL if there was no error.

Remarks: This function MUST NOT be used with the producer.

◆ rd_kafka_message_timestamp()

RD_EXPORT int64_t rd_kafka_message_timestamp	(	const rd_kafka_message_t *	rkmessage,
		rd_kafka_timestamp_type_t *	tstype
	)

Returns the message timestamp for a consumed message.

The timestamp is the number of milliseconds since the epoch (UTC).

tstype (if not NULL) is updated to indicate the type of timestamp.

Returns: message timestamp, or -1 if not available.

Remarks: Message timestamps require broker version 0.10.0 or later.

◆ rd_kafka_message_latency()

RD_EXPORT int64_t rd_kafka_message_latency ( const rd_kafka_message_t * rkmessage )

Returns the latency for a produced message measured from the produce() call.

Returns: the latency in microseconds, or -1 if not available.

◆ rd_kafka_message_broker_id()

RD_EXPORT int32_t rd_kafka_message_broker_id ( const rd_kafka_message_t * rkmessage )

Returns the broker id of the broker the message was produced to or fetched from.

Returns: a broker id if known, else -1.

◆ rd_kafka_message_headers()

RD_EXPORT rd_kafka_resp_err_t rd_kafka_message_headers	(	const rd_kafka_message_t *	rkmessage,
		rd_kafka_headers_t **	hdrsp
	)

Get the message header list.

The returned pointer in *hdrsp is associated with the rkmessage and must not be used after destruction of the message object or the header list is replaced with rd_kafka_message_set_headers().

Returns: RD_KAFKA_RESP_ERR_NO_ERROR if headers were returned, RD_KAFKA_RESP_ERR__NOENT if the message has no headers, or another error code if the headers could not be parsed.

Remarks: Headers require broker version 0.11.0.0 or later.; As an optimization the raw protocol headers are parsed on the first call to this function.

◆ rd_kafka_message_detach_headers()

RD_EXPORT rd_kafka_resp_err_t rd_kafka_message_detach_headers	(	rd_kafka_message_t *	rkmessage,
		rd_kafka_headers_t **	hdrsp
	)

Get the message header list and detach the list from the message making the application the owner of the headers. The application must eventually destroy the headers using rd_kafka_headers_destroy(). The message's headers will be set to NULL.

Otherwise same semantics as rd_kafka_message_headers()

See also: rd_kafka_message_headers

◆ rd_kafka_message_set_headers()

RD_EXPORT void rd_kafka_message_set_headers	(	rd_kafka_message_t *	rkmessage,
		rd_kafka_headers_t *	hdrs
	)

Replace the message's current headers with a new list.

Parameters

rkmessage	The message to set headers.
hdrs	New header list. The message object assumes ownership of the list, the list will be destroyed automatically with the message object. The new headers list may be updated until the message object is passed or returned to librdkafka.

Remarks: The existing headers object, if any, will be destroyed.

◆ rd_kafka_header_cnt()

RD_EXPORT size_t rd_kafka_header_cnt ( const rd_kafka_headers_t * hdrs )

Returns the number of header key/value pairs.

Parameters

hdrs	Headers to count

◆ rd_kafka_message_status()

RD_EXPORT rd_kafka_msg_status_t rd_kafka_message_status ( const rd_kafka_message_t * rkmessage )

Returns the message's persistence status in the topic log.

Remarks: The message status is not available in on_acknowledgement interceptors.

◆ rd_kafka_conf_new()

RD_EXPORT rd_kafka_conf_t* rd_kafka_conf_new ( void )

Create configuration object.

When providing your own configuration to the rd_kafka_*_new_*() calls the rd_kafka_conf_t objects needs to be created with this function which will set up the defaults. I.e.:

rd_kafka_conf_t *myconf;
rd_kafka_conf_res_t res;
 
myconf = rd_kafka_conf_new();
res = rd_kafka_conf_set(myconf, "socket.timeout.ms", "600",
                        errstr, sizeof(errstr));
if (res != RD_KAFKA_CONF_OK)
   die("%s\n", errstr);
 
rk = rd_kafka_new(..., myconf);

Please see CONFIGURATION.md for the default settings or use rd_kafka_conf_properties_show() to provide the information at runtime.

The properties are identical to the Apache Kafka configuration properties whenever possible.

Remarks: A successful call to rd_kafka_new() will assume ownership of the conf object and rd_kafka_conf_destroy() must not be called.

Returns: A new rd_kafka_conf_t object with defaults set.

See also: rd_kafka_new(), rd_kafka_conf_set(), rd_kafka_conf_destroy()

◆ rd_kafka_conf_dup()

RD_EXPORT rd_kafka_conf_t* rd_kafka_conf_dup ( const rd_kafka_conf_t * conf )

Creates a copy/duplicate of configuration object conf.

Remarks: Interceptors are NOT copied to the new configuration object.

See also: rd_kafka_interceptor_f_on_conf_dup

◆ rd_kafka_conf()

const RD_EXPORT rd_kafka_conf_t* rd_kafka_conf ( rd_kafka_t * rk )

Returns: the configuration object used by an rd_kafka_t instance. For use with rd_kafka_conf_get(), et.al., to extract configuration properties from a running client.

Remarks: the returned object is read-only and its lifetime is the same as the rd_kafka_t object.

◆ rd_kafka_conf_set()

RD_EXPORT rd_kafka_conf_res_t rd_kafka_conf_set	(	rd_kafka_conf_t *	conf,
		const char *	name,
		const char *	value,
		char *	errstr,
		size_t	errstr_size
	)

Sets a configuration property.

conf must have been previously created with rd_kafka_conf_new().

Fallthrough: Topic-level configuration properties may be set using this interface in which case they are applied on the default_topic_conf. If no default_topic_conf has been set one will be created. Any sub-sequent rd_kafka_conf_set_default_topic_conf() calls will replace the current default topic configuration.

Returns: rd_kafka_conf_res_t to indicate success or failure. In case of failure errstr is updated to contain a human readable error string.

Remarks: Setting properties or values that were disabled at build time due to missing dependencies will return RD_KAFKA_CONF_INVALID.

◆ rd_kafka_conf_set_background_event_cb()

RD_EXPORT void rd_kafka_conf_set_background_event_cb	(	rd_kafka_conf_t *	conf,
		void()(rd_kafka_t rk, rd_kafka_event_t rkev, void opaque)	event_cb
	)

Generic event callback to be used with the event API to trigger callbacks for rd_kafka_event_t objects from a background thread serving the background queue.

How to use:

First set the event callback on the configuration object with this function, followed by creating an rd_kafka_t instance with rd_kafka_new().
Get the instance's background queue with rd_kafka_queue_get_background() and pass it as the reply/response queue to an API that takes an event queue, such as rd_kafka_CreateTopics().
As the response event is ready and enqueued on the background queue the event callback will be triggered from the background thread.
Prior to destroying the client instance, loose your reference to the background queue by calling rd_kafka_queue_destroy().

The application must destroy the rkev passed to event cb using rd_kafka_event_destroy().

The event_cb opaque argument is the opaque set with rd_kafka_conf_set_opaque().

Remarks: This callback is a specialized alternative to the poll-based event API described in the Event interface section.; The event_cb will be called spontaneously from a background thread completely managed by librdkafka. Take care to perform proper locking of application objects.

Warning: The application MUST NOT call rd_kafka_destroy() from the event callback.

See also: rd_kafka_queue_get_background

◆ rd_kafka_conf_set_dr_cb()

RD_EXPORT void rd_kafka_conf_set_dr_cb	(	rd_kafka_conf_t *	conf,
		void()(rd_kafka_t rk, void payload, size_t len, rd_kafka_resp_err_t err, void opaque, void *msg_opaque)	dr_cb
	)

Deprecated:: See rd_kafka_conf_set_dr_msg_cb()

◆ rd_kafka_conf_set_dr_msg_cb()

RD_EXPORT void rd_kafka_conf_set_dr_msg_cb	(	rd_kafka_conf_t *	conf,
		void()(rd_kafka_t rk, const rd_kafka_message_t rkmessage, void opaque)	dr_msg_cb
	)

Producer: Set delivery report callback in provided conf object.

The delivery report callback will be called once for each message accepted by rd_kafka_produce() (et.al) with err set to indicate the result of the produce request.

The callback is called when a message is succesfully produced or if librdkafka encountered a permanent failure. Delivery errors occur when the retry count is exceeded, when the message.timeout.ms timeout is exceeded or there is a permanent error like RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART.

An application must call rd_kafka_poll() at regular intervals to serve queued delivery report callbacks.

The broker-assigned offset can be retrieved with rkmessage->offset and the timestamp can be retrieved using rd_kafka_message_timestamp().

The dr_msg_cb opaque argument is the opaque set with rd_kafka_conf_set_opaque(). The per-message msg_opaque value is available in rd_kafka_message_t._private.

Remarks: The Idempotent Producer may return invalid timestamp (RD_KAFKA_TIMESTAMP_NOT_AVAILABLE), and and offset (RD_KAFKA_OFFSET_INVALID) for retried messages that were previously successfully delivered but not properly acknowledged.

◆ rd_kafka_conf_set_consume_cb()

RD_EXPORT void rd_kafka_conf_set_consume_cb	(	rd_kafka_conf_t *	conf,
		void()(rd_kafka_message_t rkmessage, void *opaque)	consume_cb
	)

Consumer: Set consume callback for use with rd_kafka_consumer_poll()

The consume_cb opaque argument is the opaque set with rd_kafka_conf_set_opaque().

◆ rd_kafka_conf_set_rebalance_cb()

RD_EXPORT void rd_kafka_conf_set_rebalance_cb	(	rd_kafka_conf_t *	conf,
		void()(rd_kafka_t rk, rd_kafka_resp_err_t err, rd_kafka_topic_partition_list_t partitions, void opaque)	rebalance_cb
	)

Consumer: Set rebalance callback for use with coordinated consumer group balancing.

The err field is set to either RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS or RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS and 'partitions' contains the full partition set that was either assigned or revoked.

Registering a rebalance_cb turns off librdkafka's automatic partition assignment/revocation and instead delegates that responsibility to the application's rebalance_cb.

The rebalance callback is responsible for updating librdkafka's assignment set based on the two events: RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS and RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS but should also be able to handle arbitrary rebalancing failures where err is neither of those.

Remarks: In this latter case (arbitrary error), the application must call rd_kafka_assign(rk, NULL) to synchronize state.

For eager/non-cooperative partition.assignment.strategy assignors, such as range and roundrobin, the application must use rd_kafka_assign() to set or clear the entire assignment. For the cooperative assignors, such as cooperative-sticky, the application must use rd_kafka_incremental_assign() for RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS and rd_kafka_incremental_unassign() for RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS.

Without a rebalance callback this is done automatically by librdkafka but registering a rebalance callback gives the application flexibility in performing other operations along with the assigning/revocation, such as fetching offsets from an alternate location (on assign) or manually committing offsets (on revoke).

rebalance_cb is always triggered exactly once when a rebalance completes with a new assignment, even if that assignment is empty. If an eager/non-cooperative assignor is configured, there will eventually be exactly one corresponding call to rebalance_cb to revoke these partitions (even if empty), whether this is due to a group rebalance or lost partitions. In the cooperative case, rebalance_cb will never be called if the set of partitions being revoked is empty (whether or not lost).

The callback's opaque argument is the opaque set with rd_kafka_conf_set_opaque().

Remarks: The partitions list is destroyed by librdkafka on return return from the rebalance_cb and must not be freed or saved by the application.; Be careful when modifying the partitions list. Changing this list should only be done to change the initial offsets for each partition. But a function like rd_kafka_position() might have unexpected effects for instance when a consumer gets assigned a partition it used to consume at an earlier rebalance. In this case, the list of partitions will be updated with the old offset for that partition. In this case, it is generally better to pass a copy of the list (see rd_kafka_topic_partition_list_copy()). The result of rd_kafka_position() is typically outdated in RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS.

See also: rd_kafka_assign(); rd_kafka_incremental_assign(); rd_kafka_incremental_unassign(); rd_kafka_assignment_lost(); rd_kafka_rebalance_protocol()

The following example shows the application's responsibilities:

static void rebalance_cb (rd_kafka_t *rk, rd_kafka_resp_err_t err,
                          rd_kafka_topic_partition_list_t *partitions,
                          void *opaque) {
 
    switch (err)
    {
      case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
         // application may load offets from arbitrary external
         // storage here and update \p partitions
         if (!strcmp(rd_kafka_rebalance_protocol(rk), "COOPERATIVE"))
                 rd_kafka_incremental_assign(rk, partitions);
         else // EAGER
                 rd_kafka_assign(rk, partitions);
         break;
 
      case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
         if (manual_commits) // Optional explicit manual commit
             rd_kafka_commit(rk, partitions, 0); // sync commit
 
         if (!strcmp(rd_kafka_rebalance_protocol(rk), "COOPERATIVE"))
                 rd_kafka_incremental_unassign(rk, partitions);
         else // EAGER
                 rd_kafka_assign(rk, NULL);
         break;
 
      default:
         handle_unlikely_error(err);
         rd_kafka_assign(rk, NULL); // sync state
         break;
     }
}

Remarks: The above example lacks error handling for assign calls, see the examples/ directory.

◆ rd_kafka_conf_set_offset_commit_cb()

RD_EXPORT void rd_kafka_conf_set_offset_commit_cb	(	rd_kafka_conf_t *	conf,
		void()(rd_kafka_t rk, rd_kafka_resp_err_t err, rd_kafka_topic_partition_list_t offsets, void opaque)	offset_commit_cb
	)

Consumer: Set offset commit callback for use with consumer groups.

The results of automatic or manual offset commits will be scheduled for this callback and is served by rd_kafka_consumer_poll().

If no partitions had valid offsets to commit this callback will be called with err == RD_KAFKA_RESP_ERR__NO_OFFSET which is not to be considered an error.

The offsets list contains per-partition information:

offset: committed offset (attempted)
err: commit error

The callback's opaque argument is the opaque set with rd_kafka_conf_set_opaque().

◆ rd_kafka_conf_set_error_cb()

RD_EXPORT void rd_kafka_conf_set_error_cb	(	rd_kafka_conf_t *	conf,
		void()(rd_kafka_t rk, int err, const char reason, void opaque)	error_cb
	)

Set error callback in provided conf object.

The error callback is used by librdkafka to signal warnings and errors back to the application.

These errors should generally be considered informational and non-permanent, the client will try to recover automatically from all type of errors. Given that the client and cluster configuration is correct the application should treat these as temporary errors.

error_cb will be triggered with err set to RD_KAFKA_RESP_ERR__FATAL if a fatal error has been raised; in this case use rd_kafka_fatal_error() to retrieve the fatal error code and error string, and then begin terminating the client instance.

If no error_cb is registered, or RD_KAFKA_EVENT_ERROR has not been set with rd_kafka_conf_set_events, then the errors will be logged instead.

The callback's opaque argument is the opaque set with rd_kafka_conf_set_opaque().

◆ rd_kafka_conf_set_throttle_cb()

RD_EXPORT void rd_kafka_conf_set_throttle_cb	(	rd_kafka_conf_t *	conf,
		void()(rd_kafka_t rk, const char broker_name, int32_t broker_id, int throttle_time_ms, void opaque)	throttle_cb
	)

Set throttle callback.

The throttle callback is used to forward broker throttle times to the application for Produce and Fetch (consume) requests.

Callbacks are triggered whenever a non-zero throttle time is returned by the broker, or when the throttle time drops back to zero.

An application must call rd_kafka_poll() or rd_kafka_consumer_poll() at regular intervals to serve queued callbacks.

The callback's opaque argument is the opaque set with rd_kafka_conf_set_opaque().

Remarks: Requires broker version 0.9.0 or later.

◆ rd_kafka_conf_set_log_cb()

RD_EXPORT void rd_kafka_conf_set_log_cb	(	rd_kafka_conf_t *	conf,
		void()(const rd_kafka_t rk, int level, const char fac, const char buf)	log_cb
	)

Set logger callback.

The default is to print to stderr, but a syslog logger is also available, see rd_kafka_log_print and rd_kafka_log_syslog for the builtin alternatives. Alternatively the application may provide its own logger callback. Or pass func as NULL to disable logging.

This is the configuration alternative to the deprecated rd_kafka_set_logger()

Remarks: The log_cb will be called spontaneously from librdkafka's internal threads unless logs have been forwarded to a poll queue through rd_kafka_set_log_queue(). An application MUST NOT call any librdkafka APIs or do any prolonged work in a non-forwarded log_cb.

◆ rd_kafka_conf_set_stats_cb()

RD_EXPORT void rd_kafka_conf_set_stats_cb	(	rd_kafka_conf_t *	conf,
		int()(rd_kafka_t rk, char json, size_t json_len, void opaque)	stats_cb
	)

Set statistics callback in provided conf object.

The statistics callback is triggered from rd_kafka_poll() every statistics.interval.ms (needs to be configured separately). Function arguments:

rk - Kafka handle
json - String containing the statistics data in JSON format
json_len - Length of json string.
opaque - Application-provided opaque as set by rd_kafka_conf_set_opaque().

For more information on the format of json, see https://github.com/edenhill/librdkafka/wiki/Statistics

If the application wishes to hold on to the json pointer and free it at a later time it must return 1 from the stats_cb. If the application returns 0 from the stats_cb then librdkafka will immediately free the json pointer.

See STATISTICS.md for a full definition of the JSON object.

◆ rd_kafka_conf_set_oauthbearer_token_refresh_cb()

RD_EXPORT void rd_kafka_conf_set_oauthbearer_token_refresh_cb	(	rd_kafka_conf_t *	conf,
		void()(rd_kafka_t rk, const char oauthbearer_config, void opaque)	oauthbearer_token_refresh_cb
	)

Set SASL/OAUTHBEARER token refresh callback in provided conf object.

Parameters

conf	the configuration to mutate.
oauthbearer_token_refresh_cb	the callback to set; callback function arguments: `rk` - Kafka handle `oauthbearer_config` - Value of configuration property sasl.oauthbearer.config. `opaque` - Application-provided opaque set via rd_kafka_conf_set_opaque()

The SASL/OAUTHBEARER token refresh callback is triggered via rd_kafka_poll() whenever OAUTHBEARER is the SASL mechanism and a token needs to be retrieved, typically based on the configuration defined in sasl.oauthbearer.config.

The callback should invoke rd_kafka_oauthbearer_set_token() or rd_kafka_oauthbearer_set_token_failure() to indicate success or failure, respectively.

The refresh operation is eventable and may be received via rd_kafka_queue_poll() with an event type of RD_KAFKA_EVENT_OAUTHBEARER_TOKEN_REFRESH.

Note that before any SASL/OAUTHBEARER broker connection can succeed the application must call rd_kafka_oauthbearer_set_token() once – either directly or, more typically, by invoking either rd_kafka_poll(), rd_kafka_consumer_poll(), rd_kafka_queue_poll(), etc, in order to cause retrieval of an initial token to occur.

Alternatively, the application can enable the SASL queue by calling rd_kafka_conf_enable_sasl_queue() on the configuration object prior to creating the client instance, get the SASL queue with rd_kafka_queue_get_sasl(), and either serve the queue manually by calling rd_kafka_queue_poll(), or redirecting the queue to the background thread to have the queue served automatically. For the latter case the SASL queue must be forwarded to the background queue with rd_kafka_queue_forward(). A convenience function is available to automatically forward the SASL queue to librdkafka's background thread, see rd_kafka_sasl_background_callbacks_enable().

An unsecured JWT refresh handler is provided by librdkafka for development and testing purposes, it is enabled by setting the enable.sasl.oauthbearer.unsecure.jwt property to true and is mutually exclusive to using a refresh callback.

See also: rd_kafka_sasl_background_callbacks_enable(); rd_kafka_queue_get_sasl()

◆ rd_kafka_conf_enable_sasl_queue()

RD_EXPORT void rd_kafka_conf_enable_sasl_queue	(	rd_kafka_conf_t *	conf,
		int	enable
	)

Enable/disable creation of a queue specific to SASL events and callbacks.

For SASL mechanisms that trigger callbacks (currently OAUTHBEARER) this configuration API allows an application to get a dedicated queue for the SASL events/callbacks. After enabling the queue with this API the application can retrieve the queue by calling rd_kafka_queue_get_sasl() on the client instance. This queue may then be served directly by the application (with rd_kafka_queue_poll(), et.al) or forwarded to another queue, such as the background queue.

A convenience function is available to automatically forward the SASL queue to librdkafka's background thread, see rd_kafka_sasl_background_callbacks_enable().

By default (enable = 0) the main queue (as served by rd_kafka_poll(), et.al.) is used for SASL callbacks.

Remarks: The SASL queue is currently only used by the SASL OAUTHBEARER mechanism's token_refresh_cb().

See also: rd_kafka_queue_get_sasl(); rd_kafka_sasl_background_callbacks_enable()

◆ rd_kafka_conf_set_socket_cb()

RD_EXPORT void rd_kafka_conf_set_socket_cb	(	rd_kafka_conf_t *	conf,
		int()(int domain, int type, int protocol, void opaque)	socket_cb
	)

Set socket callback.

The socket callback is responsible for opening a socket according to the supplied domain, type and protocol. The socket shall be created with CLOEXEC set in a racefree fashion, if possible.

The callback's opaque argument is the opaque set with rd_kafka_conf_set_opaque().

Default:

on linux: racefree CLOEXEC
others : non-racefree CLOEXEC

Remarks: The callback will be called from an internal librdkafka thread.

◆ rd_kafka_conf_set_connect_cb()

RD_EXPORT void rd_kafka_conf_set_connect_cb	(	rd_kafka_conf_t *	conf,
		int()(int sockfd, const struct sockaddr addr, int addrlen, const char id, void opaque)	connect_cb
	)

Set connect callback.

The connect callback is responsible for connecting socket sockfd to peer address addr. The id field contains the broker identifier.

connect_cb shall return 0 on success (socket connected) or an error number (errno) on error.

The callback's opaque argument is the opaque set with rd_kafka_conf_set_opaque().

Remarks: The callback will be called from an internal librdkafka thread.

◆ rd_kafka_conf_set_closesocket_cb()

RD_EXPORT void rd_kafka_conf_set_closesocket_cb	(	rd_kafka_conf_t *	conf,
		int()(int sockfd, void opaque)	closesocket_cb
	)

Set close socket callback.

Close a socket (optionally opened with socket_cb()).

The callback's opaque argument is the opaque set with rd_kafka_conf_set_opaque().

Remarks: The callback will be called from an internal librdkafka thread.

◆ rd_kafka_conf_set_open_cb()

RD_EXPORT void rd_kafka_conf_set_open_cb	(	rd_kafka_conf_t *	conf,
		int()(const char pathname, int flags, mode_t mode, void *opaque)	open_cb
	)

Set open callback.

The open callback is responsible for opening the file specified by pathname, flags and mode. The file shall be opened with CLOEXEC set in a racefree fashion, if possible.

Default:

on linux: racefree CLOEXEC
others : non-racefree CLOEXEC

The callback's opaque argument is the opaque set with rd_kafka_conf_set_opaque().

Remarks: The callback will be called from an internal librdkafka thread.

◆ rd_kafka_conf_set_ssl_cert_verify_cb()

RD_EXPORT rd_kafka_conf_res_t rd_kafka_conf_set_ssl_cert_verify_cb	(	rd_kafka_conf_t *	conf,
		int()(rd_kafka_t rk, const char broker_name, int32_t broker_id, int x509_error, int depth, const char buf, size_t size, char errstr, size_t errstr_size, void *opaque)	ssl_cert_verify_cb
	)

Sets the verification callback of the broker certificate.

The verification callback is triggered from internal librdkafka threads upon connecting to a broker. On each connection attempt the callback will be called for each certificate in the broker's certificate chain, starting at the root certification, as long as the application callback returns 1 (valid certificate). broker_name and broker_id correspond to the broker the connection is being made to. The x509_error argument indicates if OpenSSL's verification of the certificate succeed (0) or failed (an OpenSSL error code). The application may set the SSL context error code by returning 0 from the verify callback and providing a non-zero SSL context error code in x509_error. If the verify callback sets x509_error to 0, returns 1, and the original x509_error was non-zero, the error on the SSL context will be cleared. x509_error is always a valid pointer to an int.

depth is the depth of the current certificate in the chain, starting at the root certificate.

The certificate itself is passed in binary DER format in buf of size size.

The callback must return 1 if verification succeeds, or 0 if verification fails and then write a human-readable error message to errstr (limited to errstr_size bytes, including nul-term).

The callback's opaque argument is the opaque set with rd_kafka_conf_set_opaque().

Returns: RD_KAFKA_CONF_OK if SSL is supported in this build, else RD_KAFKA_CONF_INVALID.

Warning: This callback will be called from internal librdkafka threads.

Remarks: See <openssl/x509_vfy.h> in the OpenSSL source distribution for a list of x509_error codes.

◆ rd_kafka_conf_set_ssl_cert()

RD_EXPORT rd_kafka_conf_res_t rd_kafka_conf_set_ssl_cert	(	rd_kafka_conf_t *	conf,
		rd_kafka_cert_type_t	cert_type,
		rd_kafka_cert_enc_t	cert_enc,
		const void *	buffer,
		size_t	size,
		char *	errstr,
		size_t	errstr_size
	)

Set certificate/key cert_type from the cert_enc encoded memory at buffer of size bytes.

Parameters

conf	Configuration object.
cert_type	Certificate or key type to configure.
cert_enc	Buffer `encoding` type.
buffer	Memory pointer to encoded certificate or key. The memory is not referenced after this function returns.
size	Size of memory at `buffer`.
errstr	Memory were a human-readable error string will be written on failure.
errstr_size	Size of `errstr`, including space for nul-terminator.

Returns: RD_KAFKA_CONF_OK on success or RD_KAFKA_CONF_INVALID if the memory in buffer is of incorrect encoding, or if librdkafka was not built with SSL support.

Remarks: Calling this method multiple times with the same cert_type will replace the previous value.; Calling this method with buffer set to NULL will clear the configuration for cert_type.; The private key may require a password, which must be specified with the ssl.key.password configuration property prior to calling this function.; Private and public keys in PEM format may also be set with the ssl.key.pem and ssl.certificate.pem configuration properties.; CA certificate in PEM format may also be set with the ssl.ca.pem configuration property.

◆ rd_kafka_conf_set_engine_callback_data()

RD_EXPORT void rd_kafka_conf_set_engine_callback_data	(	rd_kafka_conf_t *	conf,
		void *	callback_data
	)

Set callback_data for OpenSSL engine.

Parameters

conf	Configuration object.
callback_data	passed to engine callbacks, e.g. `ENGINE_load_ssl_client_cert`.

Remarks: The ssl.engine.location configuration must be set for this to have affect.; The memory pointed to by value must remain valid for the lifetime of the configuration object and any Kafka clients that use it.

◆ rd_kafka_conf_set_opaque()

RD_EXPORT void rd_kafka_conf_set_opaque	(	rd_kafka_conf_t *	conf,
		void *	opaque
	)

Sets the application's opaque pointer that will be passed to callbacks.

See also: rd_kafka_opaque()

◆ rd_kafka_conf_set_default_topic_conf()

RD_EXPORT void rd_kafka_conf_set_default_topic_conf	(	rd_kafka_conf_t *	conf,
		rd_kafka_topic_conf_t *	tconf
	)

Sets the default topic configuration to use for automatically subscribed topics (e.g., through pattern-matched topics). The topic config object is not usable after this call.

Warning: Any topic configuration settings that have been set on the global rd_kafka_conf_t object will be overwritten by this call since the implicitly created default topic config object is replaced by the user-supplied one.

Deprecated:: Set default topic level configuration on the global rd_kafka_conf_t object instead.

◆ rd_kafka_conf_get_default_topic_conf()

RD_EXPORT rd_kafka_topic_conf_t* rd_kafka_conf_get_default_topic_conf ( rd_kafka_conf_t * conf )

Gets the default topic configuration as previously set with rd_kafka_conf_set_default_topic_conf() or that was implicitly created by configuring a topic-level property on the global conf object.

Returns: the conf's default topic configuration (if any), or NULL.

Warning: The returned topic configuration object is owned by the conf object. It may be modified but not destroyed and its lifetime is the same as the conf object or the next call to rd_kafka_conf_set_default_topic_conf().

◆ rd_kafka_conf_get()

RD_EXPORT rd_kafka_conf_res_t rd_kafka_conf_get	(	const rd_kafka_conf_t *	conf,
		const char *	name,
		char *	dest,
		size_t *	dest_size
	)

Retrieve configuration value for property name.

If dest is non-NULL the value will be written to dest with at most dest_size.

*dest_size is updated to the full length of the value, thus if *dest_size initially is smaller than the full length the application may reallocate dest to fit the returned *dest_size and try again.

If dest is NULL only the full length of the value is returned.

Fallthrough: Topic-level configuration properties from the default_topic_conf may be retrieved using this interface.

Returns: RD_KAFKA_CONF_OK if the property name matched, else RD_KAFKA_CONF_UNKNOWN.

◆ rd_kafka_topic_conf_get()

RD_EXPORT rd_kafka_conf_res_t rd_kafka_topic_conf_get	(	const rd_kafka_topic_conf_t *	conf,
		const char *	name,
		char *	dest,
		size_t *	dest_size
	)

Retrieve topic configuration value for property name.

See also: rd_kafka_conf_get()

◆ rd_kafka_conf_dump()

const RD_EXPORT char** rd_kafka_conf_dump	(	rd_kafka_conf_t *	conf,
		size_t *	cntp
	)

Dump the configuration properties and values of conf to an array with "key", "value" pairs.

The number of entries in the array is returned in *cntp.

The dump must be freed with rd_kafka_conf_dump_free().

◆ rd_kafka_topic_conf_dump()

const RD_EXPORT char** rd_kafka_topic_conf_dump	(	rd_kafka_topic_conf_t *	conf,
		size_t *	cntp
	)

Dump the topic configuration properties and values of conf to an array with "key", "value" pairs.

The number of entries in the array is returned in *cntp.

The dump must be freed with rd_kafka_conf_dump_free().

◆ rd_kafka_conf_properties_show()

RD_EXPORT void rd_kafka_conf_properties_show ( FILE * fp )

Prints a table to fp of all supported configuration properties, their default values as well as a description.

Remarks: All properties and properties and values are shown, even those that have been disabled at build time due to missing dependencies.

◆ rd_kafka_topic_conf_new()

RD_EXPORT rd_kafka_topic_conf_t* rd_kafka_topic_conf_new ( void )

Create topic configuration object.

See also: Same semantics as for rd_kafka_conf_new().

◆ rd_kafka_topic_conf_set()

RD_EXPORT rd_kafka_conf_res_t rd_kafka_topic_conf_set	(	rd_kafka_topic_conf_t *	conf,
		const char *	name,
		const char *	value,
		char *	errstr,
		size_t	errstr_size
	)

Sets a single rd_kafka_topic_conf_t value by property name.

topic_conf should have been previously set up with rd_kafka_topic_conf_new().

Returns: rd_kafka_conf_res_t to indicate success or failure.

◆ rd_kafka_topic_conf_set_opaque()

RD_EXPORT void rd_kafka_topic_conf_set_opaque	(	rd_kafka_topic_conf_t *	conf,
		void *	rkt_opaque
	)

Sets the application's opaque pointer that will be passed to all topic callbacks as the rkt_opaque argument.

See also: rd_kafka_topic_opaque()

◆ rd_kafka_topic_conf_set_partitioner_cb()

RD_EXPORT void rd_kafka_topic_conf_set_partitioner_cb	(	rd_kafka_topic_conf_t *	topic_conf,
		int32_t()(const rd_kafka_topic_t rkt, const void keydata, size_t keylen, int32_t partition_cnt, void rkt_opaque, void *msg_opaque)	partitioner
	)

Producer: Set partitioner callback in provided topic conf object.

The partitioner may be called in any thread at any time, it may be called multiple times for the same message/key.

The callback's rkt_opaque argument is the opaque set by rd_kafka_topic_conf_set_opaque(). The callback's msg_opaque argument is the per-message opaque passed to produce().

Partitioner function constraints:

MUST NOT call any rd_kafka_*() functions except: rd_kafka_topic_partition_available()
MUST NOT block or execute for prolonged periods of time.
MUST return a value between 0 and partition_cnt-1, or the special RD_KAFKA_PARTITION_UA value if partitioning could not be performed.

◆ rd_kafka_topic_conf_set_msg_order_cmp()

RD_EXPORT void rd_kafka_topic_conf_set_msg_order_cmp	(	rd_kafka_topic_conf_t *	topic_conf,
		int()(const rd_kafka_message_t a, const rd_kafka_message_t *b)	msg_order_cmp
	)

Producer: Set message queueing order comparator callback.

The callback may be called in any thread at any time, it may be called multiple times for the same message.

Ordering comparator function constraints:

MUST be stable sort (same input gives same output).
MUST NOT call any rd_kafka_*() functions.
MUST NOT block or execute for prolonged periods of time.

The comparator shall compare the two messages and return:

< 0 if message a should be inserted before message b.
>=0 if message a should be inserted after message b.

Remarks: Insert sorting will be used to enqueue the message in the correct queue position, this comes at a cost of O(n).; If queuing.strategy=fifo new messages are enqueued to the tail of the queue regardless of msg_order_cmp, but retried messages are still affected by msg_order_cmp.

Warning: THIS IS AN EXPERIMENTAL API, SUBJECT TO CHANGE OR REMOVAL, DO NOT USE IN PRODUCTION.

◆ rd_kafka_topic_partition_available()

RD_EXPORT int rd_kafka_topic_partition_available	(	const rd_kafka_topic_t *	rkt,
		int32_t	partition
	)

Check if partition is available (has a leader broker).

Returns: 1 if the partition is available, else 0.

Warning: This function must only be called from inside a partitioner function