http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src/rdkafka.h ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src/rdkafka.h b/thirdparty/librdkafka-0.11.1/src/rdkafka.h deleted file mode 100644 index efb781c..0000000 --- a/thirdparty/librdkafka-0.11.1/src/rdkafka.h +++ /dev/null @@ -1,3820 +0,0 @@ -/* - * librdkafka - Apache Kafka C library - * - * Copyright (c) 2012-2013 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -/** - * @file rdkafka.h - * @brief 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. - * - * @sa For the C++ interface see rdkafkacpp.h - * - * @tableofcontents - */ - - -/* @cond NO_DOC */ -#pragma once - -#include <stdio.h> -#include <inttypes.h> -#include <sys/types.h> - -#ifdef __cplusplus -extern "C" { -#if 0 -} /* Restore indent */ -#endif -#endif - -#ifdef _MSC_VER -#include <basetsd.h> -#ifndef WIN32_MEAN_AND_LEAN -#define WIN32_MEAN_AND_LEAN -#endif -#include <Winsock2.h> /* for sockaddr, .. */ -typedef SSIZE_T ssize_t; -#define RD_UNUSED -#define RD_INLINE __inline -#define RD_DEPRECATED __declspec(deprecated) -#undef RD_EXPORT -#ifdef LIBRDKAFKA_STATICLIB -#define RD_EXPORT -#else -#ifdef LIBRDKAFKA_EXPORTS -#define RD_EXPORT __declspec(dllexport) -#else -#define RD_EXPORT __declspec(dllimport) -#endif -#ifndef LIBRDKAFKA_TYPECHECKS -#define LIBRDKAFKA_TYPECHECKS 0 -#endif -#endif - -#else -#include <sys/socket.h> /* for sockaddr, .. */ - -#define RD_UNUSED __attribute__((unused)) -#define RD_INLINE inline -#define RD_EXPORT -#define RD_DEPRECATED __attribute__((deprecated)) - -#ifndef LIBRDKAFKA_TYPECHECKS -#define LIBRDKAFKA_TYPECHECKS 1 -#endif -#endif - - -/** - * @brief Type-checking macros - * Compile-time checking that \p ARG is of type \p TYPE. - * @returns \p RET - */ -#if LIBRDKAFKA_TYPECHECKS -#define _LRK_TYPECHECK(RET,TYPE,ARG) \ - ({ if (0) { TYPE __t RD_UNUSED = (ARG); } RET; }) - -#define _LRK_TYPECHECK2(RET,TYPE,ARG,TYPE2,ARG2) \ - ({ \ - if (0) { \ - TYPE __t RD_UNUSED = (ARG); \ - TYPE2 __t2 RD_UNUSED = (ARG2); \ - } \ - RET; }) -#else -#define _LRK_TYPECHECK(RET,TYPE,ARG) (RET) -#define _LRK_TYPECHECK2(RET,TYPE,ARG,TYPE2,ARG2) (RET) -#endif - -/* @endcond */ - - -/** - * @name librdkafka version - * @{ - * - * - */ - -/** - * @brief librdkafka version - * - * Interpreted as hex \c MM.mm.rr.xx: - * - MM = Major - * - mm = minor - * - rr = revision - * - xx = pre-release id (0xff is the final release) - * - * E.g.: \c 0x000801ff = 0.8.1 - * - * @remark This value should only be used during compile time, - * for runtime checks of version use rd_kafka_version() - */ -#define RD_KAFKA_VERSION 0x000b01ff - -/** - * @brief Returns the librdkafka version as integer. - * - * @returns Version integer. - * - * @sa See RD_KAFKA_VERSION for how to parse the integer format. - * @sa Use rd_kafka_version_str() to retreive the version as a string. - */ -RD_EXPORT -int rd_kafka_version(void); - -/** - * @brief Returns the librdkafka version as string. - * - * @returns Version string - */ -RD_EXPORT -const char *rd_kafka_version_str (void); - -/**@}*/ - - -/** - * @name Constants, errors, types - * @{ - * - * - */ - - -/** - * @enum rd_kafka_type_t - * - * @brief rd_kafka_t handle type. - * - * @sa rd_kafka_new() - */ -typedef enum rd_kafka_type_t { - RD_KAFKA_PRODUCER, /**< Producer client */ - RD_KAFKA_CONSUMER /**< Consumer client */ -} rd_kafka_type_t; - - -/** - * @enum Timestamp types - * - * @sa rd_kafka_message_timestamp() - */ -typedef enum rd_kafka_timestamp_type_t { - RD_KAFKA_TIMESTAMP_NOT_AVAILABLE, /**< Timestamp not available */ - RD_KAFKA_TIMESTAMP_CREATE_TIME, /**< Message creation time */ - RD_KAFKA_TIMESTAMP_LOG_APPEND_TIME /**< Log append time */ -} rd_kafka_timestamp_type_t; - - - -/** - * @brief Retrieve supported debug contexts for use with the \c \"debug\" - * configuration property. (runtime) - * - * @returns Comma-separated list of available debugging contexts. - */ -RD_EXPORT -const char *rd_kafka_get_debug_contexts(void); - -/** - * @brief 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_DEBUG_CONTEXTS \ - "all,generic,broker,topic,metadata,queue,msg,protocol,cgrp,security,fetch,feature" - - -/* @cond NO_DOC */ -/* Private types to provide ABI compatibility */ -typedef struct rd_kafka_s rd_kafka_t; -typedef struct rd_kafka_topic_s rd_kafka_topic_t; -typedef struct rd_kafka_conf_s rd_kafka_conf_t; -typedef struct rd_kafka_topic_conf_s rd_kafka_topic_conf_t; -typedef struct rd_kafka_queue_s rd_kafka_queue_t; -/* @endcond */ - - -/** - * @enum rd_kafka_resp_err_t - * @brief Error codes. - * - * The negative error codes delimited by two underscores - * (\c RD_KAFKA_RESP_ERR__..) denotes errors internal to librdkafka and are - * displayed as \c \"Local: \<error string..\>\", while the error codes - * delimited by a single underscore (\c RD_KAFKA_RESP_ERR_..) denote broker - * errors and are displayed as \c \"Broker: \<error string..\>\". - * - * @sa Use rd_kafka_err2str() to translate an error code a human readable string - */ -typedef enum { - /* Internal errors to rdkafka: */ - /** Begin internal error codes */ - RD_KAFKA_RESP_ERR__BEGIN = -200, - /** Received message is incorrect */ - RD_KAFKA_RESP_ERR__BAD_MSG = -199, - /** Bad/unknown compression */ - RD_KAFKA_RESP_ERR__BAD_COMPRESSION = -198, - /** Broker is going away */ - RD_KAFKA_RESP_ERR__DESTROY = -197, - /** Generic failure */ - RD_KAFKA_RESP_ERR__FAIL = -196, - /** Broker transport failure */ - RD_KAFKA_RESP_ERR__TRANSPORT = -195, - /** Critical system resource */ - RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE = -194, - /** Failed to resolve broker */ - RD_KAFKA_RESP_ERR__RESOLVE = -193, - /** Produced message timed out*/ - RD_KAFKA_RESP_ERR__MSG_TIMED_OUT = -192, - /** Reached the end of the topic+partition queue on - * the broker. Not really an error. */ - RD_KAFKA_RESP_ERR__PARTITION_EOF = -191, - /** Permanent: Partition does not exist in cluster. */ - RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION = -190, - /** File or filesystem error */ - RD_KAFKA_RESP_ERR__FS = -189, - /** Permanent: Topic does not exist in cluster. */ - RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC = -188, - /** All broker connections are down. */ - RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN = -187, - /** Invalid argument, or invalid configuration */ - RD_KAFKA_RESP_ERR__INVALID_ARG = -186, - /** Operation timed out */ - RD_KAFKA_RESP_ERR__TIMED_OUT = -185, - /** Queue is full */ - RD_KAFKA_RESP_ERR__QUEUE_FULL = -184, - /** ISR count < required.acks */ - RD_KAFKA_RESP_ERR__ISR_INSUFF = -183, - /** Broker node update */ - RD_KAFKA_RESP_ERR__NODE_UPDATE = -182, - /** SSL error */ - RD_KAFKA_RESP_ERR__SSL = -181, - /** Waiting for coordinator to become available. */ - RD_KAFKA_RESP_ERR__WAIT_COORD = -180, - /** Unknown client group */ - RD_KAFKA_RESP_ERR__UNKNOWN_GROUP = -179, - /** Operation in progress */ - RD_KAFKA_RESP_ERR__IN_PROGRESS = -178, - /** Previous operation in progress, wait for it to finish. */ - RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS = -177, - /** This operation would interfere with an existing subscription */ - RD_KAFKA_RESP_ERR__EXISTING_SUBSCRIPTION = -176, - /** Assigned partitions (rebalance_cb) */ - RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS = -175, - /** Revoked partitions (rebalance_cb) */ - RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS = -174, - /** Conflicting use */ - RD_KAFKA_RESP_ERR__CONFLICT = -173, - /** Wrong state */ - RD_KAFKA_RESP_ERR__STATE = -172, - /** Unknown protocol */ - RD_KAFKA_RESP_ERR__UNKNOWN_PROTOCOL = -171, - /** Not implemented */ - RD_KAFKA_RESP_ERR__NOT_IMPLEMENTED = -170, - /** Authentication failure*/ - RD_KAFKA_RESP_ERR__AUTHENTICATION = -169, - /** No stored offset */ - RD_KAFKA_RESP_ERR__NO_OFFSET = -168, - /** Outdated */ - RD_KAFKA_RESP_ERR__OUTDATED = -167, - /** Timed out in queue */ - RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE = -166, - /** Feature not supported by broker */ - RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE = -165, - /** Awaiting cache update */ - RD_KAFKA_RESP_ERR__WAIT_CACHE = -164, - /** Operation interrupted (e.g., due to yield)) */ - RD_KAFKA_RESP_ERR__INTR = -163, - /** Key serialization error */ - RD_KAFKA_RESP_ERR__KEY_SERIALIZATION = -162, - /** Value serialization error */ - RD_KAFKA_RESP_ERR__VALUE_SERIALIZATION = -161, - /** Key deserialization error */ - RD_KAFKA_RESP_ERR__KEY_DESERIALIZATION = -160, - /** Value deserialization error */ - RD_KAFKA_RESP_ERR__VALUE_DESERIALIZATION = -159, - - /** End internal error codes */ - RD_KAFKA_RESP_ERR__END = -100, - - /* Kafka broker errors: */ - /** Unknown broker error */ - RD_KAFKA_RESP_ERR_UNKNOWN = -1, - /** Success */ - RD_KAFKA_RESP_ERR_NO_ERROR = 0, - /** Offset out of range */ - RD_KAFKA_RESP_ERR_OFFSET_OUT_OF_RANGE = 1, - /** Invalid message */ - RD_KAFKA_RESP_ERR_INVALID_MSG = 2, - /** Unknown topic or partition */ - RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART = 3, - /** Invalid message size */ - RD_KAFKA_RESP_ERR_INVALID_MSG_SIZE = 4, - /** Leader not available */ - RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE = 5, - /** Not leader for partition */ - RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION = 6, - /** Request timed out */ - RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT = 7, - /** Broker not available */ - RD_KAFKA_RESP_ERR_BROKER_NOT_AVAILABLE = 8, - /** Replica not available */ - RD_KAFKA_RESP_ERR_REPLICA_NOT_AVAILABLE = 9, - /** Message size too large */ - RD_KAFKA_RESP_ERR_MSG_SIZE_TOO_LARGE = 10, - /** StaleControllerEpochCode */ - RD_KAFKA_RESP_ERR_STALE_CTRL_EPOCH = 11, - /** Offset metadata string too large */ - RD_KAFKA_RESP_ERR_OFFSET_METADATA_TOO_LARGE = 12, - /** Broker disconnected before response received */ - RD_KAFKA_RESP_ERR_NETWORK_EXCEPTION = 13, - /** Group coordinator load in progress */ - RD_KAFKA_RESP_ERR_GROUP_LOAD_IN_PROGRESS = 14, - /** Group coordinator not available */ - RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE = 15, - /** Not coordinator for group */ - RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP = 16, - /** Invalid topic */ - RD_KAFKA_RESP_ERR_TOPIC_EXCEPTION = 17, - /** Message batch larger than configured server segment size */ - RD_KAFKA_RESP_ERR_RECORD_LIST_TOO_LARGE = 18, - /** Not enough in-sync replicas */ - RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS = 19, - /** Message(s) written to insufficient number of in-sync replicas */ - RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS_AFTER_APPEND = 20, - /** Invalid required acks value */ - RD_KAFKA_RESP_ERR_INVALID_REQUIRED_ACKS = 21, - /** Specified group generation id is not valid */ - RD_KAFKA_RESP_ERR_ILLEGAL_GENERATION = 22, - /** Inconsistent group protocol */ - RD_KAFKA_RESP_ERR_INCONSISTENT_GROUP_PROTOCOL = 23, - /** Invalid group.id */ - RD_KAFKA_RESP_ERR_INVALID_GROUP_ID = 24, - /** Unknown member */ - RD_KAFKA_RESP_ERR_UNKNOWN_MEMBER_ID = 25, - /** Invalid session timeout */ - RD_KAFKA_RESP_ERR_INVALID_SESSION_TIMEOUT = 26, - /** Group rebalance in progress */ - RD_KAFKA_RESP_ERR_REBALANCE_IN_PROGRESS = 27, - /** Commit offset data size is not valid */ - RD_KAFKA_RESP_ERR_INVALID_COMMIT_OFFSET_SIZE = 28, - /** Topic authorization failed */ - RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED = 29, - /** Group authorization failed */ - RD_KAFKA_RESP_ERR_GROUP_AUTHORIZATION_FAILED = 30, - /** Cluster authorization failed */ - RD_KAFKA_RESP_ERR_CLUSTER_AUTHORIZATION_FAILED = 31, - /** Invalid timestamp */ - RD_KAFKA_RESP_ERR_INVALID_TIMESTAMP = 32, - /** Unsupported SASL mechanism */ - RD_KAFKA_RESP_ERR_UNSUPPORTED_SASL_MECHANISM = 33, - /** Illegal SASL state */ - RD_KAFKA_RESP_ERR_ILLEGAL_SASL_STATE = 34, - /** Unuspported version */ - RD_KAFKA_RESP_ERR_UNSUPPORTED_VERSION = 35, - /** Topic already exists */ - RD_KAFKA_RESP_ERR_TOPIC_ALREADY_EXISTS = 36, - /** Invalid number of partitions */ - RD_KAFKA_RESP_ERR_INVALID_PARTITIONS = 37, - /** Invalid replication factor */ - RD_KAFKA_RESP_ERR_INVALID_REPLICATION_FACTOR = 38, - /** Invalid replica assignment */ - RD_KAFKA_RESP_ERR_INVALID_REPLICA_ASSIGNMENT = 39, - /** Invalid config */ - RD_KAFKA_RESP_ERR_INVALID_CONFIG = 40, - /** Not controller for cluster */ - RD_KAFKA_RESP_ERR_NOT_CONTROLLER = 41, - /** Invalid request */ - RD_KAFKA_RESP_ERR_INVALID_REQUEST = 42, - /** Message format on broker does not support request */ - RD_KAFKA_RESP_ERR_UNSUPPORTED_FOR_MESSAGE_FORMAT = 43, - /** Isolation policy volation */ - RD_KAFKA_RESP_ERR_POLICY_VIOLATION = 44, - /** Broker received an out of order sequence number */ - RD_KAFKA_RESP_ERR_OUT_OF_ORDER_SEQUENCE_NUMBER = 45, - /** Broker received a duplicate sequence number */ - RD_KAFKA_RESP_ERR_DUPLICATE_SEQUENCE_NUMBER = 46, - /** Producer attempted an operation with an old epoch */ - RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH = 47, - /** Producer attempted a transactional operation in an invalid state */ - RD_KAFKA_RESP_ERR_INVALID_TXN_STATE = 48, - /** Producer attempted to use a producer id which is not - * currently assigned to its transactional id */ - RD_KAFKA_RESP_ERR_INVALID_PRODUCER_ID_MAPPING = 49, - /** Transaction timeout is larger than the maximum - * value allowed by the broker's max.transaction.timeout.ms */ - RD_KAFKA_RESP_ERR_INVALID_TRANSACTION_TIMEOUT = 50, - /** Producer attempted to update a transaction while another - * concurrent operation on the same transaction was ongoing */ - RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS = 51, - /** Indicates that the transaction coordinator sending a - * WriteTxnMarker is no longer the current coordinator for a - * given producer */ - RD_KAFKA_RESP_ERR_TRANSACTION_COORDINATOR_FENCED = 52, - /** Transactional Id authorization failed */ - RD_KAFKA_RESP_ERR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED = 53, - /** Security features are disabled */ - RD_KAFKA_RESP_ERR_SECURITY_DISABLED = 54, - /** Operation not attempted */ - RD_KAFKA_RESP_ERR_OPERATION_NOT_ATTEMPTED = 55, - - RD_KAFKA_RESP_ERR_END_ALL, -} rd_kafka_resp_err_t; - - -/** - * @brief Error code value, name and description. - * Typically for use with language bindings to automatically expose - * the full set of librdkafka error codes. - */ -struct rd_kafka_err_desc { - rd_kafka_resp_err_t code;/**< Error code */ - const char *name; /**< Error name, same as code enum sans prefix */ - const char *desc; /**< Human readable error description. */ -}; - - -/** - * @brief Returns the full list of error codes. - */ -RD_EXPORT -void rd_kafka_get_err_descs (const struct rd_kafka_err_desc **errdescs, - size_t *cntp); - - - - -/** - * @brief Returns a human readable representation of a kafka error. - * - * @param err Error code to translate - */ -RD_EXPORT -const char *rd_kafka_err2str (rd_kafka_resp_err_t err); - - - -/** - * @brief Returns the error code name (enum name). - * - * @param err Error code to translate - */ -RD_EXPORT -const char *rd_kafka_err2name (rd_kafka_resp_err_t err); - - -/** - * @brief 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: - * - rd_kafka_topic_new() - * - rd_kafka_consume_start() - * - rd_kafka_consume_stop() - * - rd_kafka_consume() - * - rd_kafka_consume_batch() - * - rd_kafka_consume_callback() - * - rd_kafka_consume_queue() - * - rd_kafka_produce() - * - * The main use for this function is to avoid converting system \p errno - * values to rd_kafka_resp_err_t codes for legacy APIs. - * - * @remark 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. - * - * @remark errno propagation from librdkafka is not safe on Windows - * and should not be used, use rd_kafka_last_error() instead. - */ -RD_EXPORT -rd_kafka_resp_err_t rd_kafka_last_error (void); - - -/** - * @brief Converts the system errno value \p errnox to a rd_kafka_resp_err_t - * error code upon failure from the following functions: - * - rd_kafka_topic_new() - * - rd_kafka_consume_start() - * - rd_kafka_consume_stop() - * - rd_kafka_consume() - * - rd_kafka_consume_batch() - * - rd_kafka_consume_callback() - * - rd_kafka_consume_queue() - * - rd_kafka_produce() - * - * @param errnox System errno value to convert - * - * @returns Appropriate error code for \p errnox - * - * @remark 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. - * - * @sa rd_kafka_last_error() - */ -RD_EXPORT RD_DEPRECATED -rd_kafka_resp_err_t rd_kafka_errno2err(int errnox); - - -/** - * @brief Returns the thread-local system errno - * - * On most platforms this is the same as \p 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. - * - * @remark 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_DEPRECATED -int rd_kafka_errno (void); - - - -/** - * @brief Topic+Partition place holder - * - * Generic place holder for a Topic+Partition and its related information - * used for multiple purposes: - * - consumer offset (see rd_kafka_commit(), et.al.) - * - group rebalancing callback (rd_kafka_conf_set_rebalance_cb()) - * - offset commit result callback (rd_kafka_conf_set_offset_commit_cb()) - */ - -/** - * @brief Generic place holder for a specific Topic+Partition. - * - * @sa rd_kafka_topic_partition_list_new() - */ -typedef struct rd_kafka_topic_partition_s { - char *topic; /**< Topic name */ - int32_t partition; /**< Partition */ - int64_t offset; /**< Offset */ - void *metadata; /**< Metadata */ - size_t metadata_size; /**< Metadata size */ - void *opaque; /**< Application opaque */ - rd_kafka_resp_err_t err; /**< Error code, depending on use. */ - void *_private; /**< INTERNAL USE ONLY, - * INITIALIZE TO ZERO, DO NOT TOUCH */ -} rd_kafka_topic_partition_t; - - -/** - * @brief Destroy a rd_kafka_topic_partition_t. - * @remark This must not be called for elements in a topic partition list. - */ -RD_EXPORT -void rd_kafka_topic_partition_destroy (rd_kafka_topic_partition_t *rktpar); - - -/** - * @brief A growable list of Topic+Partitions. - * - */ -typedef struct rd_kafka_topic_partition_list_s { - int cnt; /**< Current number of elements */ - int size; /**< Current allocated size */ - rd_kafka_topic_partition_t *elems; /**< Element array[] */ -} rd_kafka_topic_partition_list_t; - - -/** - * @brief Create a new list/vector Topic+Partition container. - * - * @param size Initial allocated size used when the expected number of - * elements is known or can be estimated. - * Avoids reallocation and possibly relocation of the - * elems array. - * - * @returns A newly allocated Topic+Partition list. - * - * @remark Use rd_kafka_topic_partition_list_destroy() to free all resources - * in use by a list and the list itself. - * @sa rd_kafka_topic_partition_list_add() - */ -RD_EXPORT -rd_kafka_topic_partition_list_t *rd_kafka_topic_partition_list_new (int size); - - -/** - * @brief Free all resources used by the list and the list itself. - */ -RD_EXPORT -void -rd_kafka_topic_partition_list_destroy (rd_kafka_topic_partition_list_t *rkparlist); - -/** - * @brief Add topic+partition to list - * - * @param rktparlist List to extend - * @param topic Topic name (copied) - * @param partition Partition id - * - * @returns The object which can be used to fill in additionals fields. - */ -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); - - -/** - * @brief Add range of partitions from \p start to \p stop inclusive. - * - * @param rktparlist List to extend - * @param topic Topic name (copied) - * @param start Start partition of range - * @param stop Last partition of range (inclusive) - */ -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); - - - -/** - * @brief Delete partition from list. - * - * @param rktparlist List to modify - * @param topic Topic name to match - * @param partition Partition to match - * - * @returns 1 if partition was found (and removed), else 0. - * - * @remark Any held indices to elems[] are unusable after this call returns 1. - */ -RD_EXPORT -int -rd_kafka_topic_partition_list_del (rd_kafka_topic_partition_list_t *rktparlist, - const char *topic, int32_t partition); - - -/** - * @brief Delete partition from list by elems[] index. - * - * @returns 1 if partition was found (and removed), else 0. - * - * @sa rd_kafka_topic_partition_list_del() - */ -RD_EXPORT -int -rd_kafka_topic_partition_list_del_by_idx ( - rd_kafka_topic_partition_list_t *rktparlist, - int idx); - - -/** - * @brief Make a copy of an existing list. - * - * @param src The existing list to copy. - * - * @returns A new list fully populated to be identical to \p src - */ -RD_EXPORT -rd_kafka_topic_partition_list_t * -rd_kafka_topic_partition_list_copy (const rd_kafka_topic_partition_list_t *src); - - - - -/** - * @brief Set offset to \p offset for \p topic and \p partition - * - * @returns RD_KAFKA_RESP_ERR_NO_ERROR on success or - * RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION if \p partition was not found - * in the list. - */ -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); - - - -/** - * @brief Find element by \p topic and \p partition. - * - * @returns a pointer to the first matching element, or NULL if not found. - */ -RD_EXPORT -rd_kafka_topic_partition_t * -rd_kafka_topic_partition_list_find (rd_kafka_topic_partition_list_t *rktparlist, - const char *topic, int32_t partition); - - -/** - * @brief Sort list using comparator \p cmp. - * - * If \p cmp is NULL the default comparator will be used that - * sorts by ascending topic name and partition. - * - */ -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 *opaque), - void *opaque); - - -/**@}*/ - - - -/** - * @name Var-arg tag types - * @{ - * - */ - -/** - * @enum rd_kafka_vtype_t - * - * @brief Var-arg tag types - * - * @sa rd_kafka_producev() - */ -typedef enum rd_kafka_vtype_t { - 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 *) Application opaque */ - RD_KAFKA_VTYPE_MSGFLAGS, /**< (int) RD_KAFKA_MSG_F_.. flags */ - RD_KAFKA_VTYPE_TIMESTAMP, /**< (int64_t) Milliseconds since epoch UTC */ -} rd_kafka_vtype_t; - - -/** - * @brief 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_END RD_KAFKA_VTYPE_END - -/*! - * Topic name (const char *) - */ -#define RD_KAFKA_V_TOPIC(topic) \ - _LRK_TYPECHECK(RD_KAFKA_VTYPE_TOPIC, const char *, topic), \ - (const char *)topic -/*! - * Topic object (rd_kafka_topic_t *) - */ -#define RD_KAFKA_V_RKT(rkt) \ - _LRK_TYPECHECK(RD_KAFKA_VTYPE_RKT, rd_kafka_topic_t *, rkt), \ - (rd_kafka_topic_t *)rkt -/*! - * Partition (int32_t) - */ -#define RD_KAFKA_V_PARTITION(partition) \ - _LRK_TYPECHECK(RD_KAFKA_VTYPE_PARTITION, int32_t, partition), \ - (int32_t)partition -/*! - * Message value/payload pointer and length (void *, size_t) - */ -#define RD_KAFKA_V_VALUE(VALUE,LEN) \ - _LRK_TYPECHECK2(RD_KAFKA_VTYPE_VALUE, void *, VALUE, size_t, LEN), \ - (void *)VALUE, (size_t)LEN -/*! - * Message key pointer and length (const void *, size_t) - */ -#define RD_KAFKA_V_KEY(KEY,LEN) \ - _LRK_TYPECHECK2(RD_KAFKA_VTYPE_KEY, const void *, KEY, size_t, LEN), \ - (void *)KEY, (size_t)LEN -/*! - * Opaque pointer (void *) - */ -#define RD_KAFKA_V_OPAQUE(opaque) \ - _LRK_TYPECHECK(RD_KAFKA_VTYPE_OPAQUE, void *, opaque), \ - (void *)opaque -/*! - * Message flags (int) - * @sa RD_KAFKA_MSG_F_COPY, et.al. - */ -#define RD_KAFKA_V_MSGFLAGS(msgflags) \ - _LRK_TYPECHECK(RD_KAFKA_VTYPE_MSGFLAGS, int, msgflags), \ - (int)msgflags -/*! - * Timestamp (int64_t) - */ -#define RD_KAFKA_V_TIMESTAMP(timestamp) \ - _LRK_TYPECHECK(RD_KAFKA_VTYPE_TIMESTAMP, int64_t, timestamp), \ - (int64_t)timestamp - -/**@}*/ - - -/** - * @name Kafka messages - * @{ - * - */ - - - -// FIXME: This doesn't show up in docs for some reason -// "Compound rd_kafka_message_t is not documented." - -/** - * @brief A Kafka message as returned by the \c rd_kafka_consume*() family - * of functions as well as provided to the Producer \c dr_msg_cb(). - * - * For the consumer this object has two purposes: - * - provide the application with a consumed message. (\c err == 0) - * - report per-topic+partition consumer errors (\c err != 0) - * - * The application must check \c err to decide what action to take. - * - * When the application is finished with a message it must call - * rd_kafka_message_destroy() unless otherwise noted. - */ -typedef struct rd_kafka_message_s { - rd_kafka_resp_err_t err; /**< Non-zero for error signaling. */ - rd_kafka_topic_t *rkt; /**< Topic */ - int32_t partition; /**< Partition */ - void *payload; /**< Producer: original message payload. - * Consumer: Depends on the value of \c err : - * - \c err==0: Message payload. - * - \c err!=0: Error string */ - size_t len; /**< Depends on the value of \c err : - * - \c err==0: Message payload length - * - \c err!=0: Error string length */ - void *key; /**< Depends on the value of \c err : - * - \c err==0: Optional message key */ - size_t key_len; /**< Depends on the value of \c err : - * - \c err==0: Optional message key length*/ - int64_t offset; /**< Consume: - * - Message offset (or offset for error - * if \c err!=0 if applicable). - * - dr_msg_cb: - * Message offset assigned by broker. - * If \c produce.offset.report is set then - * each message will have this field set, - * otherwise only the last message in - * each produced internal batch will - * have this field set, otherwise 0. */ - void *_private; /**< Consume: - * - rdkafka private pointer: DO NOT MODIFY - * - dr_msg_cb: - * msg_opaque from produce() call */ -} rd_kafka_message_t; - - -/** - * @brief Frees resources for \p rkmessage and hands ownership back to rdkafka. - */ -RD_EXPORT -void rd_kafka_message_destroy(rd_kafka_message_t *rkmessage); - - - - -/** - * @brief Returns the error string for an errored rd_kafka_message_t or NULL if - * there was no error. - * - * @remark This function MUST NOT be used with the producer. - */ -static RD_INLINE const char * -RD_UNUSED -rd_kafka_message_errstr(const rd_kafka_message_t *rkmessage) { - if (!rkmessage->err) - return NULL; - - if (rkmessage->payload) - return (const char *)rkmessage->payload; - - return rd_kafka_err2str(rkmessage->err); -} - - - -/** - * @brief Returns the message timestamp for a consumed message. - * - * The timestamp is the number of milliseconds since the epoch (UTC). - * - * \p tstype (if not NULL) is updated to indicate the type of timestamp. - * - * @returns message timestamp, or -1 if not available. - * - * @remark Message timestamps require broker version 0.10.0 or later. - */ -RD_EXPORT -int64_t rd_kafka_message_timestamp (const rd_kafka_message_t *rkmessage, - rd_kafka_timestamp_type_t *tstype); - - - -/** - * @brief Returns the latency for a produced message measured from - * the produce() call. - * - * @returns the latency in microseconds, or -1 if not available. - */ -RD_EXPORT -int64_t rd_kafka_message_latency (const rd_kafka_message_t *rkmessage); - - -/**@}*/ - - -/** - * @name Configuration interface - * @{ - * - * @brief Main/global configuration property interface - * - */ - -/** - * @enum rd_kafka_conf_res_t - * @brief Configuration result type - */ -typedef enum { - RD_KAFKA_CONF_UNKNOWN = -2, /**< Unknown configuration name. */ - RD_KAFKA_CONF_INVALID = -1, /**< Invalid configuration value. */ - RD_KAFKA_CONF_OK = 0 /**< Configuration okay */ -} rd_kafka_conf_res_t; - - -/** - * @brief Create configuration object. - * - * When providing your own configuration to the \c 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.: - * @code - * rd_kafka_conf_t *myconf; - * rd_kafka_conf_res_t res; - * - * myconf = rd_kafka_conf_new(); - * res = rd_kafka_conf_set(myconf, "socket.timeout.ms", "600", - * errstr, sizeof(errstr)); - * if (res != RD_KAFKA_CONF_OK) - * die("%s\n", errstr); - * - * rk = rd_kafka_new(..., myconf); - * @endcode - * - * 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. - * - * @returns A new rd_kafka_conf_t object with defaults set. - * - * @sa rd_kafka_conf_set(), rd_kafka_conf_destroy() - */ -RD_EXPORT -rd_kafka_conf_t *rd_kafka_conf_new(void); - - -/** - * @brief Destroys a conf object. - */ -RD_EXPORT -void rd_kafka_conf_destroy(rd_kafka_conf_t *conf); - - -/** - * @brief Creates a copy/duplicate of configuration object \p conf - * - * @remark Interceptors are NOT copied to the new configuration object. - * @sa rd_kafka_interceptor_f_on_conf_dup - */ -RD_EXPORT -rd_kafka_conf_t *rd_kafka_conf_dup(const rd_kafka_conf_t *conf); - - -/** - * @brief Same as rd_kafka_conf_dup() but with an array of property name - * prefixes to filter out (ignore) when copying. - */ -RD_EXPORT -rd_kafka_conf_t *rd_kafka_conf_dup_filter (const rd_kafka_conf_t *conf, - size_t filter_cnt, - const char **filter); - - - -/** - * @brief Sets a configuration property. - * - * \p 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 \c default_topic_conf. - * If no \c 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 \c rd_kafka_conf_res_t to indicate success or failure. - * In case of failure \p errstr is updated to contain a human readable - * error string. - */ -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); - - -/** - * @brief Enable event sourcing. - * \p events is a bitmask of \c RD_KAFKA_EVENT_* of events to enable - * for consumption by `rd_kafka_queue_poll()`. - */ -RD_EXPORT -void rd_kafka_conf_set_events(rd_kafka_conf_t *conf, int events); - - -/** - @deprecated See rd_kafka_conf_set_dr_msg_cb() -*/ -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)); - -/** - * @brief \b Producer: Set delivery report callback in provided \p conf object. - * - * The delivery report callback will be called once for each message - * accepted by rd_kafka_produce() (et.al) with \p 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, or the retry counter for - * temporary errors has been exhausted. - * - * An application must call rd_kafka_poll() at regular intervals to - * serve queued delivery report callbacks. - */ -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)); - - -/** - * @brief \b Consumer: Set consume callback for use with rd_kafka_consumer_poll() - * - */ -RD_EXPORT -void rd_kafka_conf_set_consume_cb (rd_kafka_conf_t *conf, - void (*consume_cb) (rd_kafka_message_t * - rkmessage, - void *opaque)); - -/** - * @brief \b Consumer: Set rebalance callback for use with - * coordinated consumer group balancing. - * - * The \p 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 \p rebalance_cb turns off librdkafka's automatic - * partition assignment/revocation and instead delegates that responsibility - * to the application's \p 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 \p err is neither of those. - * @remark In this latter case (arbitrary error), the application must - * call rd_kafka_assign(rk, NULL) to synchronize state. - * - * 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 assinging/revocation, - * such as fetching offsets from an alternate location (on assign) - * or manually committing offsets (on revoke). - * - * @remark The \p partitions list is destroyed by librdkafka on return - * return from the rebalance_cb and must not be freed or - * saved by the application. - * - * The following example shows the application's responsibilities: - * @code - * static void rebalance_cb (rd_kafka_t *rk, rd_kafka_resp_err_t err, - * rd_kafka_topic_partition_list_t *partitions, - * void *opaque) { - * - * switch (err) - * { - * case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS: - * // application may load offets from arbitrary external - * // storage here and update \p partitions - * - * rd_kafka_assign(rk, partitions); - * break; - * - * case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS: - * if (manual_commits) // Optional explicit manual commit - * rd_kafka_commit(rk, partitions, 0); // sync commit - * - * rd_kafka_assign(rk, NULL); - * break; - * - * default: - * handle_unlikely_error(err); - * rd_kafka_assign(rk, NULL); // sync state - * break; - * } - * } - * @endcode - */ -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)); - - - -/** - * @brief \b 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 \p err == RD_KAFKA_RESP_ERR__NO_OFFSET which is not to be considered - * an error. - * - * The \p offsets list contains per-partition information: - * - \c offset: committed offset (attempted) - * - \c err: commit error - */ -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)); - - -/** - * @brief Set error callback in provided conf object. - * - * The error callback is used by librdkafka to signal critical errors - * back to the application. - * - * If no \p error_cb is registered then the errors will be logged instead. - */ -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)); - -/** - * @brief 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. - * - * @remark Requires broker version 0.9.0 or later. - */ -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)); - - -/** - * @brief 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 \p func as NULL to disable logging. - * - * This is the configuration alternative to the deprecated rd_kafka_set_logger() - * - * @remark The log_cb will be called spontaneously from librdkafka's internal - * threads unless logs have been forwarded to a poll queue through - * \c rd_kafka_set_log_queue(). - * An application MUST NOT call any librdkafka APIs or do any prolonged - * work in a non-forwarded \c log_cb. - */ -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)); - - -/** - * @brief Set statistics callback in provided conf object. - * - * The statistics callback is triggered from rd_kafka_poll() every - * \c statistics.interval.ms (needs to be configured separately). - * Function arguments: - * - \p rk - Kafka handle - * - \p json - String containing the statistics data in JSON format - * - \p json_len - Length of \p json string. - * - \p opaque - Application-provided opaque. - * - * If the application wishes to hold on to the \p json pointer and free - * it at a later time it must return 1 from the \p stats_cb. - * If the application returns 0 from the \p stats_cb then librdkafka - * will immediately free the \p json pointer. - */ -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)); - - - -/** - * @brief Set socket callback. - * - * The socket callback is responsible for opening a socket - * according to the supplied \p domain, \p type and \p protocol. - * The socket shall be created with \c CLOEXEC set in a racefree fashion, if - * possible. - * - * Default: - * - on linux: racefree CLOEXEC - * - others : non-racefree CLOEXEC - * - * @remark The callback will be called from an internal librdkafka thread. - */ -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)); - - - -/** - * @brief Set connect callback. - * - * The connect callback is responsible for connecting socket \p sockfd - * to peer address \p addr. - * The \p id field contains the broker identifier. - * - * \p connect_cb shall return 0 on success (socket connected) or an error - * number (errno) on error. - * - * @remark 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 (*connect_cb) (int sockfd, - const struct sockaddr *addr, - int addrlen, - const char *id, - void *opaque)); - -/** - * @brief Set close socket callback. - * - * Close a socket (optionally opened with socket_cb()). - * - * @remark 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 (*closesocket_cb) (int sockfd, - void *opaque)); - - - -#ifndef _MSC_VER -/** - * @brief Set open callback. - * - * The open callback is responsible for opening the file specified by - * pathname, flags and mode. - * The file shall be opened with \c CLOEXEC set in a racefree fashion, if - * possible. - * - * Default: - * - on linux: racefree CLOEXEC - * - others : non-racefree CLOEXEC - * - * @remark 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 (*open_cb) (const char *pathname, - int flags, mode_t mode, - void *opaque)); -#endif - -/** - * @brief Sets the application's opaque pointer that will be passed to callbacks - */ -RD_EXPORT -void rd_kafka_conf_set_opaque(rd_kafka_conf_t *conf, void *opaque); - -/** - * @brief Retrieves the opaque pointer previously set with rd_kafka_conf_set_opaque() - */ -RD_EXPORT -void *rd_kafka_opaque(const rd_kafka_t *rk); - - - -/** - * 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. - */ -RD_EXPORT -void rd_kafka_conf_set_default_topic_conf (rd_kafka_conf_t *conf, - rd_kafka_topic_conf_t *tconf); - - - -/** - * @brief Retrieve configuration value for property \p name. - * - * If \p dest is non-NULL the value will be written to \p dest with at - * most \p dest_size. - * - * \p *dest_size is updated to the full length of the value, thus if - * \p *dest_size initially is smaller than the full length the application - * may reallocate \p dest to fit the returned \p *dest_size and try again. - * - * If \p dest is NULL only the full length of the value is returned. - * - * Fallthrough: - * Topic-level configuration properties from the \c default_topic_conf - * may be retrieved using this interface. - * - * @returns \p RD_KAFKA_CONF_OK if the property name matched, else - * \p RD_KAFKA_CONF_UNKNOWN. - */ -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); - - -/** - * @brief Retrieve topic configuration value for property \p name. - * - * @sa rd_kafka_conf_get() - */ -RD_EXPORT -rd_kafka_conf_res_t rd_kafka_topic_conf_get (const rd_kafka_topic_conf_t *conf, - const char *name, - char *dest, size_t *dest_size); - - -/** - * @brief Dump the configuration properties and values of \p conf to an array - * with \"key\", \"value\" pairs. - * - * The number of entries in the array is returned in \p *cntp. - * - * The dump must be freed with `rd_kafka_conf_dump_free()`. - */ -RD_EXPORT -const char **rd_kafka_conf_dump(rd_kafka_conf_t *conf, size_t *cntp); - - -/** - * @brief Dump the topic configuration properties and values of \p conf - * to an array with \"key\", \"value\" pairs. - * - * The number of entries in the array is returned in \p *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); - -/** - * @brief Frees a configuration dump returned from `rd_kafka_conf_dump()` or - * `rd_kafka_topic_conf_dump(). - */ -RD_EXPORT -void rd_kafka_conf_dump_free(const char **arr, size_t cnt); - -/** - * @brief Prints a table to \p fp of all supported configuration properties, - * their default values as well as a description. - */ -RD_EXPORT -void rd_kafka_conf_properties_show(FILE *fp); - -/**@}*/ - - -/** - * @name Topic configuration - * @{ - * - * @brief Topic configuration property interface - * - */ - - -/** - * @brief Create topic configuration object - * - * @sa Same semantics as for rd_kafka_conf_new(). - */ -RD_EXPORT -rd_kafka_topic_conf_t *rd_kafka_topic_conf_new(void); - - -/** - * @brief Creates a copy/duplicate of topic configuration object \p conf. - */ -RD_EXPORT -rd_kafka_topic_conf_t *rd_kafka_topic_conf_dup(const rd_kafka_topic_conf_t - *conf); - - -/** - * @brief Destroys a topic conf object. - */ -RD_EXPORT -void rd_kafka_topic_conf_destroy(rd_kafka_topic_conf_t *topic_conf); - - -/** - * @brief Sets a single rd_kafka_topic_conf_t value by property name. - * - * \p 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 -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); - -/** - * @brief Sets the application's opaque pointer that will be passed to all topic - * callbacks as the \c rkt_opaque argument. - */ -RD_EXPORT -void rd_kafka_topic_conf_set_opaque(rd_kafka_topic_conf_t *conf, void *opaque); - - -/** - * @brief \b 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. - * - * 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 \c RD_KAFKA_PARTITION_UA value if partitioning - * could not be performed. - */ -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)); - -/** - * @brief 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 -int rd_kafka_topic_partition_available(const rd_kafka_topic_t *rkt, - int32_t partition); - - -/******************************************************************* - * * - * Partitioners provided by rdkafka * - * * - *******************************************************************/ - -/** - * @brief Random partitioner. - * - * Will try not to return unavailable partitions. - * - * @returns a random partition between 0 and \p partition_cnt - 1. - * - */ -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 *opaque, void *msg_opaque); - -/** - * @brief Consistent partitioner. - * - * Uses consistent hashing to map identical keys onto identical partitions. - * - * @returns a \"random\" partition between 0 and \p partition_cnt - 1 based on - * the CRC value of the key - */ -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 *opaque, void *msg_opaque); - -/** - * @brief 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. - * - * @returns a \"random\" partition between 0 and \p partition_cnt - 1 based on - * the CRC value of the key (if provided) - */ -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 *opaque, void *msg_opaque); - - -/**@}*/ - - - -/** - * @name Main Kafka and Topic object handles - * @{ - * - * - */ - - - - -/** - * @brief Creates a new Kafka handle and starts its operation according to the - * specified \p type (\p RD_KAFKA_CONSUMER or \p RD_KAFKA_PRODUCER). - * - * \p conf is an optional struct created with `rd_kafka_conf_new()` that will - * be used instead of the default configuration. - * The \p 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. - * - * \p errstr must be a pointer to memory of at least size \p 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. - * - * @remark \b RD_KAFKA_CONSUMER: When a new \p 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. - * @remark 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 \p errstr) - * - * @sa To destroy the Kafka handle, use rd_kafka_destroy(). - */ -RD_EXPORT -rd_kafka_t *rd_kafka_new(rd_kafka_type_t type, rd_kafka_conf_t *conf, - char *errstr, size_t errstr_size); - - -/** - * @brief Destroy Kafka handle. - * - * @remark This is a blocking operation. - */ -RD_EXPORT -void rd_kafka_destroy(rd_kafka_t *rk); - - - -/** - * @brief Returns Kafka handle name. - */ -RD_EXPORT -const char *rd_kafka_name(const rd_kafka_t *rk); - - -/** - * @brief Returns Kafka handle type. - */ -RD_EXPORT -rd_kafka_type_t rd_kafka_type(const rd_kafka_t *rk); - - -/** - * @brief Returns this client's broker-assigned group member id - * - * @remark 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 \p free() or - * rd_kafka_mem_free() - */ -RD_EXPORT -char *rd_kafka_memberid (const rd_kafka_t *rk); - - - -/** - * @brief Returns the ClusterId as reported in broker metadata. - * - * @param timeout_ms If 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. - - * @remark Requires broker version >=0.10.0 and api.version.request=true. - * - * @remark 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 -char *rd_kafka_clusterid (rd_kafka_t *rk, int timeout_ms); - - -/** - * @brief Creates a new topic handle for topic named \p topic. - * - * \p 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 \p 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 \p errno to an rd_kafka_resp_err_t error code. - * - * @sa rd_kafka_topic_destroy() - */ -RD_EXPORT -rd_kafka_topic_t *rd_kafka_topic_new(rd_kafka_t *rk, const char *topic, - rd_kafka_topic_conf_t *conf); - - - -/** - * @brief Loose application's topic handle refcount as previously created - * with `rd_kafka_topic_new()`. - * - * @remark 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 -void rd_kafka_topic_destroy(rd_kafka_topic_t *rkt); - - -/** - * @brief Returns the topic name. - */ -RD_EXPORT -const char *rd_kafka_topic_name(const rd_kafka_topic_t *rkt); - - -/** - * @brief Get the \p rkt_opaque pointer that was set in the topic configuration. - */ -RD_EXPORT -void *rd_kafka_topic_opaque (const rd_kafka_topic_t *rkt); - - -/** - * @brief 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_PARTITION_UA ((int32_t)-1) - - -/** - * @brief Polls the provided kafka handle for events. - * - * Events will cause application provided callbacks to be called. - * - * The \p 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 \p timeout_ms. - * To wait indefinately for an event, provide -1. - * - * @remark An application should make sure to call poll() at regular - * intervals to serve any queued callbacks waiting to be called. - * - * Events: - * - delivery report callbacks (if dr_cb/dr_msg_cb is configured) [producer] - * - error callbacks (rd_kafka_conf_set_error_cb()) [all] - * - stats callbacks (rd_kafka_conf_set_stats_cb()) [all] - * - throttle callbacks (rd_kafka_conf_set_throttle_cb()) [all] - * - * @returns the number of events served. - */ -RD_EXPORT -int rd_kafka_poll(rd_kafka_t *rk, int timeout_ms); - - -/** - * @brief 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. - * - * @remark This function MUST ONLY be called from within a librdkafka callback. - */ -RD_EXPORT -void rd_kafka_yield (rd_kafka_t *rk); - - - - -/** - * @brief Pause producing or consumption for the provided list of partitions. - * - * Success or error is returned per-partition \p err in the \p partitions list. - * - * @returns RD_KAFKA_RESP_ERR_NO_ERROR - */ -RD_EXPORT rd_kafka_resp_err_t -rd_kafka_pause_partitions (rd_kafka_t *rk, - rd_kafka_topic_partition_list_t *partitions); - - - -/** - * @brief Resume producing consumption for the provided list of partitions. - * - * Success or error is returned per-partition \p err in the \p 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); - - - - -/** - * @brief Query broker for low (oldest/beginning) and high (newest/end) offsets - * for partition. - * - * Offsets are returned in \p *low and \p *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_query_watermark_offsets (rd_kafka_t *rk, - const char *topic, int32_t partition, - int64_t *low, int64_t *high, int timeout_ms); - - -/** - * @brief 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 \p *low and \p *high respectively. - * - * @returns RD_KAFKA_RESP_ERR_NO_ERROR on success or an error code on failure. - * - * @remark Shall only be used with an active consumer instance. - */ -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); - - - -/** - * @brief 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 \c offset in \p offsets - * on input, and \c offset will contain the offset on output. - * - * The function will block for at most \p timeout_ms milliseconds. - * - * @remark Duplicate Topic+Partitions are not supported. - * @remark Per-partition errors may be returned in \c rd_kafka_topic_partition_t.err - * - * @returns an error code for general errors, else RD_KAFKA_RESP_ERR_NO_ERROR - * in which case per-partition errors might be set. - */ -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); - - -/** - * @brief 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. - * - * @remark rd_kafka_mem_free() must only be used for pointers returned by APIs - * that explicitly mention using this function for freeing. - */ -RD_EXPORT -void rd_kafka_mem_free (rd_kafka_t *rk, void *ptr); - - -/**@}*/ - - - - - -/** - * @name 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. - */ - - -/** - * @brief Create a new message queue. - * - * See rd_kafka_consume_start_queue(), rd_kafka_consume_queue(), et.al. - */ -RD_EXPORT -rd_kafka_queue_t *rd_kafka_queue_new(rd_kafka_t *rk); - -/** - * Destroy a queue, purging all of its enqueued messages. - */ -RD_EXPORT -void rd_kafka_queue_destroy(rd_kafka_queue_t *rkqu); - - -/** - * @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_main (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. - * - * @remark 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_consumer (rd_kafka_t *rk); - -/** - * @returns a reference to the partition's queue, or NULL if - * partition is invalid. - * - * Use rd_kafka_queue_destroy() to loose the reference. - * - * @remark rd_kafka_queue_destroy() MUST be called on this queue - * - * @remark This function only works on consumers. - */ -RD_EXPORT -rd_kafka_queue_t *rd_kafka_queue_get_partition (rd_kafka_t *rk, - const char *topic, - int32_t partition); - -/** - * @brief Forward/re-route queue \p src to \p dst. - * If \p dst is \c NULL the forwarding is removed. - * - * The internal refcounts for both queues are increased. - * - * @remark Regardless of whether \p dst is NULL or not, after calling this - * function, \p src will not forward it's fetch queue to the consumer - * queue. - */ -RD_EXPORT -void rd_kafka_queue_forward (rd_kafka_queue_t *src, rd_kafka_queue_t *dst); - -/** - * @brief 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 (\c log_cb) - * in its thread of choice. - * - * @param rkqu Queue to forward logs to. If the value is NULL the logs - * are forwarded to the main queue. - * - * @remark The configuration property \c log.queue MUST also be set to true. - * - * @remark librdkafka maintains its own reference to the provided queue. - * - * @returns RD_KAFKA_RESP_ERR_NO_ERROR on success or an error code on error. - */ -RD_EXPORT -rd_kafka_resp_err_t rd_kafka_set_log_queue (rd_kafka_t *rk, - rd_kafka_queue_t *rkqu); - - -/** - * @returns the current number of elements in queue. - */ -RD_EXPORT -size_t rd_kafka_queue_length (rd_kafka_queue_t *rkqu); - - -/** - * @brief 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 \p payload (of size \p size) to - * whenever a new element is enqueued on a previously empty queue. - * - * To remove event triggering call with \p fd = -1. - * - * librdkafka will maintain a copy of the \p payload. - * - * @remark When using forwarded queues the IO event must only be enabled - * on the final forwarded-to (destination) queue. - */ -RD_EXPORT -void rd_kafka_queue_io_event_enable (rd_kafka_queue_t *rkqu, int fd, - const void *payload, size_t size); - -/**@}*/ - -/** - * - * @name Simple Consumer API (legacy) - * @{ - * - */ - - -#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 */ - - -/** @cond NO_DOC */ -#define RD_KAFKA_OFFSET_TAIL_BASE -2000 /* internal: do not use */ -/** @endcond */ - -/** - * @brief Start consuming \p CNT messages from topic's current end offset. - * - * That is, if current end offset is 12345 and \p CNT is 200, it will start - * consuming from offset \c 12345-200 = \c 12145. */ -#define RD_KAFKA_OFFSET_TAIL(CNT) (RD_KAFKA_OFFSET_TAIL_BASE - (CNT)) - -/** - * @brief Start consuming messages for topic \p rkt and \p partition - * at offset \p offset which may either be an absolute \c (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 \c 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 \p 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 \c errno to `rd_kafka_resp_err_t` - */ -RD_EXPORT -int rd_kafka_consume_start(rd_kafka_topic_t *rkt, int32_t partition, - int64_t offset); - -/** - * @brief Same as rd_kafka_consume_start() but re-routes incoming messages to - * the provided queue \p 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_start_queue(rd_kafka_topic_t *rkt, int32_t partition, - int64_t offset, rd_kafka_queue_t *rkqu); - -/** - * @brief Stop consuming messages for topic \p rkt and \p 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 -int rd_kafka_consume_stop(rd_kafka_topic_t *rkt, int32_t partition); - - - -/** - * @brief Seek consumer for topic+partition to \p offset which is either an - * absolute or logical offset. - * - * If \p timeout_ms is not 0 the call will wait this long for the - * seek to be performed. If the timeout is reached the internal state - * will be unknown and this function returns `RD_KAFKA_RESP_ERR__TIMED_OUT`. - * If \p timeout_ms is 0 it will initiate the seek but return - * immediately without any error reporting (e.g., async). - * - * This call triggers a fetch queue barrier flush. - * - * @returns `RD_KAFKA_RESP_ERR__NO_ERROR` on success else an error code. - */ -RD_EXPORT -rd_kafka_resp_err_t rd_kafka_seek (rd_kafka_topic_t *rkt, - int32_t partition, - int64_t offset, - int timeout_ms); - - -/** - * @brief Consume a single message from topic \p rkt and \p partition - * - * \p 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 \c 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 - \p timeout_ms was reached with no new messages fetched. - * - ENOENT - \p rkt + \p partition is unknown. - * (no prior `rd_kafka_consume_start()` call) - * - * NOTE: The returned message's \c ..->err must be checked for errors. - * NOTE: \c ..->err \c == \c 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). - * - * @remark on_consume() interceptors may be called from this function prior to - * passing message to application. - */ -RD_EXPORT -rd_kafka_message_t *rd_kafka_consume(rd_kafka_topic_t *rkt, int32_t partition, - int timeout_ms); - - - -/** - * @brief Consume up to \p rkmessages_size from topic \p rkt and \p partition - * putting a pointer to each message in the application provided - * array \p rkmessages (of size \p rkmessages_size entries). - * - * `rd_kafka_consume_batch()` provides higher throughput performance - * than `rd_kafka_consume()`. - * - * \p timeout_ms is the maximum amount of time to wait for all of - * \p rkmessages_size messages to be put into \p rkmessages. - * If no messages were available within the timeout period this function - * returns 0 and \p 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 \p rkmessages, - * or -1 on error (same error codes as for `rd_kafka_consume()`. - * - * @sa rd_kafka_consume() - * - * @remark 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); - - - -/** - * @brief Consumes messages from topic \p rkt and \p 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()`. - * - * \p timeout_ms is the maximum amount of time to wait for one or more messages - * to arrive. - * - * The provided \p consume_cb function is called for each message, - * the application \b MUST \b NOT call `rd_kafka_message_destroy()` on the - * provided \p rkmessage. - * - * The \p opaque argument is passed to the 'consume_cb' as \p opaque. - * - * @returns the number of messages processed or -1 on error. - * - * @sa rd_kafka_consume() - * - * @remark 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 (*consume_cb) (rd_kafka_message_t - *rkmessage, - void *opaque), - void *opaque); - - -/** - * @name Simple Consumer API (legacy): Queue consumers - * @{ - * - * The following `..._queue()` functions are analogue to the functions above - * but reads messages from the provided queue \p rkqu instead. - * \p 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. - */ - -/** - * @brief Consume from queue - * - * @sa rd_kafka_consume() - */ -RD_EXPORT -rd_kafka_message_t *rd_kafka_consume_queue(rd_kafka_queue_t *rkqu, - int timeout_ms); - -/** - * @brief Consume batch of messages from queue - * - * @sa rd_kafka_consume_batch() - */ -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); - -/** - * @brief Consume multiple messages from queue with callback - * - * @sa rd_kafka_consume_callback() - */ -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 *opaque), - void *opaque); - - -/**@}*/ - - - - -/** - * @name Simple Consumer API (legacy): Topic+partition offset store. - * @{ - * - * If \c 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. - */ - - -/** - * @brief Store offset \p offset for topic \p rkt partition \p partition. - * - * The offset will be committed (written) to the offset store according - * to \c `auto.commit.interval.ms` or manual offset-less commit() - * - * @remark \c `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_offset_store(rd_kafka_topic_t *rkt, - int32_t partition, int64_t offset); - - -/** - * @brief Store offsets for one or more partitions. - * - * The offset will be committed (written) to the offset store according - * to \c `auto.commit.interval.ms` or manual offset-less commit(). - * - * Per-partition success/error status propagated through each partition's - * \c .err field. - * - * @remark \c `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 if - * none of the offsets could be stored. - */ -RD_EXPORT rd_kafka_resp_err_t -rd_kafka_offsets_store(rd_kafka_t *rk, - rd_kafka_topic_partition_list_t *offsets); -/**@}*/ - - - - -/** - * @name KafkaConsumer (C) - * @{ - * @brief High-level KafkaConsumer C API - * - * - * - */ - -/** - * @brief Subscribe to topic set using balanced consumer groups. - * - * Wildcard (regex) topics are supported by the librdkafka assignor: - * any topic name in the \p topics list that is prefixed with \c \"^\" will - * be regex-matched to the full list of topics in the cluster and matching - * topics will be added to the subscription list. - * - * @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. - */ -RD_EXPORT rd_kafka_resp_err_t -rd_kafka_subscribe (rd_kafka_t *rk, - const rd_kafka_topic_partition_list_t *topics); - - -/** - * @brief Unsubscribe from the current subscription set. - */ -RD_EXPORT -rd_kafka_resp_err_t rd_kafka_unsubscribe (rd_kafka_t *rk); - - -/** - * @brief Returns the current topic subscription - * - * @returns An error code on failure, otherwise \p topic is updated - * to point to a newly allocated topic list (possibly empty). - * - * @remark The application is responsible for calling - * rd_kafka_topic_partition_list_destroy on the returned list. - */ -RD_EXPORT rd_kafka_resp_err_t -rd_kafka_subscription (rd_kafka_t *rk, - rd_kafka_topic_partition_list_t **topics); - - - -/** - * @brief Poll the consumer for messages or events. - * - * Will block for at most \p timeout_ms milliseconds. - * - * @remark 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 \p ->err is - * RD_KAFKA_RESP_ERR_NO_ERROR, or an event or error for any other - * value. - * - * @remark on_consume() interceptors may be called from this function prior to - * passing message to application. - * - * @sa rd_kafka_message_t - */ -RD_EXPORT -rd_kafka_message_t *rd_kafka_consumer_poll (rd_kafka_t *rk, int timeout_ms); - -/** - * @brief Close down the KafkaConsumer. - * - * @remark This call will block until the consumer has revoked its assignment, - * calling the \c rebalance_cb if it is configured, committed offsets - * to broker, and left the consumer group. - * The maximum blocking time is roughly limited to session.timeout.ms. - * - * @returns An error code indicating if the consumer close was succesful - * or not. - * - * @remark The application still needs to call rd_kafka_destroy() after - * this call finishes to clean up the underlying handle resources. - * - */ -RD_EXPORT -rd_kafka_resp_err_t rd_kafka_consumer_close (rd_kafka_t *rk); - - - -/** - * @brief Atomic assignment of partitions to consume. - * - * The new \p partitions will replace the existing assignment. - * - * When used from a rebalance callback the application shall pass the - * partition list passed to the callback (or a copy of it) (even if the list - * is empty) rather than NULL to maintain internal join state. - - * A zero-length \p partitions will treat the partitions as a valid, - * albeit empty, assignment, and maintain internal state, while a \c NULL - * value for \p partitions will reset and clear the internal state. - */ -RD_EXPORT rd_kafka_resp_err_t -rd_kafka_assign (rd_kafka_t *rk, - const rd_kafka_topic_partition_list_t *partitions); - -/** - * @brief Returns the current partition assignment - * - * @returns An error code on failure, otherwise \p partitions is updated - * to point to a newly allocated partition list (possibly empty). - * - * @remark The application is responsible for calling - * rd_kafka_topic_partition_list_destroy on the returned list. - */ -RD_EXPORT rd_kafka_resp_err_t -rd_kafka_assignment (rd_kafka_t *rk, - rd_kafka_topic_partition_list_t **partitions); - - - - -/** - * @brief Commit offsets on broker for the provided list of partitions. - * - * \p offsets should contain \c topic, \c partition, \c offset and possibly - * \c metadata. - * If \p offsets is NULL the current partition assignment will be used instead. - * - * If \p 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. - */ -RD_EXPORT rd_kafka_resp_err_t -rd_kafka_commit (rd_kafka_t *rk, const rd_kafka_topic_partition_list_t *offsets, - int async); - - -/** - * @brief Commit message's offset on broker for the message's partition. - * - * @sa rd_kafka_commit - */ -RD_EXPORT rd_kafka_resp_err_t -rd_kafka_commit_message (rd_kafka_t *rk, const rd_kafka_message_t *rkmessage, - int async); - - -/** - * @brief Commit offsets on broker for the provided list of partitions. - * - * See rd_kafka_commit for \p offsets semantics. - * - * The result of the offset commit will be posted on the provided \p 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 \p cb callback is required. \p opaque is passed to the callback. - * - * If using the event API the callback is ignored and the offset commit result - * will be returned as an RD_KAFKA_EVENT_COMMIT event. The \p opaque - * value will be available with rd_kafka_event_opaque() - * - * If \p rkqu is NULL a temporary queue will be created and the callback will - * be served by this call. - * - * @sa rd_kafka_commit() - * @sa rd_kafka_conf_set_offset_commit_cb() - */ -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 *opaque), - void *opaque); - - -/** - * @brief Retrieve committed offsets for topics+partitions. - * - * The \p 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. - * - * @returns RD_KAFKA_RESP_ERR_NO_ERROR on success in which case the - * \p offset or \p err field of each \p 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_committed (rd_kafka_t *rk, - rd_kafka_topic_partition_list_t *partitions, - int timeout_ms); - - - -/** - * @brief Retrieve current positions (offsets) for topics+partitions. - * - * The \p 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. - * - * @returns RD_KAFKA_RESP_ERR_NO_ERROR on success in which case the - * \p offset or \p err field of each \p 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); - - -/**@}*/ - - - -/** - * @name Producer API - * @{ - * - * - */ - - -/** - * @brief Producer message flags - */ -#define RD_KAFKA_MSG_F_FREE 0x1 /**< 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. - */ - - - -/** - * @brief Produce and send a single message to broker. - * - * \p 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. - * - * \p 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) - * - * \p msgflags is zero or more of the following flags OR:ed together: - * RD_KAFKA_MSG_F_BLOCK - block \p produce*() call if - * \p queue.buffering.max.messages or - * \p 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 \c RD_KAFKA_MSG_F_BLOCK above. - * - * RD_KAFKA_MSG_F_FREE - rdkafka will free(3) \p payload when it is done - * with it. - * RD_KAFKA_MSG_F_COPY - the \p payload data will be copied and the - * \p payload pointer will not be used by rdkafka - * after the call returns. - * - * .._F_FREE and .._F_COPY are mutually exclusive. - * - * 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. - * - * \p payload is the message payload of size \p len bytes. - * - * \p key is an optional message key of size \p keylen bytes, if non-NULL it - * will be passed to the topic partitioner as well as be s
<TRUNCATED>
