librdkafka
The Apache Kafka C/C++ client library
 All Data Structures Files Functions Variables Typedefs Enumerations Enumerator Macros Pages
Data Structures | Macros | Typedefs | Functions
rdkafka.h File Reference

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

Go to the source code of this file.

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_t
 Topic+Partition place holder. More...
 
struct  rd_kafka_topic_partition_list_t
 A growable list of Topic+Partitions. More...
 
struct  rd_kafka_vu_t
 VTYPE + argument container for use with rd_kafka_produce_va() More...
 
struct  rd_kafka_message_t
 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_t
 Broker information. More...
 
struct  rd_kafka_metadata_partition_t
 Partition information. More...
 
struct  rd_kafka_metadata_topic_t
 Topic information. More...
 
struct  rd_kafka_metadata_t
 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)
 
RD_EXPORT const char * rd_kafka_AclBinding_name (const rd_kafka_AclBinding_t *acl)
 
RD_EXPORT const char * rd_kafka_AclBinding_principal (const rd_kafka_AclBinding_t *acl)
 
RD_EXPORT const 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)
 
RD_EXPORT const 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.
 
RD_EXPORT const
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...
 
RD_EXPORT const
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...
 
RD_EXPORT const
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...
 
RD_EXPORT const rd_kafka_error_t * rd_kafka_DeleteAcls_result_response_error (const rd_kafka_DeleteAcls_result_response_t *result_response)
 
RD_EXPORT const
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...
 
RD_EXPORT const 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...
 
RD_EXPORT const char * rd_kafka_topic_result_error_string (const rd_kafka_topic_result_t *topicres)
 
RD_EXPORT const char * rd_kafka_topic_result_name (const rd_kafka_topic_result_t *topicres)
 
RD_EXPORT const 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...
 
RD_EXPORT const char * rd_kafka_group_result_name (const rd_kafka_group_result_t *groupres)
 
RD_EXPORT const
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, 10*1000);
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...
 
RD_EXPORT const 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...
 
RD_EXPORT const 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.
 
RD_EXPORT const char * rd_kafka_err2str (rd_kafka_resp_err_t err)
 Returns a human readable representation of a kafka error. More...
 
RD_EXPORT const 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)
 
RD_EXPORT const char * rd_kafka_error_name (const rd_kafka_error_t *error)
 
RD_EXPORT const 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.
 
RD_EXPORT const 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.
 
RD_EXPORT const 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...
 
RD_EXPORT const 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...
 
RD_EXPORT const 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.
 
RD_EXPORT const 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...
 
RD_EXPORT const 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)
 
RD_EXPORT const 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...
 
RD_EXPORT const
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)
 
RD_EXPORT const 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)
 
RD_EXPORT const 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...
 
RD_EXPORT const 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)
 
RD_EXPORT const
rd_kafka_CreateTopics_result_t * 		rd_kafka_event_CreateTopics_result (rd_kafka_event_t *rkev)
 Get CreateTopics result. More...
 
RD_EXPORT const
rd_kafka_DeleteTopics_result_t * 		rd_kafka_event_DeleteTopics_result (rd_kafka_event_t *rkev)
 Get DeleteTopics result. More...
 
RD_EXPORT const
rd_kafka_CreatePartitions_result_t * 		rd_kafka_event_CreatePartitions_result (rd_kafka_event_t *rkev)
 Get CreatePartitions result. More...
 
RD_EXPORT const
rd_kafka_AlterConfigs_result_t * 		rd_kafka_event_AlterConfigs_result (rd_kafka_event_t *rkev)
 Get AlterConfigs result. More...
 
RD_EXPORT const
rd_kafka_DescribeConfigs_result_t * 		rd_kafka_event_DescribeConfigs_result (rd_kafka_event_t *rkev)
 Get DescribeConfigs result. More...
 
RD_EXPORT const
rd_kafka_DeleteRecords_result_t * 		rd_kafka_event_DeleteRecords_result (rd_kafka_event_t *rkev)
 
RD_EXPORT const
rd_kafka_DeleteGroups_result_t * 		rd_kafka_event_DeleteGroups_result (rd_kafka_event_t *rkev)
 Get DeleteGroups result. More...
 
RD_EXPORT const
rd_kafka_DeleteConsumerGroupOffsets_result_t * 		rd_kafka_event_DeleteConsumerGroupOffsets_result (rd_kafka_event_t *rkev)
 Get DeleteConsumerGroupOffsets result. More...
 
RD_EXPORT const
rd_kafka_CreateAcls_result_t * 		rd_kafka_event_CreateAcls_result (rd_kafka_event_t *rkev)
 
RD_EXPORT const
rd_kafka_DescribeAcls_result_t * 		rd_kafka_event_DescribeAcls_result (rd_kafka_event_t *rkev)
 
RD_EXPORT const
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.
 
typedef rd_kafka_AclBinding_t rd_kafka_AclBindingFilter_t
 ACL Binding filter is used to filter access control lists.
 
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...
 
RD_EXPORT const
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...
 
RD_EXPORT const
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...
 
RD_EXPORT const
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...
 
RD_EXPORT const char * rd_kafka_ConfigSource_name (rd_kafka_ConfigSource_t confsource)
 
RD_EXPORT const char * rd_kafka_ConfigEntry_name (const rd_kafka_ConfigEntry_t *entry)
 
RD_EXPORT const 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)
 
RD_EXPORT const
rd_kafka_ConfigEntry_t ** 		rd_kafka_ConfigEntry_synonyms (const rd_kafka_ConfigEntry_t *entry, size_t *cntp)
 
RD_EXPORT const char * rd_kafka_ResourcePatternType_name (rd_kafka_ResourcePatternType_t resource_pattern_type)
 
RD_EXPORT const 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...
 
RD_EXPORT const
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)
 
RD_EXPORT const 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)
 
RD_EXPORT const 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...
 
RD_EXPORT const
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...
 
RD_EXPORT const
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...
 
RD_EXPORT const
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...
 
RD_EXPORT const
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...
 
RD_EXPORT const
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...
 
RD_EXPORT const 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...
 
RD_EXPORT const 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...
 
RD_EXPORT const 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

#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()
#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.
#define RD_KAFKA_RESP_ERR_GROUP_LOAD_IN_PROGRESS   RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS

Group coordinator load in progress

#define RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE   RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE

Group coordinator not available

#define RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP   RD_KAFKA_RESP_ERR_NOT_COORDINATOR

Not coordinator for group

#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

#define RD_KAFKA_V_TOPIC (   topic)
Value:
_LRK_TYPECHECK(RD_KAFKA_VTYPE_TOPIC, const char *, topic), \
(const char *)topic
RD_KAFKA_VTYPE_TOPIC
Definition: rdkafka.h:1088

Topic name (const char *)

rd_kafka_vu_t field: u.cstr

#define RD_KAFKA_V_RKT (   rkt)
Value:
_LRK_TYPECHECK(RD_KAFKA_VTYPE_RKT, rd_kafka_topic_t *, rkt), \
(rd_kafka_topic_t *)rkt
RD_KAFKA_VTYPE_RKT
Definition: rdkafka.h:1089

Topic object (rd_kafka_topic_t *)

rd_kafka_vu_t field: u.rkt

#define RD_KAFKA_V_PARTITION (   partition)
Value:
_LRK_TYPECHECK(RD_KAFKA_VTYPE_PARTITION, int32_t, partition), \
(int32_t)partition
RD_KAFKA_VTYPE_PARTITION
Definition: rdkafka.h:1090

Partition (int32_t)

rd_kafka_vu_t field: u.i32

#define RD_KAFKA_V_VALUE (   VALUE,
  LEN 
)
Value:
_LRK_TYPECHECK2(RD_KAFKA_VTYPE_VALUE, void *, VALUE, size_t, LEN), \
(void *)VALUE, (size_t)LEN
RD_KAFKA_VTYPE_VALUE
Definition: rdkafka.h:1091

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

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

#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
RD_KAFKA_VTYPE_KEY
Definition: rdkafka.h:1092

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

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

#define RD_KAFKA_V_OPAQUE (   msg_opaque)
Value:
_LRK_TYPECHECK(RD_KAFKA_VTYPE_OPAQUE, void *, msg_opaque), \
(void *)msg_opaque
RD_KAFKA_VTYPE_OPAQUE
Definition: rdkafka.h:1093

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

rd_kafka_vu_t field: u.ptr

#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

#define RD_KAFKA_V_TIMESTAMP (   timestamp)
Value:
_LRK_TYPECHECK(RD_KAFKA_VTYPE_TIMESTAMP, int64_t, timestamp), \
(int64_t)timestamp
RD_KAFKA_VTYPE_TIMESTAMP
Definition: rdkafka.h:1099

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

#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
RD_KAFKA_VTYPE_HEADER
Definition: rdkafka.h:1100

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

#define RD_KAFKA_V_HEADERS (   HDRS)
Value:
_LRK_TYPECHECK(RD_KAFKA_VTYPE_HEADERS, rd_kafka_headers_t *, HDRS), \
(rd_kafka_headers_t *)HDRS
RD_KAFKA_VTYPE_HEADERS
Definition: rdkafka.h:1102

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

#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.

#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.

#define RD_KAFKA_OFFSET_BEGINNING   -2

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

#define RD_KAFKA_OFFSET_END   -1

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

#define RD_KAFKA_OFFSET_STORED   -1000

Start consuming from offset retrieved \ from offset store

#define RD_KAFKA_OFFSET_INVALID   -1001

Invalid offset

#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.

#define RD_KAFKA_MSG_F_FREE   0x1

Producer message flags.

Delegate freeing of payload to rdkafka. \

#define RD_KAFKA_MSG_F_COPY   0x2

rdkafka will make a copy of the payload. \

#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.

#define RD_KAFKA_MSG_F_PARTITION   0x8

produce_batch() will honor \ per-message partition.

#define RD_KAFKA_PURGE_F_QUEUE   0x1

Flags for rd_kafka_purge()

Purge messages in internal queues.

#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.

#define RD_KAFKA_PURGE_F_NON_BLOCKING   0x4

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

#define RD_KAFKA_EVENT_NONE   0x0

Unset value

#define RD_KAFKA_EVENT_DR   0x1

Producer Delivery report batch

#define RD_KAFKA_EVENT_FETCH   0x2

Fetched message (consumer)

#define RD_KAFKA_EVENT_LOG   0x4

Log message

#define RD_KAFKA_EVENT_ERROR   0x8

Error

#define RD_KAFKA_EVENT_REBALANCE   0x10

Group rebalance (consumer)

#define RD_KAFKA_EVENT_OFFSET_COMMIT   0x20

Offset commit result

#define RD_KAFKA_EVENT_STATS   0x40

Stats

#define RD_KAFKA_EVENT_CREATETOPICS_RESULT   100

CreateTopics_result_t

#define RD_KAFKA_EVENT_DELETETOPICS_RESULT   101

DeleteTopics_result_t

#define RD_KAFKA_EVENT_CREATEPARTITIONS_RESULT   102

CreatePartitions_result_t

#define RD_KAFKA_EVENT_ALTERCONFIGS_RESULT   103

AlterConfigs_result_t

#define RD_KAFKA_EVENT_DESCRIBECONFIGS_RESULT   104

DescribeConfigs_result_t

#define RD_KAFKA_EVENT_DELETERECORDS_RESULT   105

DeleteRecords_result_t

#define RD_KAFKA_EVENT_DELETEGROUPS_RESULT   106

DeleteGroups_result_t

#define RD_KAFKA_EVENT_DELETECONSUMERGROUPOFFSETS_RESULT   107

DeleteConsumerGroupOffsets_result_t

#define RD_KAFKA_EVENT_OAUTHBEARER_TOKEN_REFRESH   0x100

SASL/OAUTHBEARER token needs to be refreshed

#define RD_KAFKA_EVENT_BACKGROUND   0x200

Enable background thread.

#define RD_KAFKA_EVENT_CREATEACLS_RESULT   0x400

CreateAcls_result_t

#define RD_KAFKA_EVENT_DESCRIBEACLS_RESULT   0x800

DescribeAcls_result_t

#define RD_KAFKA_EVENT_DELETEACLS_RESULT   0x1000

DeleteAcls_result_t

Typedef Documentation

typedef rd_kafka_event_t rd_kafka_CreateTopics_result_t

CreateTopics result type

typedef rd_kafka_event_t rd_kafka_DeleteTopics_result_t

DeleteTopics result type

typedef rd_kafka_event_t rd_kafka_CreateAcls_result_t

CreateAcls result type

typedef rd_kafka_event_t rd_kafka_DescribeAcls_result_t

DescribeAcls result type

typedef rd_kafka_event_t rd_kafka_DeleteAcls_result_t

DeleteAcls result type

typedef rd_kafka_event_t rd_kafka_CreatePartitions_result_t

CreatePartitions result type

typedef rd_kafka_event_t rd_kafka_AlterConfigs_result_t

AlterConfigs result type

typedef rd_kafka_event_t rd_kafka_DescribeConfigs_result_t

CreateTopics result type

typedef rd_kafka_event_t rd_kafka_DeleteRecords_result_t

DeleteRecords result type

typedef rd_kafka_event_t rd_kafka_DeleteGroups_result_t

DeleteGroups result type

typedef rd_kafka_event_t rd_kafka_DeleteConsumerGroupOffsets_result_t

DeleteConsumerGroupOffsets result type

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
confConfiguration set up to this point.
plug_opaquepPlugin can set this pointer to a per-configuration opaque pointer.
errstrString buffer of size errstr_size where plugin must write a human readable error string in the case the initializer fails (returns non-zero).
errstr_sizeMaximum 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.
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
confConfiguration object.
ic_opaqueThe interceptor's opaque pointer specified in ..add..().
nameThe configuration property to set.
valThe configuration value to set, or NULL for reverting to default in which case the previous value should be freed.
errstrA human readable error string in case the interceptor fails.
errstr_sizeMaximum 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.
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_opaqueThe interceptor's opaque pointer specified in ..add..().
new_confNew configuration object.
old_confOld configuration object to copy properties from.
filter_cntNumber of property names to filter in filter.
filterProperty 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().
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_opaqueThe interceptor's opaque pointer specified in ..add..().
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
rkThe client instance.
confThe client instance's final configuration.
ic_opaqueThe interceptor's opaque pointer specified in ..add..().
errstrA human readable error string in case the interceptor fails.
errstr_sizeMaximum 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..().
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
rkThe client instance.
ic_opaqueThe interceptor's opaque pointer specified in ..add..().
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
rkThe client instance.
rkmessageThe message being produced. Immutable.
ic_opaqueThe 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.
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
rkThe client instance.
rkmessageThe message being produced. Immutable.
ic_opaqueThe 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.
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
rkThe client instance.
rkmessageThe message being consumed. Immutable.
ic_opaqueThe 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.
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
rkThe client instance.
offsetsList of topic+partition+offset+error that were committed. The error message of each partition should be checked for error.
errThe commit error, if any.
ic_opaqueThe 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.
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
rkThe client instance.
sockfdSocket file descriptor.
brokernameBroker request is being sent to.
brokeridBroker request is being sent to.
ApiKeyKafka protocol request type.
ApiVersionKafka protocol request type version.
CorrIdKafka protocol request correlation id.
sizeSize of request.
ic_opaqueThe 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.
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
rkThe client instance.
sockfdSocket file descriptor (always -1).
brokernameBroker response was received from, possibly empty string on error.
brokeridBroker response was received from.
ApiKeyKafka protocol request type or -1 on error.
ApiVersionKafka protocol request type version or -1 on error.
CorrIdKafka protocol request correlation id, possibly -1 on error.
sizeSize of response, possibly 0 on error.
rttRequest round-trip-time in microseconds, possibly -1 on error.
errReceive error.
ic_opaqueThe 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.
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
rkThe client instance.
thread_typeThread type.
thread_nameHuman-readable thread name, may not be unique.
ic_opaqueThe 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.
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
rkThe client instance.
thread_typeThread type.n
thread_nameHuman-readable thread name, may not be unique.
ic_opaqueThe 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.
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.
typedef struct rd_kafka_NewTopic_s rd_kafka_NewTopic_t

Defines a new topic to be created.

typedef struct rd_kafka_DeleteTopic_s rd_kafka_DeleteTopic_t

Represents a topic to be deleted.

typedef struct rd_kafka_NewPartitions_s rd_kafka_NewPartitions_t

Defines a new partition to be created.

typedef struct rd_kafka_ConfigEntry_s rd_kafka_ConfigEntry_t

Apache Kafka configuration entry.

typedef struct rd_kafka_ConfigResource_s rd_kafka_ConfigResource_t

Apache Kafka configuration resource.

typedef struct rd_kafka_DeleteRecords_s rd_kafka_DeleteRecords_t

! Represents records to be deleted

typedef struct rd_kafka_DeleteGroup_s rd_kafka_DeleteGroup_t

Represents a group to be deleted.

typedef struct rd_kafka_DeleteConsumerGroupOffsets_s rd_kafka_DeleteConsumerGroupOffsets_t

Represents consumer group committed offsets to be deleted.

typedef struct rd_kafka_DeleteAcls_result_response_s rd_kafka_DeleteAcls_result_response_t

- delete access control lists.

Enumeration Type Documentation

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

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

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

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

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.

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

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

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

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

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

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

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

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

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

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_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_EXPORT const char* rd_kafka_version_str ( void  )

Returns the librdkafka version as string.

Returns
Version string
RD_EXPORT const 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_EXPORT const char* rd_kafka_err2str ( rd_kafka_resp_err_t  err)

Returns a human readable representation of a kafka error.

Parameters
errError code to translate
RD_EXPORT const char* rd_kafka_err2name ( rd_kafka_resp_err_t  err)

Returns the error code name (enum name).

Parameters
errError code to translate
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_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
errnoxSystem 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_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_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
rkClient instance.
errstrA 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_sizeWritable size in errstr.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR if no fatal error has been raised, else any other error code.
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
rkClient instance.
errThe underlying error code.
reasonA 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_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_EXPORT const 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_EXPORT const 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_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_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_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_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_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_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_EXPORT rd_kafka_topic_partition_list_t* rd_kafka_topic_partition_list_new ( int  size)

Create a new list/vector Topic+Partition container.

Parameters
sizeInitial 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_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
rktparlistList to extend
topicTopic name (copied)
partitionPartition id
Returns
The object which can be used to fill in additionals fields.
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
rktparlistList to extend
topicTopic name (copied)
startStart partition of range
stopLast partition of range (inclusive)
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
rktparlistList to modify
topicTopic name to match
partitionPartition 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_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_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
srcThe existing list to copy.
Returns
A new list fully populated to be identical to src
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_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_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_EXPORT rd_kafka_headers_t* rd_kafka_headers_new ( size_t  initial_count)

Create a new headers list.

Parameters
initial_countPreallocate space for this number of headers. Any number of headers may be added, updated and removed regardless of the initial count.
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
hdrsHeaders list.
nameHeader name.
name_sizeHeader name size (not including the null-terminator). If -1 the name length is automatically acquired using strlen().
valuePointer to header value, or NULL (set size to 0 or -1).
value_sizeSize 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_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_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
hdrsHeaders list.
nameHeader 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_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
hdrsHeaders to iterate.
idxIterator index, start at 0 and increment by one for each call as long as RD_KAFKA_RESP_ERR_NO_ERROR is returned.
nameHeader 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_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_EXPORT const 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_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_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_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_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_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_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
rkmessageThe message to set headers.
hdrsNew 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_EXPORT size_t rd_kafka_header_cnt ( const rd_kafka_headers_t *  hdrs)

Returns the number of header key/value pairs.

Parameters
hdrsHeaders to count
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_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.:

1 rd_kafka_conf_t *myconf;
2 rd_kafka_conf_res_t res;
3 
4 myconf = rd_kafka_conf_new();
5 res = rd_kafka_conf_set(myconf, "socket.timeout.ms", "600",
6  errstr, sizeof(errstr));
7 if (res != RD_KAFKA_CONF_OK)
8  die("%s\n", errstr);
9 
10 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_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_EXPORT const 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_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_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:

  1. First set the event callback on the configuration object with this function, followed by creating an rd_kafka_t instance with rd_kafka_new().
  2. 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().
  3. As the response event is ready and enqueued on the background queue the event callback will be triggered from the background thread.
  4. 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_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_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_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_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:

1 static void rebalance_cb (rd_kafka_t *rk, rd_kafka_resp_err_t err,
2  rd_kafka_topic_partition_list_t *partitions,
3  void *opaque) {
4 
5  switch (err)
6  {
7  case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
8  // application may load offets from arbitrary external
9  // storage here and update \p partitions
10  if (!strcmp(rd_kafka_rebalance_protocol(rk), "COOPERATIVE"))
11  rd_kafka_incremental_assign(rk, partitions);
12  else // EAGER
13  rd_kafka_assign(rk, partitions);
14  break;
15 
16  case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
17  if (manual_commits) // Optional explicit manual commit
18  rd_kafka_commit(rk, partitions, 0); // sync commit
19 
20  if (!strcmp(rd_kafka_rebalance_protocol(rk), "COOPERATIVE"))
21  rd_kafka_incremental_unassign(rk, partitions);
22  else // EAGER
23  rd_kafka_assign(rk, NULL);
24  break;
25 
26  default:
27  handle_unlikely_error(err);
28  rd_kafka_assign(rk, NULL); // sync state
29  break;
30  }
31 }
Remarks
The above example lacks error handling for assign calls, see the examples/ directory.
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_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_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_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_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_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
confthe configuration to mutate.
oauthbearer_token_refresh_cbthe 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_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_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_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_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_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_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_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
confConfiguration object.
cert_typeCertificate or key type to configure.
cert_encBuffer encoding type.
bufferMemory pointer to encoded certificate or key. The memory is not referenced after this function returns.
sizeSize of memory at buffer.
errstrMemory were a human-readable error string will be written on failure.
errstr_sizeSize 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_EXPORT void rd_kafka_conf_set_engine_callback_data ( rd_kafka_conf_t *  conf,
void *  callback_data 
)

Set callback_data for OpenSSL engine.

Parameters
confConfiguration object.
callback_datapassed 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_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_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_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_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_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_EXPORT const 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_EXPORT const 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_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_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_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_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_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_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_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
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.

Will try not to return unavailable partitions.

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

Returns
a random partition between 0 and partition_cnt - 1.
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.

Uses consistent hashing to map identical keys onto identical partitions.

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

Returns
a "random" partition between 0 and partition_cnt - 1 based on the CRC value of the key
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.

This is the default partitioner. Uses consistent hashing to map identical keys onto identical partitions, and messages without keys will be assigned via the random partitioner.

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

Returns
a "random" partition between 0 and partition_cnt - 1 based on the CRC value of the key (if provided)
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).

Uses consistent hashing to map identical keys onto identical partitions using Java-compatible Murmur2 hashing.

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

Returns
a partition between 0 and partition_cnt - 1.
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).

Uses consistent hashing to map identical keys onto identical partitions using Java-compatible Murmur2 hashing. Messages without keys will be assigned via the random partitioner.

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

Returns
a partition between 0 and partition_cnt - 1.
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.

Uses consistent hashing to map identical keys onto identical partitions using FNV-1a hashing.

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

Returns
a partition between 0 and partition_cnt - 1.
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.

Uses consistent hashing to map identical keys onto identical partitions using FNV-1a hashing. Messages without keys will be assigned via the random partitioner.

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

Returns
a partition between 0 and partition_cnt - 1.
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).

conf is an optional struct created with rd_kafka_conf_new() that will be used instead of the default configuration. The conf object is freed by this function on success and must not be used or destroyed by the application sub-sequently. See rd_kafka_conf_set() et.al for more information.

errstr must be a pointer to memory of at least size errstr_size where rd_kafka_new() may write a human readable error message in case the creation of a new handle fails. In which case the function returns NULL.

Remarks
RD_KAFKA_CONSUMER: When a new RD_KAFKA_CONSUMER rd_kafka_t handle is created it may either operate in the legacy simple consumer mode using the rd_kafka_consume_start() interface, or the High-level KafkaConsumer API.
An application must only use one of these groups of APIs on a given rd_kafka_t RD_KAFKA_CONSUMER handle.
Returns
The Kafka handle on success or NULL on error (see errstr)
See also
To destroy the Kafka handle, use rd_kafka_destroy().
RD_EXPORT void rd_kafka_destroy ( rd_kafka_t *  rk)

Destroy Kafka handle.

Remarks
This is a blocking operation.
rd_kafka_consumer_close() will be called from this function if the instance type is RD_KAFKA_CONSUMER, a group.id was configured, and the rd_kafka_consumer_close() was not explicitly called by the application. This in turn may trigger consumer callbacks, such as rebalance_cb. Use rd_kafka_destroy_flags() with RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE to avoid this behaviour.
See also
rd_kafka_destroy_flags()
RD_EXPORT char* rd_kafka_memberid ( const rd_kafka_t *  rk)

Returns this client's broker-assigned group member id.

Remarks
This currently requires the high-level KafkaConsumer
Returns
An allocated string containing the current broker-assigned group member id, or NULL if not available. The application must free the string with free() or rd_kafka_mem_free()
RD_EXPORT char* rd_kafka_clusterid ( rd_kafka_t *  rk,
int  timeout_ms 
)

Returns the ClusterId as reported in broker metadata.

Parameters
rkClient instance.
timeout_msIf there is no cached value from metadata retrieval then this specifies the maximum amount of time (in milliseconds) the call will block waiting for metadata to be retrieved. Use 0 for non-blocking calls.
Remarks
Requires broker version >=0.10.0 and api.version.request=true.
The application must free the returned pointer using rd_kafka_mem_free().
Returns
a newly allocated string containing the ClusterId, or NULL if no ClusterId could be retrieved in the allotted timespan.
RD_EXPORT int32_t rd_kafka_controllerid ( rd_kafka_t *  rk,
int  timeout_ms 
)

Returns the current ControllerId as reported in broker metadata.

Parameters
rkClient instance.
timeout_msIf there is no cached value from metadata retrieval then this specifies the maximum amount of time (in milliseconds) the call will block waiting for metadata to be retrieved. Use 0 for non-blocking calls.
Remarks
Requires broker version >=0.10.0 and api.version.request=true.
Returns
the controller broker id (>= 0), or -1 if no ControllerId could be retrieved in the allotted timespan.
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.

conf is an optional configuration for the topic created with rd_kafka_topic_conf_new() that will be used instead of the default topic configuration. The conf object is freed by this function and must not be used or destroyed by the application sub-sequently. See rd_kafka_topic_conf_set() et.al for more information.

Topic handles are refcounted internally and calling rd_kafka_topic_new() again with the same topic name will return the previous topic handle without updating the original handle's configuration. Applications must eventually call rd_kafka_topic_destroy() for each succesfull call to rd_kafka_topic_new() to clear up resources.

Returns
the new topic handle or NULL on error (use rd_kafka_errno2err() to convert system errno to an rd_kafka_resp_err_t error code.
See also
rd_kafka_topic_destroy()
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().

Remarks
Since topic objects are refcounted (both internally and for the app) the topic object might not actually be destroyed by this call, but the application must consider the object destroyed.
RD_EXPORT int rd_kafka_poll ( rd_kafka_t *  rk,
int  timeout_ms 
)

Polls the provided kafka handle for events.

Events will cause application provided callbacks to be called.

The timeout_ms argument specifies the maximum amount of time (in milliseconds) that the call will block waiting for events. For non-blocking calls, provide 0 as timeout_ms. To wait indefinately for an event, provide -1.

Remarks
An application should make sure to call poll() at regular intervals to serve any queued callbacks waiting to be called.
If your producer doesn't have any callback set (in particular via rd_kafka_conf_set_dr_msg_cb or rd_kafka_conf_set_error_cb) you might chose not to call poll(), though this is not recommended.

Events:

Returns
the number of events served.
RD_EXPORT void rd_kafka_yield ( rd_kafka_t *  rk)

Cancels the current callback dispatcher (rd_kafka_poll(), rd_kafka_consume_callback(), etc).

A callback may use this to force an immediate return to the calling code (caller of e.g. rd_kafka_poll()) without processing any further events.

Remarks
This function MUST ONLY be called from within a librdkafka callback.
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.

Success or error is returned per-partition err in the partitions list.

Returns
RD_KAFKA_RESP_ERR_NO_ERROR
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.

Success or error is returned per-partition err in the partitions list.

Returns
RD_KAFKA_RESP_ERR_NO_ERROR
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.

Offsets are returned in *low and *high respectively.

Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or an error code on failure.
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.

The low offset is updated periodically (if statistics.interval.ms is set) while the high offset is updated on each fetched message set from the broker.

If there is no cached offset (either low or high, or both) then RD_KAFKA_OFFSET_INVALID will be returned for the respective offset.

Offsets are returned in *low and *high respectively.

Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or an error code on failure.
Remarks
Shall only be used with an active consumer instance.
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.

The returned offset for each partition is the earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition.

The timestamps to query are represented as offset in offsets on input, and offset will contain the offset on output.

The function will block for at most timeout_ms milliseconds.

Remarks
Duplicate Topic+Partitions are not supported.
Per-partition errors may be returned in rd_kafka_topic_partition_t.err
Returns
RD_KAFKA_RESP_ERR_NO_ERROR if offsets were be queried (do note that per-partition errors might be set), RD_KAFKA_RESP_ERR__TIMED_OUT if not all offsets could be fetched within timeout_ms, RD_KAFKA_RESP_ERR__INVALID_ARG if the offsets list is empty, RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION if all partitions are unknown, RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE if unable to query leaders for the given partitions.
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.

This is typically an abstraction for the calloc(3) call and makes sure the application can use the same memory allocator as librdkafka for allocating pointers that are used by librdkafka.

rk can be set to return memory allocated by a specific rk instance otherwise pass NULL for rk.

Remarks
Memory allocated by rd_kafka_mem_calloc() must be freed using rd_kafka_mem_free()
RD_EXPORT void* rd_kafka_mem_malloc ( rd_kafka_t *  rk,
size_t  size 
)

Allocate memory using the same allocator librdkafka uses.

This is typically an abstraction for the malloc(3) call and makes sure the application can use the same memory allocator as librdkafka for allocating pointers that are used by librdkafka.

rk can be set to return memory allocated by a specific rk instance otherwise pass NULL for rk.

Remarks
Memory allocated by rd_kafka_mem_malloc() must be freed using rd_kafka_mem_free()
RD_EXPORT void rd_kafka_mem_free ( rd_kafka_t *  rk,
void *  ptr 
)

Free pointer returned by librdkafka.

This is typically an abstraction for the free(3) call and makes sure the application can use the same memory allocator as librdkafka for freeing pointers returned by librdkafka.

In standard setups it is usually not necessary to use this interface rather than the free(3) functione.

rk must be set for memory returned by APIs that take an rk argument, for other APIs pass NULL for rk.

Remarks
rd_kafka_mem_free() must only be used for pointers returned by APIs that explicitly mention using this function for freeing.
RD_EXPORT rd_kafka_queue_t* rd_kafka_queue_new ( rd_kafka_t *  rk)

Create a new message queue.

See rd_kafka_consume_start_queue(), rd_kafka_consume_queue(), et.al.

RD_EXPORT void rd_kafka_queue_destroy ( rd_kafka_queue_t *  rkqu)

Destroy a queue, purging all of its enqueued messages.

RD_EXPORT rd_kafka_queue_t* rd_kafka_queue_get_main ( rd_kafka_t *  rk)
Returns
a reference to the main librdkafka event queue. This is the queue served by rd_kafka_poll().

Use rd_kafka_queue_destroy() to loose the reference.

RD_EXPORT rd_kafka_queue_t* rd_kafka_queue_get_sasl ( rd_kafka_t *  rk)
Returns
a reference to the SASL callback queue, if a SASL mechanism with callbacks is configured (currently only OAUTHBEARER), else returns NULL.

Use rd_kafka_queue_destroy() to loose the reference.

See also
rd_kafka_sasl_background_callbacks_enable()
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.

This serves as an alternative for applications that do not call rd_kafka_poll() (et.al.) at regular intervals (or not at all), as a means of automatically trigger the refresh callbacks, which are needed to initiate connections to the brokers in the case a custom OAUTHBEARER refresh callback is configured.

Returns
NULL on success or an error object on error.
See also
rd_kafka_queue_get_sasl()
rd_kafka_conf_set_oauthbearer_token_refresh_cb()
RD_EXPORT rd_kafka_queue_t* rd_kafka_queue_get_consumer ( rd_kafka_t *  rk)
Returns
a reference to the librdkafka consumer queue. This is the queue served by rd_kafka_consumer_poll().

Use rd_kafka_queue_destroy() to loose the reference.

Remarks
rd_kafka_queue_destroy() MUST be called on this queue prior to calling rd_kafka_consumer_close().
RD_EXPORT rd_kafka_queue_t* rd_kafka_queue_get_partition ( rd_kafka_t *  rk,
const char *  topic,
int32_t  partition 
)
Returns
a reference to the partition's queue, or NULL if partition is invalid.

Use rd_kafka_queue_destroy() to loose the reference.

Remarks
rd_kafka_queue_destroy() MUST be called on this queue
This function only works on consumers.
RD_EXPORT rd_kafka_queue_t* rd_kafka_queue_get_background ( rd_kafka_t *  rk)
Returns
a reference to the background thread queue, or NULL if the background queue is not enabled.

The background thread queue provides the application with an automatically polled queue that triggers the event callback in a background thread, this background thread is completely managed by librdkafka.

The background thread queue is automatically created if a generic event handler callback is configured with rd_kafka_conf_set_background_event_cb() or if rd_kafka_queue_get_background() is called.

The background queue is polled and served by librdkafka and MUST NOT be polled, forwarded, or otherwise managed by the application, it may only be used as the destination queue passed to queue-enabled APIs, such as the Admin API.

Use rd_kafka_queue_destroy() to loose the reference.

Warning
The background queue MUST NOT be read from (polled, consumed, etc), or forwarded from.
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.

The internal refcounts for both queues are increased.

Remarks
Regardless of whether dst is NULL or not, after calling this function, src will not forward it's fetch queue to the consumer queue.
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.

This allows an application to serve log callbacks (log_cb) in its thread of choice.

Parameters
rkClient instance.
rkquQueue to forward logs to. If the value is NULL the logs are forwarded to the main queue.
Remarks
The configuration property log.queue MUST also be set to true.
librdkafka maintains its own reference to the provided queue.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or an error code on error, eg RD_KAFKA_RESP_ERR__NOT_CONFIGURED when log.queue is not set to true.
RD_EXPORT size_t rd_kafka_queue_length ( rd_kafka_queue_t *  rkqu)
Returns
the current number of elements in queue.
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.

To ease integration with IO based polling loops this API allows an application to create a separate file-descriptor that librdkafka will write payload (of size size) to whenever a new element is enqueued on a previously empty queue.

To remove event triggering call with fd = -1.

librdkafka will maintain a copy of the payload.

Remarks
IO and callback event triggering are mutually exclusive.
When using forwarded queues the IO event must only be enabled on the final forwarded-to (destination) queue.
The file-descriptor/socket must be set to non-blocking.
RD_EXPORT void rd_kafka_queue_cb_event_enable ( rd_kafka_queue_t *  rkqu,
void(*)(rd_kafka_t *rk, void *qev_opaque)  event_cb,
void *  qev_opaque 
)

Enable callback event triggering for queue.

The callback will be called from an internal librdkafka thread when a new element is enqueued on a previously empty queue.

To remove event triggering call with event_cb = NULL.

The qev_opaque is passed to the callback's qev_opaque argument.

Remarks
IO and callback event triggering are mutually exclusive.
Since the callback may be triggered from internal librdkafka threads, the application must not perform any pro-longed work in the callback, or call any librdkafka APIs (for the same rd_kafka_t handle).
RD_EXPORT void rd_kafka_queue_yield ( rd_kafka_queue_t *  rkqu)

Cancels the current rd_kafka_queue_poll() on rkqu.

An application may use this from another thread to force an immediate return to the calling code (caller of rd_kafka_queue_poll()). Must not be used from signal handlers since that may cause deadlocks.

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:

  • RD_KAFKA_OFFSET_BEGINNING
  • RD_KAFKA_OFFSET_END
  • RD_KAFKA_OFFSET_STORED
  • RD_KAFKA_OFFSET_TAIL

rdkafka will attempt to keep queued.min.messages (config property) messages in the local queue by repeatedly fetching batches of messages from the broker until the threshold is reached.

The application shall use one of the rd_kafka_consume*() functions to consume messages from the local queue, each kafka message being represented as a rd_kafka_message_t * object.

rd_kafka_consume_start() must not be called multiple times for the same topic and partition without stopping consumption first with rd_kafka_consume_stop().

Returns
0 on success or -1 on error in which case errno is set accordingly:
  • EBUSY - Conflicts with an existing or previous subscription (RD_KAFKA_RESP_ERR__CONFLICT)
  • EINVAL - Invalid offset, or incomplete configuration (lacking group.id) (RD_KAFKA_RESP_ERR__INVALID_ARG)
  • ESRCH - requested partition is invalid. (RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION)
  • ENOENT - topic is unknown in the Kafka cluster. (RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC)

Use rd_kafka_errno2err() to convert sytem errno to rd_kafka_resp_err_t

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().

The application must use one of the rd_kafka_consume_*_queue() functions to receive fetched messages.

rd_kafka_consume_start_queue() must not be called multiple times for the same topic and partition without stopping consumption first with rd_kafka_consume_stop(). rd_kafka_consume_start() and rd_kafka_consume_start_queue() must not be combined for the same topic and partition.

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.

NOTE: To enforce synchronisation this call will block until the internal fetcher has terminated and offsets are committed to configured storage method.

The application needs to be stop all consumers before calling rd_kafka_destroy() on the main object handle.

Returns
0 on success or -1 on error (see errno).
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.

If timeout_ms is specified (not 0) the seek call will wait this long for the consumer to update its fetcher state for the given partition with the new offset. This guarantees that no previously fetched messages for the old offset (or fetch position) will be passed to the application.

If the timeout is reached the internal state will be unknown to the caller and this function returns RD_KAFKA_RESP_ERR__TIMED_OUT.

If timeout_ms is 0 it will initiate the seek but return immediately without any error reporting (e.g., async).

This call will purge all pre-fetched messages for the given partition, which may be up to queued.max.message.kbytes in size. Repeated use of seek may thus lead to increased network usage as messages are re-fetched from the broker.

Remarks
Seek must only be performed for already assigned/consumed partitions, use rd_kafka_assign() (et.al) to set the initial starting offset for a new assignmenmt.
Returns
RD_KAFKA_RESP_ERR__NO_ERROR on success else an error code.
Deprecated:
Use rd_kafka_seek_partitions().
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.

The offset may be either absolute (>= 0) or a logical offset.

If timeout_ms is specified (not 0) the seek call will wait this long for the consumer to update its fetcher state for the given partition with the new offset. This guarantees that no previously fetched messages for the old offset (or fetch position) will be passed to the application.

If the timeout is reached the internal state will be unknown to the caller and this function returns RD_KAFKA_RESP_ERR__TIMED_OUT.

If timeout_ms is 0 it will initiate the seek but return immediately without any error reporting (e.g., async).

This call will purge all pre-fetched messages for the given partition, which may be up to queued.max.message.kbytes in size. Repeated use of seek may thus lead to increased network usage as messages are re-fetched from the broker.

Individual partition errors are reported in the per-partition .err field of partitions.

Remarks
Seek must only be performed for already assigned/consumed partitions, use rd_kafka_assign() (et.al) to set the initial starting offset for a new assignmenmt.
Returns
NULL on success or an error object on failure.
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.

timeout_ms is maximum amount of time to wait for a message to be received. Consumer must have been previously started with rd_kafka_consume_start().

Returns
a message object on success or NULL on error. The message object must be destroyed with rd_kafka_message_destroy() when the application is done with it.

Errors (when returning NULL):

  • ETIMEDOUT - timeout_ms was reached with no new messages fetched.
  • ENOENT - rkt + partition is unknown. (no prior rd_kafka_consume_start() call)

NOTE: The returned message's ..->err must be checked for errors. NOTE: ..->err == RD_KAFKA_RESP_ERR__PARTITION_EOF signals that the end of the partition has been reached, which should typically not be considered an error. The application should handle this case (e.g., ignore).

Remarks
on_consume() interceptors may be called from this function prior to passing message to application.
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).

rd_kafka_consume_batch() provides higher throughput performance than rd_kafka_consume().

timeout_ms is the maximum amount of time to wait for all of rkmessages_size messages to be put into rkmessages. If no messages were available within the timeout period this function returns 0 and rkmessages remains untouched. This differs somewhat from rd_kafka_consume().

The message objects must be destroyed with rd_kafka_message_destroy() when the application is done with it.

Returns
the number of rkmessages added in rkmessages, or -1 on error (same error codes as for rd_kafka_consume().
See also
rd_kafka_consume()
Remarks
on_consume() interceptors may be called from this function prior to passing message to application.
RD_EXPORT int rd_kafka_consume_callback ( rd_kafka_topic_t *  rkt,
int32_t  partition,
int  timeout_ms,
void(*)(rd_kafka_message_t *rkmessage, void *commit_opaque)  consume_cb,
void *  commit_opaque 
)

Consumes messages from topic rkt and partition, calling the provided callback for each consumed messsage.

rd_kafka_consume_callback() provides higher throughput performance than both rd_kafka_consume() and rd_kafka_consume_batch().

timeout_ms is the maximum amount of time to wait for one or more messages to arrive.

The provided consume_cb function is called for each message, the application MUST NOT call rd_kafka_message_destroy() on the provided rkmessage.

The commit_opaque argument is passed to the consume_cb as commit_opaque.

Returns
the number of messages processed or -1 on error.
See also
rd_kafka_consume()
Remarks
on_consume() interceptors may be called from this function prior to passing message to application.
This function will return early if a transaction control message is received, these messages are not exposed to the application but still enqueued on the consumer queue to make sure their offsets are stored.
Deprecated:
This API is deprecated and subject for future removal. There is no new callback-based consume interface, use the poll/queue based alternatives.
RD_EXPORT rd_kafka_message_t* rd_kafka_consume_queue ( rd_kafka_queue_t *  rkqu,
int  timeout_ms 
)

Consume from queue.

See also
rd_kafka_consume()
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.

See also
rd_kafka_consume_batch()
RD_EXPORT int rd_kafka_consume_callback_queue ( rd_kafka_queue_t *  rkqu,
int  timeout_ms,
void(*)(rd_kafka_message_t *rkmessage, void *commit_opaque)  consume_cb,
void *  commit_opaque 
)

Consume multiple messages from queue with callback.

See also
rd_kafka_consume_callback()
Deprecated:
This API is deprecated and subject for future removal. There is no new callback-based consume interface, use the poll/queue based alternatives.
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.

The offset + 1 will be committed (written) to broker (or file) according to auto.commit.interval.ms or manual offset-less commit()

Warning
This method may only be called for partitions that are currently assigned. Non-assigned partitions will fail with RD_KAFKA_RESP_ERR__STATE. Since v1.9.0.
Avoid storing offsets after calling rd_kafka_seek() (et.al) as this may later interfere with resuming a paused partition, instead store offsets prior to calling seek.
Remarks
enable.auto.offset.store must be set to "false" when using this API.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or an error code on error.
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.

The offset will be committed (written) to the offset store according to auto.commit.interval.ms or manual offset-less commit().

Per-partition success/error status propagated through each partition's .err for all return values (even NO_ERROR) except INVALID_ARG.

Warning
This method may only be called for partitions that are currently assigned. Non-assigned partitions will fail with RD_KAFKA_RESP_ERR__STATE. Since v1.9.0.
Avoid storing offsets after calling rd_kafka_seek() (et.al) as this may later interfere with resuming a paused partition, instead store offsets prior to calling seek.
Remarks
The .offset field is stored as is, it will NOT be + 1.
enable.auto.offset.store must be set to "false" when using this API.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on (partial) success, or RD_KAFKA_RESP_ERR__INVALID_ARG if enable.auto.offset.store is true, or RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION or RD_KAFKA_RESP_ERR__STATE if none of the offsets could be stored.
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.

Wildcard (regex) topics are supported: any topic name in the topics list that is prefixed with "^" will be regex-matched to the full list of topics in the cluster and matching topics will be added to the subscription list.

The full topic list is retrieved every topic.metadata.refresh.interval.ms to pick up new or delete topics that match the subscription. If there is any change to the matched topics the consumer will immediately rejoin the group with the updated set of subscribed topics.

Regex and full topic names can be mixed in topics.

Remarks
Only the .topic field is used in the supplied topics list, all other fields are ignored.
subscribe() is an asynchronous method which returns immediately: background threads will (re)join the group, wait for group rebalance, issue any registered rebalance_cb, assign() the assigned partitions, and then start fetching messages. This cycle may take up to session.timeout.ms * 2 or more to complete.
After this call returns a consumer error will be returned by rd_kafka_consumer_poll (et.al) for each unavailable topic in the topics. The error will be RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART for non-existent topics, and RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED for unauthorized topics. The consumer error will be raised through rd_kafka_consumer_poll() (et.al.) with the rd_kafka_message_t.err field set to one of the error codes mentioned above. The subscribe function itself is asynchronous and will not return an error on unavailable topics.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or RD_KAFKA_RESP_ERR__INVALID_ARG if list is empty, contains invalid topics or regexes or duplicate entries, RD_KAFKA_RESP_ERR__FATAL if the consumer has raised a fatal error.
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.

Returns
An error code on failure, otherwise topic is updated to point to a newly allocated topic list (possibly empty).
Remarks
The application is responsible for calling rd_kafka_topic_partition_list_destroy on the returned list.
RD_EXPORT rd_kafka_message_t* rd_kafka_consumer_poll ( rd_kafka_t *  rk,
int  timeout_ms 
)

Poll the consumer for messages or events.

Will block for at most timeout_ms milliseconds.

Remarks
An application should make sure to call consumer_poll() at regular intervals, even if no messages are expected, to serve any queued callbacks waiting to be called. This is especially important when a rebalance_cb has been registered as it needs to be called and handled properly to synchronize internal consumer state.
Returns
A message object which is a proper message if ->err is RD_KAFKA_RESP_ERR_NO_ERROR, or an event or error for any other value.
Remarks
on_consume() interceptors may be called from this function prior to passing message to application.
When subscribing to topics the application must call poll at least every max.poll.interval.ms to remain a member of the consumer group.

Noteworthy errors returned in ->err:

  • RD_KAFKA_RESP_ERR__MAX_POLL_EXCEEDED - application failed to call poll within max.poll.interval.ms.
See also
rd_kafka_message_t
RD_EXPORT rd_kafka_resp_err_t rd_kafka_consumer_close ( rd_kafka_t *  rk)

Close the consumer.

This call will block until the consumer has revoked its assignment, calling the rebalance_cb if it is configured, committed offsets to broker, and left the consumer group (if applicable). The maximum blocking time is roughly limited to session.timeout.ms.

Returns
An error code indicating if the consumer close was succesful or not. RD_KAFKA_RESP_ERR__FATAL is returned if the consumer has raised a fatal error.
Remarks
The application still needs to call rd_kafka_destroy() after this call finishes to clean up the underlying handle resources.
RD_EXPORT rd_kafka_error_t* rd_kafka_consumer_close_queue ( rd_kafka_t *  rk,
rd_kafka_queue_t *  rkqu 
)

Asynchronously close the consumer.

Performs the same actions as rd_kafka_consumer_close() but in a background thread.

Rebalance events/callbacks (etc) will be forwarded to the application-provided rkqu. The application must poll/serve this queue until rd_kafka_consumer_closed() returns true.

Remarks
Depending on consumer group join state there may or may not be rebalance events emitted on rkqu.
Returns
an error object if the consumer close failed, else NULL.
See also
rd_kafka_consumer_closed()
RD_EXPORT int rd_kafka_consumer_closed ( rd_kafka_t *  rk)
Returns
1 if the consumer is closed, else 0.

Should be used in conjunction with rd_kafka_consumer_close_queue() to know when the consumer has been closed.

See also
rd_kafka_consumer_close_queue()
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.

If a COOPERATIVE assignor (i.e. incremental rebalancing) is being used, this method should be used in a rebalance callback to adjust the current assignment appropriately in the case where the rebalance type is RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS. The application must pass the partition list passed to the callback (or a copy of it), even if the list is empty. partitions must not be NULL. This method may also be used outside the context of a rebalance callback.

Returns
NULL on success, or an error object if the operation was unsuccessful.
Remarks
The returned error object (if not NULL) must be destroyed with rd_kafka_error_destroy().
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.

If a COOPERATIVE assignor (i.e. incremental rebalancing) is being used, this method should be used in a rebalance callback to adjust the current assignment appropriately in the case where the rebalance type is RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS. The application must pass the partition list passed to the callback (or a copy of it), even if the list is empty. partitions must not be NULL. This method may also be used outside the context of a rebalance callback.

Returns
NULL on success, or an error object if the operation was unsuccessful.
Remarks
The returned error object (if not NULL) must be destroyed with rd_kafka_error_destroy().
RD_EXPORT const 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.

Returns
NULL on error, or one of "NONE", "EAGER", "COOPERATIVE" on success.
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.

The new partitions will replace the existing assignment.

A zero-length partitions will treat the partitions as a valid, albeit empty assignment, and maintain internal state, while a NULL value for partitions will reset and clear the internal state.

When used from a rebalance callback, the application should pass the partition list passed to the callback (or a copy of it) even if the list is empty (i.e. should not pass NULL in this case) so as to maintain internal join state. This is not strictly required - the application may adjust the assignment provided by the group. However, this is rarely useful in practice.

Returns
An error code indicating if the new assignment was applied or not. RD_KAFKA_RESP_ERR__FATAL is returned if the consumer has raised a fatal error.
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().

Returns
An error code on failure, otherwise partitions is updated to point to a newly allocated partition list (possibly empty).
Remarks
The application is responsible for calling rd_kafka_topic_partition_list_destroy on the returned list.
This assignment represents the partitions assigned through the assign functions and not the partitions assigned to this consumer instance by the consumer group leader. They are usually the same following a rebalance but not necessarily since an application is free to assign any partitions.
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.

Remarks
Calling rd_kafka_assign(), rd_kafka_incremental_assign() or rd_kafka_incremental_unassign() resets this flag.
Returns
Returns 1 if the current partition assignment is considered lost, 0 otherwise.
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.

offsets should contain topic, partition, offset and possibly metadata. The offset should be the offset where consumption will resume, i.e., the last processed offset + 1. If offsets is NULL the current partition assignment will be used instead.

If async is false this operation will block until the broker offset commit is done, returning the resulting success or error code.

If a rd_kafka_conf_set_offset_commit_cb() offset commit callback has been configured the callback will be enqueued for a future call to rd_kafka_poll(), rd_kafka_consumer_poll() or similar.

Returns
An error code indiciating if the commit was successful, or successfully scheduled if asynchronous, or failed. RD_KAFKA_RESP_ERR__FATAL is returned if the consumer has raised a fatal error.
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.

See also
rd_kafka_commit
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(*)(rd_kafka_t *rk, rd_kafka_resp_err_t err, rd_kafka_topic_partition_list_t *offsets, void *commit_opaque)  cb,
void *  commit_opaque 
)

Commit offsets on broker for the provided list of partitions.

See rd_kafka_commit for offsets semantics.

The result of the offset commit will be posted on the provided rkqu queue.

If the application uses one of the poll APIs (rd_kafka_poll(), rd_kafka_consumer_poll(), rd_kafka_queue_poll(), ..) to serve the queue the cb callback is required.

The commit_opaque argument is passed to the callback as commit_opaque, or if using the event API the callback is ignored and the offset commit result will be returned as an RD_KAFKA_EVENT_COMMIT event and the commit_opaque value will be available with rd_kafka_event_opaque().

If rkqu is NULL a temporary queue will be created and the callback will be served by this call.

See also
rd_kafka_commit()
rd_kafka_conf_set_offset_commit_cb()
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.

The offset field of each requested partition will either be set to stored offset or to RD_KAFKA_OFFSET_INVALID in case there was no stored offset for that partition.

Committed offsets will be returned according to the isolation.level configuration property, if set to read_committed (default) then only stable offsets for fully committed transactions will be returned, while read_uncommitted may return offsets for not yet committed transactions.

Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success in which case the offset or err field of each partitions' element is filled in with the stored offset, or a partition specific error. Else returns an error code.
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.

The offset field of each requested partition will be set to the offset of the last consumed message + 1, or RD_KAFKA_OFFSET_INVALID in case there was no previous message.

Remarks
In this context the last consumed message is the offset consumed by the current librdkafka instance and, in case of rebalancing, not necessarily the last message fetched from the partition.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success in which case the offset or err field of each partitions' element is filled in with the stored offset, or a partition specific error. Else returns an error code.
RD_EXPORT rd_kafka_consumer_group_metadata_t* rd_kafka_consumer_group_metadata ( rd_kafka_t *  rk)
Returns
the current consumer group metadata associated with this consumer, or NULL if rk is not a consumer configured with a group.id. This metadata object should be passed to the transactional producer's rd_kafka_send_offsets_to_transaction() API.
Remarks
The returned pointer must be freed by the application using rd_kafka_consumer_group_metadata_destroy().
See also
rd_kafka_send_offsets_to_transaction()
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.

Parameters
group_idThe group id.
Remarks
The returned pointer must be freed by the application using rd_kafka_consumer_group_metadata_destroy().
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.

Parameters
group_idThe group id.
generation_idThe group generation id.
member_idThe group member id.
group_instance_idThe group instance id (may be NULL).
Remarks
The returned pointer must be freed by the application using rd_kafka_consumer_group_metadata_destroy().
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.

Remarks
The serialized metadata format is private and is not compatible across different versions or even builds of librdkafka. It should only be used in the same process runtime and must only be passed to rd_kafka_consumer_group_metadata_read().
Parameters
cgmdMetadata to be serialized.
bufferpOn success this pointer will be updated to point to na allocated buffer containing the serialized metadata. The buffer must be freed with rd_kafka_mem_free().
sizepThe pointed to size will be updated with the size of the serialized buffer.
Returns
NULL on success or an error object on failure.
See also
rd_kafka_consumer_group_metadata_read()
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.

Remarks
The serialized metadata format is private and is not compatible across different versions or even builds of librdkafka. It should only be used in the same process runtime and must only be passed to rd_kafka_consumer_group_metadata_read().
Parameters
cgmdpOn success this pointer will be updated to point to a new consumer group metadata object which must be freed with rd_kafka_consumer_group_metadata_destroy().
bufferPointer to the serialized data.
sizeSize of the serialized data.
Returns
NULL on success or an error object on failure.
See also
rd_kafka_consumer_group_metadata_write()
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.

rkt is the target topic which must have been previously created with rd_kafka_topic_new().

rd_kafka_produce() is an asynch non-blocking API. See rd_kafka_conf_set_dr_msg_cb on how to setup a callback to be called once the delivery status (success or failure) is known. The delivery report is trigged by the application calling rd_kafka_poll() (at regular intervals) or rd_kafka_flush() (at termination).

Since producing is asynchronous, you should call rd_kafka_flush() before you destroy the producer. Otherwise, any outstanding messages will be silently discarded.

When temporary errors occur, librdkafka automatically retries to produce the messages. Retries are triggered after retry.backoff.ms and when the leader broker for the given partition is available. Otherwise, librdkafka falls back to polling the topic metadata to monitor when a new leader is elected (see the topic.metadata.refresh.fast.interval.ms and topic.metadata.refresh.interval.ms configurations) and then performs a retry. A delivery error will occur if the message could not be produced within message.timeout.ms.

See the "Message reliability" chapter in INTRODUCTION.md for more information.

partition is the target partition, either:

  • RD_KAFKA_PARTITION_UA (unassigned) for automatic partitioning using the topic's partitioner function, or
  • a fixed partition (0..N)

msgflags is zero or more of the following flags OR:ed together: RD_KAFKA_MSG_F_BLOCK - block produce*() call if queue.buffering.max.messages or queue.buffering.max.kbytes are exceeded. Messages are considered in-queue from the point they are accepted by produce() until their corresponding delivery report callback/event returns. It is thus a requirement to call rd_kafka_poll() (or equiv.) from a separate thread when F_BLOCK is used. See WARNING on RD_KAFKA_MSG_F_BLOCK above.

RD_KAFKA_MSG_F_FREE - rdkafka will free(3) payload when it is done with it. RD_KAFKA_MSG_F_COPY - the payload data will be copied and the payload pointer will not be used by rdkafka after the call returns. RD_KAFKA_MSG_F_PARTITION - produce_batch() will honour per-message partition, either set manually or by the configured partitioner.

.._F_FREE and .._F_COPY are mutually exclusive. If neither of these are set, the caller must ensure that the memory backing payload remains valid and is not modified or reused until the delivery callback is invoked. Other buffers passed to rd_kafka_produce() don't have this restriction on reuse, i.e. the memory backing the key or the topic name may be reused as soon as rd_kafka_produce() returns.

If the function returns -1 and RD_KAFKA_MSG_F_FREE was specified, then the memory associated with the payload is still the caller's responsibility.

payload is the message payload of size len bytes.

key is an optional message key of size keylen bytes, if non-NULL it will be passed to the topic partitioner as well as be sent with the message to the broker and passed on to the consumer.

msg_opaque is an optional application-provided per-message opaque pointer that will provided in the message's delivery report callback (dr_msg_cb or dr_cb) and the rd_kafka_message_t _private field.

Remarks
on_send() and on_acknowledgement() interceptors may be called from this function. on_acknowledgement() will only be called if the message fails partitioning.
If the producer is transactional (transactional.id is configured) producing is only allowed during an on-going transaction, namely after rd_kafka_begin_transaction() has been called.
Returns
0 on success or -1 on error in which case errno is set accordingly:
  • ENOBUFS - maximum number of outstanding messages has been reached: "queue.buffering.max.messages" (RD_KAFKA_RESP_ERR__QUEUE_FULL)
  • EMSGSIZE - message is larger than configured max size: "messages.max.bytes". (RD_KAFKA_RESP_ERR_MSG_SIZE_TOO_LARGE)
  • ESRCH - requested partition is unknown in the Kafka cluster. (RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION)
  • ENOENT - topic is unknown in the Kafka cluster. (RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC)
  • ECANCELED - fatal error has been raised on producer, see rd_kafka_fatal_error(), (RD_KAFKA_RESP_ERR__FATAL).
  • ENOEXEC - transactional state forbids producing (RD_KAFKA_RESP_ERR__STATE)
See also
Use rd_kafka_errno2err() to convert errno to rdkafka error code.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_producev ( rd_kafka_t *  rk,
  ... 
)

Produce and send a single message to broker.

The message is defined by a va-arg list using rd_kafka_vtype_t tag tuples which must be terminated with a single RD_KAFKA_V_END.

Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success, else an error code as described in rd_kafka_produce(). RD_KAFKA_RESP_ERR__CONFLICT is returned if _V_HEADER and _V_HEADERS are mixed.
See also
rd_kafka_produce, rd_kafka_produceva, RD_KAFKA_V_END
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.

The message is defined by an array of rd_kafka_vu_t of count cnt.

Returns
an error object on failure or NULL on success. See rd_kafka_producev() for specific error codes.
See also
rd_kafka_produce, rd_kafka_producev, RD_KAFKA_V_END
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.

If partition is RD_KAFKA_PARTITION_UA the configured partitioner will be run for each message (slower), otherwise the messages will be enqueued to the specified partition directly (faster).

The messages are provided in the array rkmessages of count message_cnt elements. The partition and msgflags are used for all provided messages.

Honoured rkmessages[] fields are:

  • payload,len Message payload and length
  • key,key_len Optional message key
  • _private Message opaque pointer (msg_opaque)
  • err Will be set according to success or failure, see rd_kafka_produce() for possible error codes. Application only needs to check for errors if return value != message_cnt.
Remarks
If RD_KAFKA_MSG_F_PARTITION is set in msgflags, the .partition field of the rkmessages is used instead of partition.
Returns
the number of messages succesfully enqueued for producing.
Remarks
This interface does NOT support setting message headers on the provided rkmessages.
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.

Remarks
This function will call rd_kafka_poll() and thus trigger callbacks.
The linger.ms time will be ignored for the duration of the call, queued messages will be sent to the broker as soon as possible.
If RD_KAFKA_EVENT_DR has been enabled (through rd_kafka_conf_set_events()) this function will not call rd_kafka_poll() but instead wait for the librdkafka-handled message count to reach zero. This requires the application to serve the event queue in a separate thread. In this mode only messages are counted, not other types of queued events.
Returns
RD_KAFKA_RESP_ERR__TIMED_OUT if timeout_ms was reached before all outstanding requests were completed, else RD_KAFKA_RESP_ERR_NO_ERROR
See also
rd_kafka_outq_len()
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.

Parameters
rkClient instance.
purge_flagsTells which messages to purge and how.

The application will need to call rd_kafka_poll() or rd_kafka_flush() afterwards to serve the delivery report callbacks of the purged messages.

Messages purged from internal queues fail with the delivery report error code set to RD_KAFKA_RESP_ERR__PURGE_QUEUE, while purged messages that are in-flight to or from the broker will fail with the error code set to RD_KAFKA_RESP_ERR__PURGE_INFLIGHT.

Warning
Purging messages that are in-flight to or from the broker will ignore any sub-sequent acknowledgement for these messages received from the broker, effectively making it impossible for the application to know if the messages were successfully produced or not. This may result in duplicate messages if the application retries these messages at a later time.
Remarks
This call may block for a short time while background thread queues are purged.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success, RD_KAFKA_RESP_ERR__INVALID_ARG if the purge flags are invalid or unknown, RD_KAFKA_RESP_ERR__NOT_IMPLEMENTED if called on a non-producer client instance.
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.

Parameters:

  • all_topics if non-zero: request info about all topics in cluster, if zero: only request info about locally known topics.
  • only_rkt only request info about this topic
  • metadatap pointer to hold metadata result. The *metadatap pointer must be released with rd_kafka_metadata_destroy().
  • timeout_ms maximum response time before failing.
Remarks
Consumer: If all_topics is non-zero the Metadata response information may trigger a re-join if any subscribed topics have changed partition count or existence state.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success (in which case *metadatap) will be set, else RD_KAFKA_RESP_ERR__TIMED_OUT on timeout or other error code on error.
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.

group is an optional group name to describe, otherwise (NULL) all groups are returned.

timeout_ms is the (approximate) maximum time to wait for response from brokers and must be a positive value.

Returns
RD_KAFKA_RESP_ERR__NO_ERROR on success and grplistp is updated to point to a newly allocated list of groups. RD_KAFKA_RESP_ERR__PARTIAL if not all brokers responded in time but at least one group is returned in grplistlp. RD_KAFKA_RESP_ERR__TIMED_OUT if no groups were returned in the given timeframe but not all brokers have yet responded, or if the list of brokers in the cluster could not be obtained within the given timeframe. RD_KAFKA_RESP_ERR__TRANSPORT if no brokers were found. Other error codes may also be returned from the request layer.

The grplistp remains untouched if any error code is returned, with the exception of RD_KAFKA_RESP_ERR__PARTIAL which behaves as RD_KAFKA_RESP_ERR__NO_ERROR (success) but with an incomplete group list.

See also
Use rd_kafka_group_list_destroy() to release list memory.
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.

Additional brokers will be discovered automatically as soon as rdkafka connects to a broker by querying the broker metadata.

If a broker name resolves to multiple addresses (and possibly address families) all will be used for connection attempts in round-robin fashion.

brokerlist is a ,-separated list of brokers in the format: <broker1>,<broker2>,.. Where each broker is in either the host or URL based format: <host>[:<port>] <proto>://<host>[:port] <proto> is either PLAINTEXT, SSL, SASL, SASL_PLAINTEXT The two formats can be mixed but ultimately the value of the security.protocol config property decides what brokers are allowed.

Example: brokerlist = "broker1:10000,broker2" brokerlist = "SSL://broker3:9000,ssl://broker2"

Returns
the number of brokers successfully added.
Remarks
Brokers may also be defined with the metadata.broker.list or bootstrap.servers configuration property (preferred method).
Deprecated:
Set bootstrap servers with the bootstrap.servers configuration property.
RD_EXPORT RD_DEPRECATED void rd_kafka_set_logger ( rd_kafka_t *  rk,
void(*)(const rd_kafka_t *rk, int level, const char *fac, const char *buf)  func 
)

Set logger function.

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

Deprecated:
Use rd_kafka_conf_set_log_cb()
Remarks
rk may be passed as NULL in the callback.
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.

Deprecated:
Set the "log_level" configuration property instead.
Remarks
If the "debug" configuration property is set the log level is automatically adjusted to LOG_DEBUG (7).
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.

Remarks
This logger is only available if librdkafka was built with syslog support.
RD_EXPORT int rd_kafka_outq_len ( rd_kafka_t *  rk)

Returns the current out queue length.

The out queue length is the sum of:

  • number of messages waiting to be sent to, or acknowledged by, the broker.
  • number of delivery reports (e.g., dr_msg_cb) waiting to be served by rd_kafka_poll() or rd_kafka_flush().
  • number of callbacks (e.g., error_cb, stats_cb, etc) waiting to be served by rd_kafka_poll(), rd_kafka_consumer_poll() or rd_kafka_flush().
  • number of events waiting to be served by background_event_cb() in the background queue (see rd_kafka_conf_set_background_event_cb).

An application should wait for the return value of this function to reach zero before terminating to make sure outstanding messages, requests (such as offset commits), callbacks and events are fully processed. See rd_kafka_flush().

Returns
number of messages and events waiting in queues.
See also
rd_kafka_flush()
RD_EXPORT void rd_kafka_dump ( FILE *  fp,
rd_kafka_t *  rk 
)

Dumps rdkafka's internal state for handle rk to stream fp.

This is only useful for debugging rdkafka, showing state and statistics for brokers, topics, partitions, etc.

RD_EXPORT int rd_kafka_thread_cnt ( void  )

Retrieve the current number of threads in use by librdkafka.

Used by regression tests.

RD_EXPORT int rd_kafka_wait_destroyed ( int  timeout_ms)

Wait for all rd_kafka_t objects to be destroyed.

Returns 0 if all kafka objects are now destroyed, or -1 if the timeout was reached.

Remarks
This function is deprecated.
RD_EXPORT int rd_kafka_unittest ( void  )

Run librdkafka's built-in unit-tests.

Returns
the number of failures, or 0 if all tests passed.
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()).

Warning
It is not permitted to call rd_kafka_poll() after directing the main queue with rd_kafka_poll_set_consumer().
RD_EXPORT rd_kafka_event_type_t rd_kafka_event_type ( const rd_kafka_event_t *  rkev)
Returns
the event type for the given event.
Remarks
As a convenience it is okay to pass rkev as NULL in which case RD_KAFKA_EVENT_NONE is returned.
RD_EXPORT const char* rd_kafka_event_name ( const rd_kafka_event_t *  rkev)
Returns
the event type's name for the given event.
Remarks
As a convenience it is okay to pass rkev as NULL in which case the name for RD_KAFKA_EVENT_NONE is returned.
RD_EXPORT void rd_kafka_event_destroy ( rd_kafka_event_t *  rkev)

Destroy an event.

Remarks
Any references to this event, such as extracted messages, will not be usable after this call.
As a convenience it is okay to pass rkev as NULL in which case no action is performed.
RD_EXPORT const rd_kafka_message_t* rd_kafka_event_message_next ( rd_kafka_event_t *  rkev)
Returns
the next message from an event.

Call repeatedly until it returns NULL.

Event types:

  • RD_KAFKA_EVENT_FETCH (1 message)
  • RD_KAFKA_EVENT_DR (>=1 message(s))
Remarks
The returned message(s) MUST NOT be freed with rd_kafka_message_destroy().
on_consume() interceptor may be called from this function prior to passing message to application.
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.

Event types:

  • RD_KAFKA_EVENT_FETCH (1 message)
  • RD_KAFKA_EVENT_DR (>=1 message(s))
Returns
the number of messages extracted.
Remarks
on_consume() interceptor may be called from this function prior to passing message to application.
RD_EXPORT size_t rd_kafka_event_message_count ( rd_kafka_event_t *  rkev)
Returns
the number of remaining messages in the event.

Event types:

  • RD_KAFKA_EVENT_FETCH (1 message)
  • RD_KAFKA_EVENT_DR (>=1 message(s))
RD_EXPORT const char* rd_kafka_event_config_string ( rd_kafka_event_t *  rkev)
Returns
the associated configuration string for the event, or NULL if the configuration property is not set or if not applicable for the given event type.

The returned memory is read-only and its lifetime is the same as the event object.

Event types:

  • RD_KAFKA_EVENT_OAUTHBEARER_TOKEN_REFRESH: value of sasl.oauthbearer.config
RD_EXPORT rd_kafka_resp_err_t rd_kafka_event_error ( rd_kafka_event_t *  rkev)
Returns
the error code for the event.

Use rd_kafka_event_error_is_fatal() to detect if this is a fatal error.

Event types:

  • all
RD_EXPORT const char* rd_kafka_event_error_string ( rd_kafka_event_t *  rkev)
Returns
the error string (if any). An application should check that rd_kafka_event_error() returns non-zero before calling this function.

Event types:

  • all
RD_EXPORT int rd_kafka_event_error_is_fatal ( rd_kafka_event_t *  rkev)
Returns
1 if the error is a fatal error, else 0.

Event types:

  • RD_KAFKA_EVENT_ERROR
See also
rd_kafka_fatal_error()
RD_EXPORT void* rd_kafka_event_opaque ( rd_kafka_event_t *  rkev)
Returns
the event opaque (if any) as passed to rd_kafka_commit() (et.al) or rd_kafka_AdminOptions_set_opaque(), depending on event type.

Event types:

  • RD_KAFKA_EVENT_OFFSET_COMMIT
  • RD_KAFKA_EVENT_CREATETOPICS_RESULT
  • RD_KAFKA_EVENT_DELETETOPICS_RESULT
  • RD_KAFKA_EVENT_CREATEPARTITIONS_RESULT
  • RD_KAFKA_EVENT_CREATEACLS_RESULT
  • RD_KAFKA_EVENT_DESCRIBEACLS_RESULT
  • RD_KAFKA_EVENT_DELETEACLS_RESULT
  • RD_KAFKA_EVENT_ALTERCONFIGS_RESULT
  • RD_KAFKA_EVENT_DESCRIBECONFIGS_RESULT
  • RD_KAFKA_EVENT_DELETEGROUPS_RESULT
  • RD_KAFKA_EVENT_DELETECONSUMERGROUPOFFSETS_RESULT
  • RD_KAFKA_EVENT_DELETERECORDS_RESULT
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.

Event types:

  • RD_KAFKA_EVENT_LOG
Returns
0 on success or -1 if unsupported event type.
RD_EXPORT int rd_kafka_event_debug_contexts ( rd_kafka_event_t *  rkev,
char *  dst,
size_t  dstsize 
)

Extract log debug context from event.

Event types:

  • RD_KAFKA_EVENT_LOG
Parameters
rkevthe event to extract data from.
dstdestination string for comma separated list.
dstsizesize of provided dst buffer.
Returns
0 on success or -1 if unsupported event type.
RD_EXPORT const char* rd_kafka_event_stats ( rd_kafka_event_t *  rkev)

Extract stats from the event.

Event types:

  • RD_KAFKA_EVENT_STATS
Returns
stats json string.
Remarks
the returned string will be freed automatically along with the event object
RD_EXPORT rd_kafka_topic_partition_list_t* rd_kafka_event_topic_partition_list ( rd_kafka_event_t *  rkev)
Returns
the topic partition list from the event.
Remarks
The list MUST NOT be freed with rd_kafka_topic_partition_list_destroy()

Event types:

  • RD_KAFKA_EVENT_REBALANCE
  • RD_KAFKA_EVENT_OFFSET_COMMIT
RD_EXPORT rd_kafka_topic_partition_t* rd_kafka_event_topic_partition ( rd_kafka_event_t *  rkev)
Returns
a newly allocated topic_partition container, if applicable for the event type, else NULL.
Remarks
The returned pointer MUST be freed with rd_kafka_topic_partition_destroy().

Event types: RD_KAFKA_EVENT_ERROR (for partition level errors)

RD_EXPORT const rd_kafka_CreateTopics_result_t* rd_kafka_event_CreateTopics_result ( rd_kafka_event_t *  rkev)

Get CreateTopics result.

Returns
the result of a CreateTopics request, or NULL if event is of different type.

Event types: RD_KAFKA_EVENT_CREATETOPICS_RESULT

RD_EXPORT const rd_kafka_DeleteTopics_result_t* rd_kafka_event_DeleteTopics_result ( rd_kafka_event_t *  rkev)

Get DeleteTopics result.

Returns
the result of a DeleteTopics request, or NULL if event is of different type.

Event types: RD_KAFKA_EVENT_DELETETOPICS_RESULT

RD_EXPORT const rd_kafka_CreatePartitions_result_t* rd_kafka_event_CreatePartitions_result ( rd_kafka_event_t *  rkev)

Get CreatePartitions result.

Returns
the result of a CreatePartitions request, or NULL if event is of different type.

Event types: RD_KAFKA_EVENT_CREATEPARTITIONS_RESULT

RD_EXPORT const rd_kafka_AlterConfigs_result_t* rd_kafka_event_AlterConfigs_result ( rd_kafka_event_t *  rkev)

Get AlterConfigs result.

Returns
the result of a AlterConfigs request, or NULL if event is of different type.

Event types: RD_KAFKA_EVENT_ALTERCONFIGS_RESULT

RD_EXPORT const rd_kafka_DescribeConfigs_result_t* rd_kafka_event_DescribeConfigs_result ( rd_kafka_event_t *  rkev)

Get DescribeConfigs result.

Returns
the result of a DescribeConfigs request, or NULL if event is of different type.

Event types: RD_KAFKA_EVENT_DESCRIBECONFIGS_RESULT

RD_EXPORT const rd_kafka_DeleteRecords_result_t* rd_kafka_event_DeleteRecords_result ( rd_kafka_event_t *  rkev)
Returns
the result of a DeleteRecords request, or NULL if event is of different type.

Event types: RD_KAFKA_EVENT_DELETERECORDS_RESULT

RD_EXPORT const rd_kafka_DeleteGroups_result_t* rd_kafka_event_DeleteGroups_result ( rd_kafka_event_t *  rkev)

Get DeleteGroups result.

Returns
the result of a DeleteGroups request, or NULL if event is of different type.

Event types: RD_KAFKA_EVENT_DELETEGROUPS_RESULT

RD_EXPORT const rd_kafka_DeleteConsumerGroupOffsets_result_t* rd_kafka_event_DeleteConsumerGroupOffsets_result ( rd_kafka_event_t *  rkev)

Get DeleteConsumerGroupOffsets result.

Returns
the result of a DeleteConsumerGroupOffsets request, or NULL if event is of different type.

Event types: RD_KAFKA_EVENT_DELETECONSUMERGROUPOFFSETS_RESULT

RD_EXPORT const rd_kafka_CreateAcls_result_t* rd_kafka_event_CreateAcls_result ( rd_kafka_event_t *  rkev)
Returns
the result of a CreateAcls request, or NULL if event is of different type.

Event types: RD_KAFKA_EVENT_CREATEACLS_RESULT

RD_EXPORT const rd_kafka_DescribeAcls_result_t* rd_kafka_event_DescribeAcls_result ( rd_kafka_event_t *  rkev)
Returns
the result of a DescribeAcls request, or NULL if event is of different type.

Event types: RD_KAFKA_EVENT_DESCRIBEACLS_RESULT

RD_EXPORT const rd_kafka_DeleteAcls_result_t* rd_kafka_event_DeleteAcls_result ( rd_kafka_event_t *  rkev)
Returns
the result of a DeleteAcls request, or NULL if event is of different type.

Event types: RD_KAFKA_EVENT_DELETEACLS_RESULT

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.

Returns
an event, or NULL.
Remarks
Use rd_kafka_event_destroy() to free the event.
See also
rd_kafka_conf_set_background_event_cb()
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.

Returns
the number of events served.
Remarks
This API must only be used for queues with callbacks registered for all expected event types. E.g., not a message queue.
Also see rd_kafka_conf_set_background_event_cb() for triggering event callbacks from a librdkafka-managed background thread.
See also
rd_kafka_conf_set_background_event_cb()
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.

Parameters
confConfiguration object.
ic_nameInterceptor name, used in logging.
on_conf_setFunction pointer.
ic_opaqueOpaque value that will be passed to the function.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or RD_KAFKA_RESP_ERR__CONFLICT if an existing intercepted with the same ic_name and function has already been added to conf.
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.

Parameters
confConfiguration object.
ic_nameInterceptor name, used in logging.
on_conf_dupFunction pointer.
ic_opaqueOpaque value that will be passed to the function.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or RD_KAFKA_RESP_ERR__CONFLICT if an existing intercepted with the same ic_name and function has already been added to conf.
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.

Parameters
confConfiguration object.
ic_nameInterceptor name, used in logging.
on_conf_destroyFunction pointer.
ic_opaqueOpaque value that will be passed to the function.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR
Remarks
Multiple on_conf_destroy() interceptors are allowed to be added to the same configuration object.
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.

Parameters
confConfiguration object.
ic_nameInterceptor name, used in logging.
on_newFunction pointer.
ic_opaqueOpaque value that will be passed to the function.
Remarks
Since the on_new() interceptor is added to the configuration object it may be copied by rd_kafka_conf_dup(). An interceptor implementation must thus be able to handle the same interceptor,ic_opaque tuple to be used by multiple client instances.
An interceptor plugin should check the return value to make sure it has not already been added.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or RD_KAFKA_RESP_ERR__CONFLICT if an existing intercepted with the same ic_name and function has already been added to conf.
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.

Parameters
rkClient instance.
ic_nameInterceptor name, used in logging.
on_destroyFunction pointer.
ic_opaqueOpaque value that will be passed to the function.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or RD_KAFKA_RESP_ERR__CONFLICT if an existing intercepted with the same ic_name and function has already been added to conf.
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.

Parameters
rkClient instance.
ic_nameInterceptor name, used in logging.
on_sendFunction pointer.
ic_opaqueOpaque value that will be passed to the function.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or RD_KAFKA_RESP_ERR__CONFLICT if an existing intercepted with the same ic_name and function has already been added to conf.
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.

Parameters
rkClient instance.
ic_nameInterceptor name, used in logging.
on_acknowledgementFunction pointer.
ic_opaqueOpaque value that will be passed to the function.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or RD_KAFKA_RESP_ERR__CONFLICT if an existing intercepted with the same ic_name and function has already been added to conf.
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.

Parameters
rkClient instance.
ic_nameInterceptor name, used in logging.
on_consumeFunction pointer.
ic_opaqueOpaque value that will be passed to the function.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or RD_KAFKA_RESP_ERR__CONFLICT if an existing intercepted with the same ic_name and function has already been added to conf.
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.

Parameters
rkClient instance.
ic_nameInterceptor name, used in logging.
on_commit()Function pointer.
ic_opaqueOpaque value that will be passed to the function.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or RD_KAFKA_RESP_ERR__CONFLICT if an existing intercepted with the same ic_name and function has already been added to conf.
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.

Parameters
rkClient instance.
ic_nameInterceptor name, used in logging.
on_request_sent()Function pointer.
ic_opaqueOpaque value that will be passed to the function.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or RD_KAFKA_RESP_ERR__CONFLICT if an existing intercepted with the same ic_name and function has already been added to conf.
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.

Parameters
rkClient instance.
ic_nameInterceptor name, used in logging.
on_response_received()Function pointer.
ic_opaqueOpaque value that will be passed to the function.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or RD_KAFKA_RESP_ERR__CONFLICT if an existing intercepted with the same ic_name and function has already been added to conf.
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.

Parameters
rkClient instance.
ic_nameInterceptor name, used in logging.
on_thread_start()Function pointer.
ic_opaqueOpaque value that will be passed to the function.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or RD_KAFKA_RESP_ERR__CONFLICT if an existing intercepted with the same ic_name and function has already been added to conf.
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.

Parameters
rkClient instance.
ic_nameInterceptor name, used in logging.
on_thread_exit()Function pointer.
ic_opaqueOpaque value that will be passed to the function.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or RD_KAFKA_RESP_ERR__CONFLICT if an existing intercepted with the same ic_name and function has already been added to conf.
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.

Returns
the error code for the given topic result.
RD_EXPORT const char* rd_kafka_topic_result_error_string ( const rd_kafka_topic_result_t *  topicres)
Returns
the human readable error string for the given topic result, or NULL if there was no error.
Remarks
lifetime of the returned string is the same as the topicres.
RD_EXPORT const char* rd_kafka_topic_result_name ( const rd_kafka_topic_result_t *  topicres)
Returns
the name of the topic for the given topic result.
Remarks
lifetime of the returned string is the same as the topicres.
RD_EXPORT const rd_kafka_error_t* rd_kafka_group_result_error ( const rd_kafka_group_result_t *  groupres)

Group result provides per-group operation result information.

Returns
the error for the given group result, or NULL on success.
Remarks
lifetime of the returned error is the same as the groupres.
RD_EXPORT const char* rd_kafka_group_result_name ( const rd_kafka_group_result_t *  groupres)
Returns
the name of the group for the given group result.
Remarks
lifetime of the returned string is the same as the groupres.
RD_EXPORT const rd_kafka_topic_partition_list_t* rd_kafka_group_result_partitions ( const rd_kafka_group_result_t *  groupres)
Returns
the partitions/offsets for the given group result, if applicable to the request type, else NULL.
Remarks
lifetime of the returned list is the same as the groupres.
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.

The options object is not modified by the Admin API request APIs, (e.g. CreateTopics) and may be reused for multiple calls.

Parameters
rkClient instance.
for_apiSpecifies what Admin API this AdminOptions object will be used for, which will enforce what AdminOptions_set_..() calls may be used based on the API, causing unsupported set..() calls to fail. Specifying RD_KAFKA_ADMIN_OP_ANY disables the enforcement allowing any option to be set, even if the option is not used in a future call to an Admin API method.
Returns
a new AdminOptions object (which must be freed with rd_kafka_AdminOptions_destroy()), or NULL if for_api was set to an unknown API op type.
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.

Parameters
optionsAdmin options.
timeout_msTimeout in milliseconds, use -1 for indefinite timeout. Defaults to socket.timeout.ms.
errstrA 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 in case of error.
errstr_sizeWritable size in errstr.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success, or RD_KAFKA_RESP_ERR__INVALID_ARG if timeout was out of range in which case an error string will be written errstr.
Remarks
This option is valid for all Admin API requests.
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.

CreateTopics: values <= 0 will return immediately after triggering topic creation, while > 0 will wait this long for topic creation to propagate in cluster. Default: 60 seconds.

DeleteTopics: same semantics as CreateTopics. CreatePartitions: same semantics as CreateTopics.

Parameters
optionsAdmin options.
timeout_msTimeout in milliseconds.
errstrA 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 in case of error.
errstr_sizeWritable size in errstr.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success, or RD_KAFKA_RESP_ERR__INVALID_ARG if timeout was out of range in which case an error string will be written errstr.
Remarks
This option is valid for CreateTopics, DeleteTopics, CreatePartitions, and DeleteRecords.
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).

Parameters
optionsAdmin options.
true_or_falseDefaults to false.
errstrA 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 in case of error.
errstr_sizeWritable size in errstr.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or an error code on failure in which case an error string will be written errstr.
Remarks
This option is valid for CreateTopics, CreatePartitions, AlterConfigs.
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.

By default, Admin requests are sent to the controller broker, with the following exceptions:

  • AlterConfigs with a BROKER resource are sent to the broker id set as the resource name.
  • DescribeConfigs with a BROKER resource are sent to the broker id set as the resource name.
Parameters
optionsAdmin Options.
broker_idThe broker to send the request to.
errstrA 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 in case of error.
errstr_sizeWritable size in errstr.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success or an error code on failure in which case an error string will be written errstr.
Remarks
This API should typically not be used, but serves as a workaround if new resource types are to the broker that the client does not know where to send.
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().

Parameters
topicTopic name to create.
num_partitionsNumber of partitions in topic, or -1 to use the broker's default partition count (>= 2.4.0).
replication_factorDefault replication factor for the topic's partitions, or -1 to use the broker's default replication factor (>= 2.4.0) or if set_replica_assignment() will be used.
errstrA 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 in case of error.
errstr_sizeWritable size in errstr.
Returns
a new allocated NewTopic object, or NULL if the input parameters are invalid. Use rd_kafka_NewTopic_destroy() to free object when done.
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).

Remarks
When this method is used, rd_kafka_NewTopic_new() must have been called with a replication_factor of -1.
An application must either set the replica assignment for all new partitions, or none.
If called, this function must be called consecutively for each partition, starting at 0.
Use rd_kafka_metadata() to retrieve the list of brokers in the cluster.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success, or an error code if the arguments were invalid.
See also
rd_kafka_AdminOptions_set_validate_only()
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.

Remarks
The name and value are not validated by the client, the validation takes place on the broker.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success, or an error code if the arguments were invalid.
See also
rd_kafka_AdminOptions_set_validate_only()
http://kafka.apache.org/documentation.html#topicconfigs
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.

Parameters
rkClient instance.
new_topicsArray of new topics to create.
new_topic_cntNumber of elements in new_topics array.
optionsOptional admin options, or NULL for defaults.
rkquQueue to emit result on.

Supported admin options:

Remarks
The result event type emitted on the supplied queue is of type RD_KAFKA_EVENT_CREATETOPICS_RESULT
RD_EXPORT const 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.

The returned topics life-time is the same as the result object.

Parameters
resultResult to get topics from.
cntpUpdated to the number of elements in the array.
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().

Parameters
topicTopic name to delete.
Returns
a new allocated DeleteTopic object. Use rd_kafka_DeleteTopic_destroy() to free object when done.
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.

Parameters
rkClient instance.
del_topicsArray of topics to delete.
del_topic_cntNumber of elements in topics array.
optionsOptional admin options, or NULL for defaults.
rkquQueue to emit result on.
Remarks
The result event type emitted on the supplied queue is of type RD_KAFKA_EVENT_DELETETOPICS_RESULT
RD_EXPORT const 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.

The returned topics life-time is the same as the result object.

Parameters
resultResult to get topic results from.
cntpis updated to the number of elements in the array.
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.

Parameters
topicTopic name to create more partitions for.
new_total_cntIncrease the topic's partition count to this value.
errstrA 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 in case of error.
errstr_sizeWritable size in errstr.
Returns
a new allocated NewPartitions object, or NULL if the input parameters are invalid. Use rd_kafka_NewPartitions_destroy() to free object when done.
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).

Remarks
An application must either set the replica assignment for all new partitions, or none.
If called, this function must be called consecutively for each new partition being created, where new_partition_idx 0 is the first new partition, 1 is the second, and so on.
broker_id_cnt should match the topic's replication factor.
Use rd_kafka_metadata() to retrieve the list of brokers in the cluster.
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success, or an error code if the arguments were invalid.
See also
rd_kafka_AdminOptions_set_validate_only()
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.

Parameters
rkClient instance.
new_partsArray of topics for which new partitions are to be created.
new_parts_cntNumber of elements in new_parts array.
optionsOptional admin options, or NULL for defaults.
rkquQueue to emit result on.

Supported admin options:

Remarks
The result event type emitted on the supplied queue is of type RD_KAFKA_EVENT_CREATEPARTITIONS_RESULT
RD_EXPORT const 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.

The returned topics life-time is the same as the result object.

Parameters
resultResult o get topic results from.
cntpis updated to the number of elements in the array.
RD_EXPORT const char* rd_kafka_ConfigSource_name ( rd_kafka_ConfigSource_t  confsource)
Returns
a string representation of the confsource.
RD_EXPORT const char* rd_kafka_ConfigEntry_name ( const rd_kafka_ConfigEntry_t *  entry)
Returns
the configuration property name
RD_EXPORT const char* rd_kafka_ConfigEntry_value ( const rd_kafka_ConfigEntry_t *  entry)
Returns
the configuration value, may be NULL for sensitive or unset properties.
RD_EXPORT rd_kafka_ConfigSource_t rd_kafka_ConfigEntry_source ( const rd_kafka_ConfigEntry_t *  entry)
Returns
the config source.
RD_EXPORT int rd_kafka_ConfigEntry_is_read_only ( const rd_kafka_ConfigEntry_t *  entry)
Returns
1 if the config property is read-only on the broker, else 0.
Remarks
Shall only be used on a DescribeConfigs result, otherwise returns -1.
RD_EXPORT int rd_kafka_ConfigEntry_is_default ( const rd_kafka_ConfigEntry_t *  entry)
Returns
1 if the config property is set to its default value on the broker, else 0.
Remarks
Shall only be used on a DescribeConfigs result, otherwise returns -1.
RD_EXPORT int rd_kafka_ConfigEntry_is_sensitive ( const rd_kafka_ConfigEntry_t *  entry)
Returns
1 if the config property contains sensitive information (such as security configuration), else 0.
Remarks
An application should take care not to include the value of sensitive configuration entries in its output.
Shall only be used on a DescribeConfigs result, otherwise returns -1.
RD_EXPORT int rd_kafka_ConfigEntry_is_synonym ( const rd_kafka_ConfigEntry_t *  entry)
Returns
1 if this entry is a synonym, else 0.
RD_EXPORT const rd_kafka_ConfigEntry_t** rd_kafka_ConfigEntry_synonyms ( const rd_kafka_ConfigEntry_t *  entry,
size_t *  cntp 
)
Returns
the synonym config entry array.
Parameters
entryEntry to get synonyms for.
cntpis updated to the number of elements in the array.
Remarks
The lifetime of the returned entry is the same as conf .
Shall only be used on a DescribeConfigs result, otherwise returns NULL.
RD_EXPORT const char* rd_kafka_ResourcePatternType_name ( rd_kafka_ResourcePatternType_t  resource_pattern_type)
Returns
a string representation of the resource_pattern_type
RD_EXPORT const char* rd_kafka_ResourceType_name ( rd_kafka_ResourceType_t  restype)
Returns
a string representation of the restype
RD_EXPORT rd_kafka_ConfigResource_t* rd_kafka_ConfigResource_new ( rd_kafka_ResourceType_t  restype,
const char *  resname 
)

Create new ConfigResource object.

Parameters
restypeThe resource type (e.g., RD_KAFKA_RESOURCE_TOPIC)
resnameThe resource name (e.g., the topic name)
Returns
a newly allocated object
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.

Parameters
configConfigResource to set config property on.
nameConfiguration name, depends on resource type.
valueConfiguration value, depends on resource type and name. Set to NULL to revert configuration value to default.

This will overwrite the current value.

Returns
RD_KAFKA_RESP_ERR_NO_ERROR if config was added to resource, or RD_KAFKA_RESP_ERR__INVALID_ARG on invalid input.
RD_EXPORT const 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.

The returned object life-times are the same as the config object.

Parameters
configConfigResource to get configs from.
cntpis updated to the number of elements in the array.
RD_EXPORT rd_kafka_ResourceType_t rd_kafka_ConfigResource_type ( const rd_kafka_ConfigResource_t *  config)
Returns
the ResourceType for config
RD_EXPORT const char* rd_kafka_ConfigResource_name ( const rd_kafka_ConfigResource_t *  config)
Returns
the name for config
RD_EXPORT rd_kafka_resp_err_t rd_kafka_ConfigResource_error ( const rd_kafka_ConfigResource_t *  config)
Returns
the error for this resource from an AlterConfigs request
RD_EXPORT const char* rd_kafka_ConfigResource_error_string ( const rd_kafka_ConfigResource_t *  config)
Returns
the error string for this resource from an AlterConfigs request, or NULL if no error.
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.

Remarks
Requires broker version >=0.11.0.0
Warning
AlterConfigs will replace all existing configuration for the provided resources with the new configuration given, reverting all other configuration to their default values.
Remarks
Multiple resources and resource types may be set, but at most one resource of type RD_KAFKA_RESOURCE_BROKER is allowed per call since these resource requests must be sent to the broker specified in the resource.
RD_EXPORT const 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.

Use rd_kafka_ConfigResource_error() and rd_kafka_ConfigResource_error_string() to extract per-resource error results on the returned array elements.

The returned object life-times are the same as the result object.

Parameters
resultResult object to get resource results from.
cntpis updated to the number of elements in the array.
Returns
an array of ConfigResource elements, or NULL if not available.
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.

The returned configuration includes default values and the rd_kafka_ConfigEntry_is_default() or rd_kafka_ConfigEntry_source() methods may be used to distinguish them from user supplied values.

The value of config entries where rd_kafka_ConfigEntry_is_sensitive() is true will always be NULL to avoid disclosing sensitive information, such as security settings.

Configuration entries where rd_kafka_ConfigEntry_is_read_only() is true can't be updated (with rd_kafka_AlterConfigs()).

Synonym configuration entries are returned if the broker supports it (broker version >= 1.1.0). See rd_kafka_ConfigEntry_synonyms().

Remarks
Requires broker version >=0.11.0.0
Multiple resources and resource types may be requested, but at most one resource of type RD_KAFKA_RESOURCE_BROKER is allowed per call since these resource requests must be sent to the broker specified in the resource.
RD_EXPORT const 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.

The returned resources life-time is the same as the result object.

Parameters
resultResult object to get resource results from.
cntpis updated to the number of elements in the array.
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().

before_offsets must contain topic, partition, and offset is the offset before which the messages will be deleted (exclusive). Set offset to RD_KAFKA_OFFSET_END (high-watermark) in order to delete all data in the partition.

Parameters
before_offsetsFor each partition delete all messages up to but not including the specified offset.
Returns
a new allocated DeleteRecords object. Use rd_kafka_DeleteRecords_destroy() to free object when done.
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.

Parameters
rkClient instance.
del_recordsThe offsets to delete (up to). Currently only one DeleteRecords_t (but containing multiple offsets) is supported.
del_record_cntThe number of elements in del_records, must be 1.
optionsOptional admin options, or NULL for defaults.
rkquQueue to emit result on.

Supported admin options:

Remarks
The result event type emitted on the supplied queue is of type RD_KAFKA_EVENT_DELETERECORDS_RESULT
RD_EXPORT const 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.

The returned object's life-time is the same as the result object.

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().

Parameters
groupName of group to delete.
Returns
a new allocated DeleteGroup object. Use rd_kafka_DeleteGroup_destroy() to free object when done.
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.

Parameters
rkClient instance.
del_groupsArray of groups to delete.
del_group_cntNumber of elements in del_groups array.
optionsOptional admin options, or NULL for defaults.
rkquQueue to emit result on.
Remarks
The result event type emitted on the supplied queue is of type RD_KAFKA_EVENT_DELETEGROUPS_RESULT
RD_EXPORT const 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.

The returned groups life-time is the same as the result object.

Parameters
resultResult to get group results from.
cntpis updated to the number of elements in the array.
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().

Parameters
groupConsumer group id.
partitionsPartitions to delete committed offsets for. Only the topic and partition fields are used.
Returns
a new allocated DeleteConsumerGroupOffsets object. Use rd_kafka_DeleteConsumerGroupOffsets_destroy() to free object when done.
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.

Parameters
rkClient instance.
del_grpoffsetsArray of group committed offsets to delete. MUST only be one single element.
del_grpoffsets_cntNumber of elements in del_grpoffsets array. MUST always be 1.
optionsOptional admin options, or NULL for defaults.
rkquQueue to emit result on.
Remarks
The result event type emitted on the supplied queue is of type RD_KAFKA_EVENT_DELETECONSUMERGROUPOFFSETS_RESULT
The current implementation only supports one group per invocation.
RD_EXPORT const 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.

The returned groups life-time is the same as the result object.

Parameters
resultResult to get group results from.
cntpis updated to the number of elements in the array.
RD_EXPORT const rd_kafka_error_t* rd_kafka_acl_result_error ( const rd_kafka_acl_result_t *  aclres)
Returns
the error object for the given acl result, or NULL on success.
RD_EXPORT const char* rd_kafka_AclOperation_name ( rd_kafka_AclOperation_t  acl_operation)
Returns
a string representation of the acl_operation
RD_EXPORT const char* rd_kafka_AclPermissionType_name ( rd_kafka_AclPermissionType_t  acl_permission_type)
Returns
a string representation of the acl_permission_type
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().

Parameters
restypeThe ResourceType.
nameThe resource name.
resource_pattern_typeThe pattern type.
principalA principal, following the kafka specification.
hostAn hostname or ip.
operationA Kafka operation.
permission_typeA Kafka permission type.
errstrAn error string for returning errors or NULL to not use it.
errstr_sizeThe errstr size or 0 to not use it.
Returns
a new allocated AclBinding object, or NULL if the input parameters are invalid. Use rd_kafka_AclBinding_destroy() to free object when done.
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.

Parameters
restypeThe ResourceType or RD_KAFKA_RESOURCE_ANY if not filtering by this field.
nameThe resource name or NULL if not filtering by this field.
resource_pattern_typeThe pattern type or RD_KAFKA_RESOURCE_PATTERN_ANY if not filtering by this field.
principalA principal or NULL if not filtering by this field.
hostAn hostname or ip or NULL if not filtering by this field.
operationA Kafka operation or RD_KAFKA_ACL_OPERATION_ANY if not filtering by this field.
permission_typeA Kafka permission type or RD_KAFKA_ACL_PERMISSION_TYPE_ANY if not filtering by this field.
errstrAn error string for returning errors or NULL to not use it.
errstr_sizeThe errstr size or 0 to not use it.
Returns
a new allocated AclBindingFilter object, or NULL if the input parameters are invalid. Use rd_kafka_AclBinding_destroy() to free object when done.
RD_EXPORT rd_kafka_ResourceType_t rd_kafka_AclBinding_restype ( const rd_kafka_AclBinding_t *  acl)
Returns
the resource type for the given acl binding.
RD_EXPORT const char* rd_kafka_AclBinding_name ( const rd_kafka_AclBinding_t *  acl)
Returns
the resource name for the given acl binding.
Remarks
lifetime of the returned string is the same as the acl.
RD_EXPORT const char* rd_kafka_AclBinding_principal ( const rd_kafka_AclBinding_t *  acl)
Returns
the principal for the given acl binding.
Remarks
lifetime of the returned string is the same as the acl.
RD_EXPORT const char* rd_kafka_AclBinding_host ( const rd_kafka_AclBinding_t *  acl)
Returns
the host for the given acl binding.
Remarks
lifetime of the returned string is the same as the acl.
RD_EXPORT rd_kafka_AclOperation_t rd_kafka_AclBinding_operation ( const rd_kafka_AclBinding_t *  acl)
Returns
the acl operation for the given acl binding.
RD_EXPORT rd_kafka_AclPermissionType_t rd_kafka_AclBinding_permission_type ( const rd_kafka_AclBinding_t *  acl)
Returns
the permission type for the given acl binding.
RD_EXPORT rd_kafka_ResourcePatternType_t rd_kafka_AclBinding_resource_pattern_type ( const rd_kafka_AclBinding_t *  acl)
Returns
the resource pattern type for the given acl binding.
RD_EXPORT const rd_kafka_error_t* rd_kafka_AclBinding_error ( const rd_kafka_AclBinding_t *  acl)
Returns
the error object for the given acl binding, or NULL on success.
RD_EXPORT const 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.

The returned acl result life-time is the same as the result object.

Parameters
resultCreateAcls result to get acl results from.
cntpis updated to the number of elements in the array.
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.

Parameters
rkClient instance.
new_aclsArray of new acls to create.
new_acls_cntNumber of elements in new_acls array.
optionsOptional admin options, or NULL for defaults.
rkquQueue to emit result on.

Supported admin options:

Remarks
The result event type emitted on the supplied queue is of type RD_KAFKA_EVENT_CREATEACLS_RESULT
RD_EXPORT const 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.

- describe access control lists.

The returned resources life-time is the same as the result object.

Parameters
resultDescribeAcls result to get acls from.
cntpis updated to the number of elements in the array.
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.

Parameters
rkClient instance.
acl_filterFilter for the returned acls.
optionsOptional admin options, or NULL for defaults.
rkquQueue to emit result on.

Supported admin options:

Remarks
The result event type emitted on the supplied queue is of type RD_KAFKA_EVENT_DESCRIBEACLS_RESULT
RD_EXPORT const 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.

The returned responses life-time is the same as the result object.

Parameters
resultDeleteAcls result to get responses from.
cntpis updated to the number of elements in the array.
RD_EXPORT const rd_kafka_error_t* rd_kafka_DeleteAcls_result_response_error ( const rd_kafka_DeleteAcls_result_response_t *  result_response)
Returns
the error object for the given DeleteAcls result response, or NULL on success.
RD_EXPORT const 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 
)
Returns
the matching acls array for the given DeleteAcls result response.
Remarks
lifetime of the returned acl bindings is the same as the result_response.
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.

Parameters
rkClient instance.
del_aclsFilters for the acls to delete.
del_acls_cntNumber of elements in del_acls array.
optionsOptional admin options, or NULL for defaults.
rkquQueue to emit result on.

Supported admin options:

Remarks
The result event type emitted on the supplied queue is of type RD_KAFKA_EVENT_DELETEACLS_RESULT
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.

Parameters
rkClient instance.
token_valuethe mandatory token value to set, often (but not necessarily) a JWS compact serialization as per https://tools.ietf.org/html/rfc7515#section-3.1.
md_lifetime_mswhen the token expires, in terms of the number of milliseconds since the epoch.
md_principal_namethe mandatory Kafka principal name associated with the token.
extensionsoptional SASL extensions key-value array with extensions_size elements (number of keys * 2), where [i] is the key and [i+1] is the key's value, to be communicated to the broker as additional key-value pairs during the initial client response as per https://tools.ietf.org/html/rfc7628#section-3.1. The key-value pairs are copied.
extension_sizethe number of SASL extension keys plus values, which must be a non-negative multiple of 2.
errstrA 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 in case of error.
errstr_sizeWritable size in errstr.

The SASL/OAUTHBEARER token refresh callback or event handler should invoke this method upon success. The extension keys must not include the reserved key "`auth`", and all extension keys and values must conform to the required format as per https://tools.ietf.org/html/rfc7628#section-3.1: 

key            = 1*(ALPHA)
value          = *(VCHAR / SP / HTAB / CR / LF )
Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success, otherwise errstr set and:
RD_KAFKA_RESP_ERR__INVALID_ARG if any of the arguments are invalid;
RD_KAFKA_RESP_ERR__NOT_IMPLEMENTED if SASL/OAUTHBEARER is not supported by this build;
RD_KAFKA_RESP_ERR__STATE if SASL/OAUTHBEARER is supported but is not configured as the client's authentication mechanism.
See also
rd_kafka_oauthbearer_set_token_failure
rd_kafka_conf_set_oauthbearer_token_refresh_cb
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.

Parameters
rkClient instance.
errstrmandatory human readable error reason for failing to acquire a token.

The SASL/OAUTHBEARER token refresh callback or event handler should invoke this method upon failure.

Returns
RD_KAFKA_RESP_ERR_NO_ERROR on success, otherwise:
RD_KAFKA_RESP_ERR__NOT_IMPLEMENTED if SASL/OAUTHBEARER is not supported by this build;
RD_KAFKA_RESP_ERR__STATE if SASL/OAUTHBEARER is supported but is not configured as the client's authentication mechanism,
RD_KAFKA_RESP_ERR__INVALID_ARG if no error string is supplied.
See also
rd_kafka_oauthbearer_set_token
rd_kafka_conf_set_oauthbearer_token_refresh_cb
RD_EXPORT rd_kafka_error_t* rd_kafka_init_transactions ( rd_kafka_t *  rk,
int  timeout_ms 
)

Initialize transactions for the producer instance.

This function ensures any transactions initiated by previous instances of the producer with the same transactional.id are completed. If the previous instance failed with a transaction in progress the previous transaction will be aborted. This function needs to be called before any other transactional or produce functions are called when the transactional.id is configured.

If the last transaction had begun completion (following transaction commit) but not yet finished, this function will await the previous transaction's completion.

When any previous transactions have been fenced this function will acquire the internal producer id and epoch, used in all future transactional messages issued by this producer instance.

Parameters
rkProducer instance.
timeout_msThe maximum time to block. On timeout the operation may continue in the background, depending on state, and it is okay to call init_transactions() again.
Remarks
This function may block up to timeout_ms milliseconds.
Returns
NULL on success or an error object on failure. Check whether the returned error object permits retrying by calling rd_kafka_error_is_retriable(), or whether a fatal error has been raised by calling rd_kafka_error_is_fatal(). Error codes: RD_KAFKA_RESP_ERR__TIMED_OUT if the transaction coordinator could be not be contacted within timeout_ms (retriable), RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE if the transaction coordinator is not available (retriable), RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS if a previous transaction would not complete within timeout_ms (retriable), RD_KAFKA_RESP_ERR__STATE if transactions have already been started or upon fatal error, RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE if the broker(s) do not support transactions (<Apache Kafka 0.11), this also raises a fatal error, RD_KAFKA_RESP_ERR_INVALID_TRANSACTION_TIMEOUT if the configured transaction.timeout.ms is outside the broker-configured range, this also raises a fatal error, RD_KAFKA_RESP_ERR__NOT_CONFIGURED if transactions have not been configured for the producer instance, RD_KAFKA_RESP_ERR__INVALID_ARG if rk is not a producer instance, or timeout_ms is out of range. Other error codes not listed here may be returned, depending on broker version.
Remarks
The returned error object (if not NULL) must be destroyed with rd_kafka_error_destroy().
RD_EXPORT rd_kafka_error_t* rd_kafka_begin_transaction ( rd_kafka_t *  rk)

Begin a new transaction.

rd_kafka_init_transactions() must have been called successfully (once) before this function is called.

Upon successful return from this function the application has to perform at least one of the following operations within transaction.timeout.ms to avoid timing out the transaction on the broker:

Any messages produced, offsets sent (rd_kafka_send_offsets_to_transaction()), etc, after the successful return of this function will be part of the transaction and committed or aborted atomatically.

Finish the transaction by calling rd_kafka_commit_transaction() or abort the transaction by calling rd_kafka_abort_transaction().

Parameters
rkProducer instance.
Returns
NULL on success or an error object on failure. Check whether a fatal error has been raised by calling rd_kafka_error_is_fatal(). Error codes: RD_KAFKA_RESP_ERR__STATE if a transaction is already in progress or upon fatal error, RD_KAFKA_RESP_ERR__NOT_CONFIGURED if transactions have not been configured for the producer instance, RD_KAFKA_RESP_ERR__INVALID_ARG if rk is not a producer instance. Other error codes not listed here may be returned, depending on broker version.
Remarks
With the transactional producer, rd_kafka_produce(), rd_kafka_producev(), et.al, are only allowed during an on-going transaction, as started with this function. Any produce call outside an on-going transaction, or for a failed transaction, will fail.
The returned error object (if not NULL) must be destroyed with rd_kafka_error_destroy().
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.

The offsets should be the next message your application will consume, i.e., the last processed message's offset + 1 for each partition. Either track the offsets manually during processing or use rd_kafka_position() (on the consumer) to get the current offsets for the partitions assigned to the consumer.

Use this method at the end of a consume-transform-produce loop prior to committing the transaction with rd_kafka_commit_transaction().

Parameters
rkProducer instance.
offsetsList of offsets to commit to the consumer group upon successful commit of the transaction. Offsets should be the next message to consume, e.g., last processed message + 1.
cgmetadataThe current consumer group metadata as returned by rd_kafka_consumer_group_metadata() on the consumer instance the provided offsets were consumed from.
timeout_msMaximum time allowed to register the offsets on the broker.
Remarks
This function must be called on the transactional producer instance, not the consumer.
The consumer must disable auto commits (set enable.auto.commit to false on the consumer).
Logical and invalid offsets (such as RD_KAFKA_OFFSET_INVALID) in offsets will be ignored, if there are no valid offsets in offsets the function will return RD_KAFKA_RESP_ERR_NO_ERROR and no action will be taken.
Returns
NULL on success or an error object on failure. Check whether the returned error object permits retrying by calling rd_kafka_error_is_retriable(), or whether an abortable or fatal error has been raised by calling rd_kafka_error_txn_requires_abort() or rd_kafka_error_is_fatal() respectively. Error codes: RD_KAFKA_RESP_ERR__STATE if not currently in a transaction, RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH if the current producer transaction has been fenced by a newer producer instance, RD_KAFKA_RESP_ERR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED if the producer is no longer authorized to perform transactional operations, RD_KAFKA_RESP_ERR_GROUP_AUTHORIZATION_FAILED if the producer is not authorized to write the consumer offsets to the group coordinator, RD_KAFKA_RESP_ERR__NOT_CONFIGURED if transactions have not been configured for the producer instance, RD_KAFKA_RESP_ERR__INVALID_ARG if rk is not a producer instance, or if the consumer_group_id or offsets are empty, RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS if a previous rd_kafka_send_offsets_to_transaction() call is still in progress. Other error codes not listed here may be returned, depending on broker version.
Remarks
The returned error object (if not NULL) must be destroyed with rd_kafka_error_destroy().
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()).

Any outstanding messages will be flushed (delivered) before actually committing the transaction.

If any of the outstanding messages fail permanently the current transaction will enter the abortable error state and this function will return an abortable error, in this case the application must call rd_kafka_abort_transaction() before attempting a new transaction with rd_kafka_begin_transaction().

Parameters
rkProducer instance.
timeout_msThe maximum time to block. On timeout the operation may continue in the background, depending on state, and it is okay to call this function again. Pass -1 to use the remaining transaction timeout, this is the recommended use.
Remarks
It is strongly recommended to always pass -1 (remaining transaction time) as the timeout_ms. Using other values risk internal state desynchronization in case any of the underlying protocol requests fail.
This function will block until all outstanding messages are delivered and the transaction commit request has been successfully handled by the transaction coordinator, or until timeout_ms expires, which ever comes first. On timeout the application may call the function again.
Will automatically call rd_kafka_flush() to ensure all queued messages are delivered before attempting to commit the transaction. If the application has enabled RD_KAFKA_EVENT_DR it must serve the event queue in a separate thread since rd_kafka_flush() will not serve delivery reports in this mode.
Returns
NULL on success or an error object on failure. Check whether the returned error object permits retrying by calling rd_kafka_error_is_retriable(), or whether an abortable or fatal error has been raised by calling rd_kafka_error_txn_requires_abort() or rd_kafka_error_is_fatal() respectively. Error codes: RD_KAFKA_RESP_ERR__STATE if not currently in a transaction, RD_KAFKA_RESP_ERR__TIMED_OUT if the transaction could not be complete commmitted within timeout_ms, this is a retriable error as the commit continues in the background, RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH if the current producer transaction has been fenced by a newer producer instance, RD_KAFKA_RESP_ERR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED if the producer is no longer authorized to perform transactional operations, RD_KAFKA_RESP_ERR__NOT_CONFIGURED if transactions have not been configured for the producer instance, RD_KAFKA_RESP_ERR__INVALID_ARG if rk is not a producer instance, Other error codes not listed here may be returned, depending on broker version.
Remarks
The returned error object (if not NULL) must be destroyed with rd_kafka_error_destroy().
RD_EXPORT rd_kafka_error_t* rd_kafka_abort_transaction ( rd_kafka_t *  rk,
int  timeout_ms 
)

Aborts the ongoing transaction.

This function should also be used to recover from non-fatal abortable transaction errors.

Any outstanding messages will be purged and fail with RD_KAFKA_RESP_ERR__PURGE_INFLIGHT or RD_KAFKA_RESP_ERR__PURGE_QUEUE. See rd_kafka_purge() for details.

Parameters
rkProducer instance.
timeout_msThe maximum time to block. On timeout the operation may continue in the background, depending on state, and it is okay to call this function again. Pass -1 to use the remaining transaction timeout, this is the recommended use.
Remarks
It is strongly recommended to always pass -1 (remaining transaction time) as the timeout_ms. Using other values risk internal state desynchronization in case any of the underlying protocol requests fail.
This function will block until all outstanding messages are purged and the transaction abort request has been successfully handled by the transaction coordinator, or until timeout_ms expires, which ever comes first. On timeout the application may call the function again. If the application has enabled RD_KAFKA_EVENT_DR it must serve the event queue in a separate thread since rd_kafka_flush() will not serve delivery reports in this mode.
Returns
NULL on success or an error object on failure. Check whether the returned error object permits retrying by calling rd_kafka_error_is_retriable(), or whether a fatal error has been raised by calling rd_kafka_error_is_fatal(). Error codes: RD_KAFKA_RESP_ERR__STATE if not currently in a transaction, RD_KAFKA_RESP_ERR__TIMED_OUT if the transaction could not be complete commmitted within timeout_ms, this is a retriable error as the commit continues in the background, RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH if the current producer transaction has been fenced by a newer producer instance, RD_KAFKA_RESP_ERR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED if the producer is no longer authorized to perform transactional operations, RD_KAFKA_RESP_ERR__NOT_CONFIGURED if transactions have not been configured for the producer instance, RD_KAFKA_RESP_ERR__INVALID_ARG if rk is not a producer instance, Other error codes not listed here may be returned, depending on broker version.
Remarks
The returned error object (if not NULL) must be destroyed with rd_kafka_error_destroy().