rdkafka.h

Go to the documentation of this file.

/*
 * librdkafka - Apache Kafka C library
 *
 * Copyright (c) 2012-2020 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.
 */
 
/* @cond NO_DOC */
#ifndef _RDKAFKA_H_
#define _RDKAFKA_H_
 
#include <stdio.h>
#include <inttypes.h>
#include <sys/types.h>
 
#ifdef __cplusplus
extern "C" {
#if 0
} /* Restore indent */
#endif
#endif
 
#ifdef _WIN32
#include <basetsd.h>
#ifndef WIN32_MEAN_AND_LEAN
#define WIN32_MEAN_AND_LEAN
#endif
#include <winsock2.h>  /* for sockaddr, .. */
#ifndef _SSIZE_T_DEFINED
#define _SSIZE_T_DEFINED
typedef SSIZE_T ssize_t;
#endif
#define RD_UNUSED
#define RD_INLINE __inline
#define RD_DEPRECATED __declspec(deprecated)
#define RD_FORMAT(...)
#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))
 
#if defined(__clang__) || defined(__GNUC__) || defined(__GNUG__)
#define RD_FORMAT(...) __attribute__((format (__VA_ARGS__)))
#else
#define RD_FORMAT(...)
#endif
 
#ifndef LIBRDKAFKA_TYPECHECKS
#define LIBRDKAFKA_TYPECHECKS 1
#endif
#endif
 
 
#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; })
 
#define _LRK_TYPECHECK3(RET,TYPE,ARG,TYPE2,ARG2,TYPE3,ARG3) \
        ({                                              \
                if (0) {                                \
                        TYPE __t RD_UNUSED = (ARG);     \
                        TYPE2 __t2 RD_UNUSED = (ARG2);  \
                        TYPE3 __t3 RD_UNUSED = (ARG3);  \
                }                                       \
                RET; })
#else
#define _LRK_TYPECHECK(RET,TYPE,ARG)  (RET)
#define _LRK_TYPECHECK2(RET,TYPE,ARG,TYPE2,ARG2) (RET)
#define _LRK_TYPECHECK3(RET,TYPE,ARG,TYPE2,ARG2,TYPE3,ARG3) (RET)
#endif
 
/* @endcond */
 
 
#define RD_KAFKA_VERSION  0x010802ff
 
RD_EXPORT
int rd_kafka_version(void);
 
RD_EXPORT
const char *rd_kafka_version_str (void);
 
typedef enum rd_kafka_type_t {
    RD_KAFKA_PRODUCER, 
    RD_KAFKA_CONSUMER  
} rd_kafka_type_t;
 
 
typedef enum rd_kafka_timestamp_type_t {
    RD_KAFKA_TIMESTAMP_NOT_AVAILABLE,   
    RD_KAFKA_TIMESTAMP_CREATE_TIME,     
    RD_KAFKA_TIMESTAMP_LOG_APPEND_TIME  
} rd_kafka_timestamp_type_t;
 
 
 
RD_EXPORT
const char *rd_kafka_get_debug_contexts(void);
 
#define RD_KAFKA_DEBUG_CONTEXTS \
        "all,generic,broker,topic,metadata,feature,queue,msg,protocol,cgrp,security,fetch,interceptor,plugin,consumer,admin,eos,mock,assignor,conf"
 
 
/* @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;
typedef struct rd_kafka_op_s rd_kafka_event_t;
typedef struct rd_kafka_topic_result_s rd_kafka_topic_result_t;
typedef struct rd_kafka_consumer_group_metadata_s
rd_kafka_consumer_group_metadata_t;
typedef struct rd_kafka_error_s rd_kafka_error_t;
typedef struct rd_kafka_headers_s rd_kafka_headers_t;
typedef struct rd_kafka_group_result_s rd_kafka_group_result_t;
/* @endcond */
 
 
typedef enum {
    /* Internal errors to rdkafka: */
    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,
 
    /* Kafka broker errors: */
    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,
#define RD_KAFKA_RESP_ERR_GROUP_LOAD_IN_PROGRESS        \
        RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS
        RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE = 15,
#define RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE       \
        RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE
        RD_KAFKA_RESP_ERR_NOT_COORDINATOR = 16,
#define RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP     \
        RD_KAFKA_RESP_ERR_NOT_COORDINATOR
        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,
} rd_kafka_resp_err_t;
 
 
struct rd_kafka_err_desc {
    rd_kafka_resp_err_t code;
    const char *name;      
    const char *desc;      
};
 
 
RD_EXPORT
void rd_kafka_get_err_descs (const struct rd_kafka_err_desc **errdescs,
                 size_t *cntp);
 
 
 
 
RD_EXPORT
const char *rd_kafka_err2str (rd_kafka_resp_err_t err);
 
 
 
RD_EXPORT
const char *rd_kafka_err2name (rd_kafka_resp_err_t err);
 
 
RD_EXPORT
rd_kafka_resp_err_t rd_kafka_last_error (void);
 
 
RD_EXPORT RD_DEPRECATED
rd_kafka_resp_err_t rd_kafka_errno2err(int errnox);
 
 
RD_EXPORT RD_DEPRECATED
int rd_kafka_errno (void);
 
 
 
 
RD_EXPORT
rd_kafka_resp_err_t rd_kafka_fatal_error (rd_kafka_t *rk,
                                          char *errstr, size_t errstr_size);
 
 
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);
 
 
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);
 
 
RD_EXPORT
rd_kafka_error_t *rd_kafka_error_new (rd_kafka_resp_err_t code,
                                      const char *fmt, ...)
        RD_FORMAT(printf, 2, 3);
 
 
typedef struct rd_kafka_topic_partition_s {
        char        *topic;             
        int32_t      partition;         
    int64_t      offset;            
        void        *metadata;          
        size_t       metadata_size;     
        void        *opaque;            
        rd_kafka_resp_err_t err;        
        void       *_private;           
} rd_kafka_topic_partition_t;
 
 
RD_EXPORT
void rd_kafka_topic_partition_destroy (rd_kafka_topic_partition_t *rktpar);
 
 
typedef struct rd_kafka_topic_partition_list_s {
        int cnt;               
        int size;              
        rd_kafka_topic_partition_t *elems; 
} rd_kafka_topic_partition_list_t;
 
 
RD_EXPORT
rd_kafka_topic_partition_list_t *rd_kafka_topic_partition_list_new (int size);
 
 
RD_EXPORT
void
rd_kafka_topic_partition_list_destroy (rd_kafka_topic_partition_list_t *rkparlist);
 
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);
 
 
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);
 
 
 
RD_EXPORT
int
rd_kafka_topic_partition_list_del (rd_kafka_topic_partition_list_t *rktparlist,
                   const char *topic, int32_t partition);
 
 
RD_EXPORT
int
rd_kafka_topic_partition_list_del_by_idx (
    rd_kafka_topic_partition_list_t *rktparlist,
    int idx);
 
 
RD_EXPORT
rd_kafka_topic_partition_list_t *
rd_kafka_topic_partition_list_copy (const rd_kafka_topic_partition_list_t *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);
 
 
 
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);
 
 
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);
 
 
typedef 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,   
} rd_kafka_vtype_t;
 
 
typedef struct rd_kafka_vu_s {
        rd_kafka_vtype_t vtype;           
        union {
                const char *cstr;
                rd_kafka_topic_t *rkt;
                int i;
                int32_t i32;
                int64_t i64;
                struct {
                        void *ptr;
                        size_t size;
                } mem;
                struct {
                        const char *name;
                        const void *val;
                        ssize_t size;
                } header;
                rd_kafka_headers_t *headers;
                void *ptr;
                char _pad[64];  
        } u;
} rd_kafka_vu_t;
 
#define RD_KAFKA_V_END RD_KAFKA_VTYPE_END
 
#define RD_KAFKA_V_TOPIC(topic)                                         \
        _LRK_TYPECHECK(RD_KAFKA_VTYPE_TOPIC, const char *, topic),      \
        (const char *)topic
#define RD_KAFKA_V_RKT(rkt)                                             \
        _LRK_TYPECHECK(RD_KAFKA_VTYPE_RKT, rd_kafka_topic_t *, rkt),    \
        (rd_kafka_topic_t *)rkt
#define RD_KAFKA_V_PARTITION(partition)                                 \
        _LRK_TYPECHECK(RD_KAFKA_VTYPE_PARTITION, int32_t, partition),   \
        (int32_t)partition
#define RD_KAFKA_V_VALUE(VALUE,LEN)                                     \
        _LRK_TYPECHECK2(RD_KAFKA_VTYPE_VALUE, void *, VALUE, size_t, LEN), \
        (void *)VALUE, (size_t)LEN
#define RD_KAFKA_V_KEY(KEY,LEN)                                         \
        _LRK_TYPECHECK2(RD_KAFKA_VTYPE_KEY, const void *, KEY, size_t, LEN), \
        (void *)KEY, (size_t)LEN
#define RD_KAFKA_V_OPAQUE(msg_opaque)                                   \
        _LRK_TYPECHECK(RD_KAFKA_VTYPE_OPAQUE, void *, msg_opaque),      \
        (void *)msg_opaque
#define RD_KAFKA_V_MSGFLAGS(msgflags)                                 \
        _LRK_TYPECHECK(RD_KAFKA_VTYPE_MSGFLAGS, int, msgflags),       \
        (int)msgflags
#define RD_KAFKA_V_TIMESTAMP(timestamp)                                 \
        _LRK_TYPECHECK(RD_KAFKA_VTYPE_TIMESTAMP, int64_t, timestamp),   \
        (int64_t)timestamp
#define RD_KAFKA_V_HEADER(NAME,VALUE,LEN)                               \
        _LRK_TYPECHECK3(RD_KAFKA_VTYPE_HEADER, const char *, NAME,      \
                        const void *, VALUE, ssize_t, LEN),             \
                (const char *)NAME, (const void *)VALUE, (ssize_t)LEN
 
#define RD_KAFKA_V_HEADERS(HDRS)                                        \
        _LRK_TYPECHECK(RD_KAFKA_VTYPE_HEADERS, rd_kafka_headers_t *, HDRS), \
                (rd_kafka_headers_t *)HDRS
 
 
RD_EXPORT rd_kafka_headers_t *rd_kafka_headers_new (size_t initial_count);
 
RD_EXPORT void rd_kafka_headers_destroy (rd_kafka_headers_t *hdrs);
 
RD_EXPORT rd_kafka_headers_t *
rd_kafka_headers_copy (const rd_kafka_headers_t *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);
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_header_remove (rd_kafka_headers_t *hdrs, const char *name);
 
 
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);
 
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);
 
 
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);
 
 
 
// FIXME: This doesn't show up in docs for some reason
// "Compound rd_kafka_message_t is not documented."
 
typedef struct rd_kafka_message_s {
    rd_kafka_resp_err_t err;   
    rd_kafka_topic_t *rkt;     
    int32_t partition;         
    void   *payload;           
    size_t  len;               
    void   *key;               
    size_t  key_len;           
    int64_t offset;            
        void  *_private;           
} rd_kafka_message_t;
 
 
RD_EXPORT
void rd_kafka_message_destroy(rd_kafka_message_t *rkmessage);
 
 
 
 
RD_EXPORT
const char *rd_kafka_message_errstr (const rd_kafka_message_t *rkmessage);
 
 
RD_EXPORT
int64_t rd_kafka_message_timestamp (const rd_kafka_message_t *rkmessage,
                    rd_kafka_timestamp_type_t *tstype);
 
 
 
RD_EXPORT
int64_t rd_kafka_message_latency (const rd_kafka_message_t *rkmessage);
 
 
RD_EXPORT
int32_t rd_kafka_message_broker_id (const rd_kafka_message_t *rkmessage);
 
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_message_headers (const rd_kafka_message_t *rkmessage,
                          rd_kafka_headers_t **hdrsp);
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_message_detach_headers (rd_kafka_message_t *rkmessage,
                                 rd_kafka_headers_t **hdrsp);
 
 
RD_EXPORT
void rd_kafka_message_set_headers (rd_kafka_message_t *rkmessage,
                                   rd_kafka_headers_t *hdrs);
 
 
RD_EXPORT size_t rd_kafka_header_cnt (const rd_kafka_headers_t *hdrs);
 
 
typedef enum {
        RD_KAFKA_MSG_STATUS_NOT_PERSISTED = 0,
 
        RD_KAFKA_MSG_STATUS_POSSIBLY_PERSISTED = 1,
 
        RD_KAFKA_MSG_STATUS_PERSISTED =  2
} rd_kafka_msg_status_t;
 
 
RD_EXPORT rd_kafka_msg_status_t
rd_kafka_message_status (const rd_kafka_message_t *rkmessage);
 
typedef enum {
    RD_KAFKA_CONF_UNKNOWN = -2, 
    RD_KAFKA_CONF_INVALID = -1, 
    RD_KAFKA_CONF_OK = 0        
} rd_kafka_conf_res_t;
 
 
RD_EXPORT
rd_kafka_conf_t *rd_kafka_conf_new(void);
 
 
RD_EXPORT
void rd_kafka_conf_destroy(rd_kafka_conf_t *conf);
 
 
RD_EXPORT
rd_kafka_conf_t *rd_kafka_conf_dup(const rd_kafka_conf_t *conf);
 
 
RD_EXPORT
rd_kafka_conf_t *rd_kafka_conf_dup_filter (const rd_kafka_conf_t *conf,
                                           size_t filter_cnt,
                                           const char **filter);
 
 
 
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);
 
 
RD_EXPORT
void rd_kafka_conf_set_events(rd_kafka_conf_t *conf, int events);
 
 
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));
 
 
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));
 
 
RD_EXPORT
void rd_kafka_conf_set_consume_cb (rd_kafka_conf_t *conf,
                                   void (*consume_cb) (rd_kafka_message_t *
                                                       rkmessage,
                                                       void *opaque));
 
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));
 
 
 
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));
 
 
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));
 
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));
 
 
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));
 
 
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));
 
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));
 
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));
 
 
 
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));
 
RD_EXPORT void
rd_kafka_conf_set_closesocket_cb (rd_kafka_conf_t *conf,
                                  int (*closesocket_cb) (int sockfd,
                                                         void *opaque));
 
 
 
#ifndef _WIN32
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
 
 
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));
 
 
typedef enum rd_kafka_cert_type_t {
        RD_KAFKA_CERT_PUBLIC_KEY,  
        RD_KAFKA_CERT_PRIVATE_KEY, 
        RD_KAFKA_CERT_CA,          
        RD_KAFKA_CERT__CNT,
} rd_kafka_cert_type_t;
 
typedef 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,
} rd_kafka_cert_enc_t;
 
 
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);
 
 
RD_EXPORT
void rd_kafka_conf_set_engine_callback_data (rd_kafka_conf_t *conf,
                                             void *callback_data);
 
 
RD_EXPORT
void rd_kafka_conf_set_opaque(rd_kafka_conf_t *conf, void *opaque);
 
RD_EXPORT
void *rd_kafka_opaque(const rd_kafka_t *rk);
 
 
 
RD_EXPORT
void rd_kafka_conf_set_default_topic_conf (rd_kafka_conf_t *conf,
                                           rd_kafka_topic_conf_t *tconf);
 
RD_EXPORT rd_kafka_topic_conf_t *
rd_kafka_conf_get_default_topic_conf (rd_kafka_conf_t *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);
 
 
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);
 
 
RD_EXPORT
const char **rd_kafka_conf_dump(rd_kafka_conf_t *conf, size_t *cntp);
 
 
RD_EXPORT
const char **rd_kafka_topic_conf_dump(rd_kafka_topic_conf_t *conf,
                       size_t *cntp);
 
RD_EXPORT
void rd_kafka_conf_dump_free(const char **arr, size_t cnt);
 
RD_EXPORT
void rd_kafka_conf_properties_show(FILE *fp);
 
RD_EXPORT
rd_kafka_topic_conf_t *rd_kafka_topic_conf_new(void);
 
 
RD_EXPORT
rd_kafka_topic_conf_t *rd_kafka_topic_conf_dup(const rd_kafka_topic_conf_t
                        *conf);
 
RD_EXPORT
rd_kafka_topic_conf_t *rd_kafka_default_topic_conf_dup (rd_kafka_t *rk);
 
 
RD_EXPORT
void rd_kafka_topic_conf_destroy(rd_kafka_topic_conf_t *topic_conf);
 
 
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);
 
RD_EXPORT
void rd_kafka_topic_conf_set_opaque(rd_kafka_topic_conf_t *conf,
                                    void *rkt_opaque);
 
 
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));
 
 
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));
 
 
RD_EXPORT
int rd_kafka_topic_partition_available(const rd_kafka_topic_t *rkt,
                    int32_t partition);
 
 
/*******************************************************************
 *                                 *
 * Partitioners provided by rdkafka                                *
 *                                 *
 *******************************************************************/
 
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);
 
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);
 
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);
 
 
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);
 
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);
 
 
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);
 
 
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);
 
 
RD_EXPORT
rd_kafka_t *rd_kafka_new(rd_kafka_type_t type, rd_kafka_conf_t *conf,
              char *errstr, size_t errstr_size);
 
 
RD_EXPORT
void        rd_kafka_destroy(rd_kafka_t *rk);
 
 
RD_EXPORT
void rd_kafka_destroy_flags (rd_kafka_t *rk, int flags);
 
#define RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE 0x8
 
 
 
RD_EXPORT
const char *rd_kafka_name(const rd_kafka_t *rk);
 
 
RD_EXPORT
rd_kafka_type_t rd_kafka_type(const rd_kafka_t *rk);
 
 
RD_EXPORT
char *rd_kafka_memberid (const rd_kafka_t *rk);
 
 
 
RD_EXPORT
char *rd_kafka_clusterid (rd_kafka_t *rk, int timeout_ms);
 
 
RD_EXPORT
int32_t rd_kafka_controllerid (rd_kafka_t *rk, int timeout_ms);
 
 
RD_EXPORT
rd_kafka_topic_t *rd_kafka_topic_new(rd_kafka_t *rk, const char *topic,
                      rd_kafka_topic_conf_t *conf);
 
 
 
RD_EXPORT
void rd_kafka_topic_destroy(rd_kafka_topic_t *rkt);
 
 
RD_EXPORT
const char *rd_kafka_topic_name(const rd_kafka_topic_t *rkt);
 
 
RD_EXPORT
void *rd_kafka_topic_opaque (const rd_kafka_topic_t *rkt);
 
 
#define RD_KAFKA_PARTITION_UA  ((int32_t)-1)
 
 
RD_EXPORT
int rd_kafka_poll(rd_kafka_t *rk, int timeout_ms);
 
 
RD_EXPORT
void rd_kafka_yield (rd_kafka_t *rk);
 
 
 
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_pause_partitions (rd_kafka_t *rk,
               rd_kafka_topic_partition_list_t *partitions);
 
 
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_resume_partitions (rd_kafka_t *rk,
                rd_kafka_topic_partition_list_t *partitions);
 
 
 
 
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);
 
 
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);
 
 
 
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);
 
 
 
RD_EXPORT
void *rd_kafka_mem_calloc (rd_kafka_t *rk, size_t num, size_t size);
 
 
 
RD_EXPORT
void *rd_kafka_mem_malloc (rd_kafka_t *rk, size_t size);
 
 
 
RD_EXPORT
void rd_kafka_mem_free (rd_kafka_t *rk, void *ptr);
 
 
RD_EXPORT
rd_kafka_queue_t *rd_kafka_queue_new(rd_kafka_t *rk);
 
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_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);
 
RD_EXPORT
rd_kafka_resp_err_t rd_kafka_set_log_queue (rd_kafka_t *rk,
                                            rd_kafka_queue_t *rkqu);
 
 
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);
 
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);
 
 
RD_EXPORT
void rd_kafka_queue_yield (rd_kafka_queue_t *rkqu);
 
 
#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_BASE -2000 /* internal: do not use */
#define RD_KAFKA_OFFSET_TAIL(CNT)  (RD_KAFKA_OFFSET_TAIL_BASE - (CNT))
 
RD_EXPORT
int rd_kafka_consume_start(rd_kafka_topic_t *rkt, int32_t partition,
                int64_t offset);
 
RD_EXPORT
int rd_kafka_consume_start_queue(rd_kafka_topic_t *rkt, int32_t partition,
                  int64_t offset, rd_kafka_queue_t *rkqu);
 
RD_EXPORT
int rd_kafka_consume_stop(rd_kafka_topic_t *rkt, int32_t partition);
 
 
 
RD_EXPORT
rd_kafka_resp_err_t rd_kafka_seek (rd_kafka_topic_t *rkt,
                                   int32_t partition,
                                   int64_t offset,
                                   int timeout_ms);
 
 
 
RD_EXPORT rd_kafka_error_t *
rd_kafka_seek_partitions (rd_kafka_t *rk,
                          rd_kafka_topic_partition_list_t *partitions,
                          int timeout_ms);
 
 
RD_EXPORT
rd_kafka_message_t *rd_kafka_consume(rd_kafka_topic_t *rkt, int32_t partition,
                      int timeout_ms);
 
 
 
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);
 
 
 
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);
 
 
RD_EXPORT
rd_kafka_message_t *rd_kafka_consume_queue(rd_kafka_queue_t *rkqu,
                                           int timeout_ms);
 
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);
 
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);
 
 
RD_EXPORT
rd_kafka_resp_err_t rd_kafka_offset_store (rd_kafka_topic_t *rkt,
                       int32_t partition, int64_t offset);
 
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_offsets_store (rd_kafka_t *rk,
                        rd_kafka_topic_partition_list_t *offsets);
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_subscribe (rd_kafka_t *rk,
                    const rd_kafka_topic_partition_list_t *topics);
 
 
RD_EXPORT
rd_kafka_resp_err_t rd_kafka_unsubscribe (rd_kafka_t *rk);
 
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_subscription (rd_kafka_t *rk,
                       rd_kafka_topic_partition_list_t **topics);
 
 
 
RD_EXPORT
rd_kafka_message_t *rd_kafka_consumer_poll (rd_kafka_t *rk, int timeout_ms);
 
RD_EXPORT
rd_kafka_resp_err_t rd_kafka_consumer_close (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);
 
 
RD_EXPORT rd_kafka_error_t *
rd_kafka_incremental_unassign (rd_kafka_t *rk,
                               const rd_kafka_topic_partition_list_t
                               *partitions);
 
 
RD_EXPORT
const char *rd_kafka_rebalance_protocol (rd_kafka_t *rk);
 
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_assign (rd_kafka_t *rk,
                 const rd_kafka_topic_partition_list_t *partitions);
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_assignment (rd_kafka_t *rk,
                     rd_kafka_topic_partition_list_t **partitions);
 
 
RD_EXPORT int
rd_kafka_assignment_lost (rd_kafka_t *rk);
 
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_commit (rd_kafka_t *rk, const rd_kafka_topic_partition_list_t *offsets,
                 int async);
 
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_commit_message (rd_kafka_t *rk, const rd_kafka_message_t *rkmessage,
                         int async);
 
 
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);
 
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_committed (rd_kafka_t *rk,
            rd_kafka_topic_partition_list_t *partitions,
            int timeout_ms);
 
 
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_position (rd_kafka_t *rk,
           rd_kafka_topic_partition_list_t *partitions);
 
 
 
 
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);
 
 
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);
 
 
RD_EXPORT void
rd_kafka_consumer_group_metadata_destroy (rd_kafka_consumer_group_metadata_t *);
 
 
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);
 
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);
 
#define RD_KAFKA_MSG_F_FREE  0x1 
#define RD_KAFKA_MSG_F_COPY  0x2 
#define RD_KAFKA_MSG_F_BLOCK 0x4 
#define RD_KAFKA_MSG_F_PARTITION 0x8 
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);
 
 
RD_EXPORT
rd_kafka_resp_err_t rd_kafka_producev (rd_kafka_t *rk, ...);
 
 
RD_EXPORT
rd_kafka_error_t *rd_kafka_produceva (rd_kafka_t *rk,
                                      const rd_kafka_vu_t *vus,
                                      size_t cnt);
 
 
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);
 
 
 
 
RD_EXPORT
rd_kafka_resp_err_t rd_kafka_flush (rd_kafka_t *rk, int timeout_ms);
 
 
 
RD_EXPORT
rd_kafka_resp_err_t rd_kafka_purge (rd_kafka_t *rk, int purge_flags);
 
 
#define RD_KAFKA_PURGE_F_QUEUE 0x1
 
#define RD_KAFKA_PURGE_F_INFLIGHT 0x2
 
 
#define RD_KAFKA_PURGE_F_NON_BLOCKING 0x4
 
 
typedef struct rd_kafka_metadata_broker {
        int32_t     id;             
        char       *host;           
        int         port;           
} rd_kafka_metadata_broker_t;
 
typedef struct rd_kafka_metadata_partition {
        int32_t     id;             
        rd_kafka_resp_err_t err;    
        int32_t     leader;         
        int         replica_cnt;    
        int32_t    *replicas;       
        int         isr_cnt;        
        int32_t    *isrs;           
} rd_kafka_metadata_partition_t;
 
typedef struct rd_kafka_metadata_topic {
        char       *topic;          
        int         partition_cnt;  
        struct rd_kafka_metadata_partition *partitions; 
        rd_kafka_resp_err_t err;    
} rd_kafka_metadata_topic_t;
 
 
typedef struct rd_kafka_metadata {
        int         broker_cnt;     
        struct rd_kafka_metadata_broker *brokers;  
        int         topic_cnt;      
        struct rd_kafka_metadata_topic *topics;    
        int32_t     orig_broker_id;   
        char       *orig_broker_name; 
} rd_kafka_metadata_t;
 
 
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);
 
RD_EXPORT
void rd_kafka_metadata_destroy(const struct rd_kafka_metadata *metadata);
 
 
struct rd_kafka_group_member_info {
        char *member_id;            
        char *client_id;            
        char *client_host;          
        void *member_metadata;      
        int   member_metadata_size; 
        void *member_assignment;    
        int   member_assignment_size; 
};
 
struct rd_kafka_group_info {
        struct rd_kafka_metadata_broker broker; 
        char *group;                            
        rd_kafka_resp_err_t err;                
        char *state;                            
        char *protocol_type;                    
        char *protocol;                         
        struct rd_kafka_group_member_info *members; 
        int   member_cnt;                       
};
 
struct rd_kafka_group_list {
        struct rd_kafka_group_info *groups;   
        int    group_cnt;                     
};
 
 
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);
 
RD_EXPORT
void rd_kafka_group_list_destroy (const struct rd_kafka_group_list *grplist);
 
 
RD_EXPORT
int rd_kafka_brokers_add(rd_kafka_t *rk, const char *brokerlist);
 
 
 
 
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));
 
 
RD_EXPORT
void rd_kafka_set_log_level(rd_kafka_t *rk, int level);
 
 
RD_EXPORT
void rd_kafka_log_print(const rd_kafka_t *rk, int level,
             const char *fac, const char *buf);
 
 
RD_EXPORT
void rd_kafka_log_syslog(const rd_kafka_t *rk, int level,
              const char *fac, const char *buf);
 
 
RD_EXPORT
int         rd_kafka_outq_len(rd_kafka_t *rk);
 
 
 
RD_EXPORT
void rd_kafka_dump(FILE *fp, rd_kafka_t *rk);
 
 
 
RD_EXPORT
int rd_kafka_thread_cnt(void);
 
 
typedef enum rd_kafka_thread_type_t {
        RD_KAFKA_THREAD_MAIN,       
        RD_KAFKA_THREAD_BACKGROUND, 
        RD_KAFKA_THREAD_BROKER      
} rd_kafka_thread_type_t;
 
 
RD_EXPORT
int rd_kafka_wait_destroyed(int timeout_ms);
 
 
RD_EXPORT
int rd_kafka_unittest (void);
 
 
RD_EXPORT
rd_kafka_resp_err_t rd_kafka_poll_set_consumer (rd_kafka_t *rk);
 
 
typedef int rd_kafka_event_type_t;
#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 
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);
 
 
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);
 
 
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);
 
 
RD_EXPORT
int rd_kafka_event_debug_contexts (rd_kafka_event_t *rkev,
            char *dst, size_t dstsize);
 
 
RD_EXPORT
const char *rd_kafka_event_stats (rd_kafka_event_t *rkev);
 
 
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);
 
 
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_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 const rd_kafka_CreateTopics_result_t *
rd_kafka_event_CreateTopics_result (rd_kafka_event_t *rkev);
 
RD_EXPORT const rd_kafka_DeleteTopics_result_t *
rd_kafka_event_DeleteTopics_result (rd_kafka_event_t *rkev);
 
RD_EXPORT const rd_kafka_CreatePartitions_result_t *
rd_kafka_event_CreatePartitions_result (rd_kafka_event_t *rkev);
 
RD_EXPORT const rd_kafka_AlterConfigs_result_t *
rd_kafka_event_AlterConfigs_result (rd_kafka_event_t *rkev);
 
RD_EXPORT const rd_kafka_DescribeConfigs_result_t *
rd_kafka_event_DescribeConfigs_result (rd_kafka_event_t *rkev);
 
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);
 
RD_EXPORT const rd_kafka_DeleteConsumerGroupOffsets_result_t *
rd_kafka_event_DeleteConsumerGroupOffsets_result (rd_kafka_event_t *rkev);
 
RD_EXPORT
rd_kafka_event_t *rd_kafka_queue_poll (rd_kafka_queue_t *rkqu, int timeout_ms);
 
RD_EXPORT
int rd_kafka_queue_poll_callback (rd_kafka_queue_t *rkqu, int timeout_ms);
 
 
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);
 
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);
 
 
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);
 
 
typedef rd_kafka_resp_err_t
(rd_kafka_interceptor_f_on_conf_destroy_t) (void *ic_opaque);
 
 
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);
 
 
typedef rd_kafka_resp_err_t
(rd_kafka_interceptor_f_on_destroy_t) (rd_kafka_t *rk, void *ic_opaque);
 
 
 
 
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);
 
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);
 
 
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);
 
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);
 
 
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);
 
 
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);
 
 
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);
 
 
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);
 
 
 
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);
 
 
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);
 
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);
 
 
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);
 
 
 
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);
 
 
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);
 
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);
 
 
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);
 
 
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);
 
 
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);
 
 
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);
 
 
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);
 
 
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);
 
 
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_topic_result_error (const rd_kafka_topic_result_t *topicres);
 
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);
 
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);
 
 
typedef 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__CNT              
} rd_kafka_admin_op_t;
 
typedef struct rd_kafka_AdminOptions_s rd_kafka_AdminOptions_t;
 
RD_EXPORT rd_kafka_AdminOptions_t *
rd_kafka_AdminOptions_new (rd_kafka_t *rk, rd_kafka_admin_op_t for_api);
 
 
RD_EXPORT void rd_kafka_AdminOptions_destroy (rd_kafka_AdminOptions_t *options);
 
 
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);
 
 
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);
 
 
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);
 
 
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);
 
 
 
RD_EXPORT void
rd_kafka_AdminOptions_set_opaque (rd_kafka_AdminOptions_t *options,
                                  void *ev_opaque);
 
 
 
 
 
 
/*
 * CreateTopics - create topics in cluster.
 *
 */
 
 
typedef struct rd_kafka_NewTopic_s rd_kafka_NewTopic_t;
 
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);
 
RD_EXPORT void
rd_kafka_NewTopic_destroy (rd_kafka_NewTopic_t *new_topic);
 
 
RD_EXPORT void
rd_kafka_NewTopic_destroy_array (rd_kafka_NewTopic_t **new_topics,
                                 size_t new_topic_cnt);
 
 
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);
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_NewTopic_set_config (rd_kafka_NewTopic_t *new_topic,
                              const char *name, const char *value);
 
 
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);
 
 
/*
 * CreateTopics result type and methods
 */
 
RD_EXPORT const rd_kafka_topic_result_t **
rd_kafka_CreateTopics_result_topics (
        const rd_kafka_CreateTopics_result_t *result,
        size_t *cntp);
 
 
 
 
 
/*
 * DeleteTopics - delete topics from cluster
 *
 */
 
typedef struct rd_kafka_DeleteTopic_s rd_kafka_DeleteTopic_t;
 
RD_EXPORT rd_kafka_DeleteTopic_t *
rd_kafka_DeleteTopic_new (const char *topic);
 
RD_EXPORT void
rd_kafka_DeleteTopic_destroy (rd_kafka_DeleteTopic_t *del_topic);
 
RD_EXPORT void
rd_kafka_DeleteTopic_destroy_array (rd_kafka_DeleteTopic_t **del_topics,
                                    size_t del_topic_cnt);
 
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);
 
 
 
/*
 * DeleteTopics result type and methods
 */
 
RD_EXPORT const rd_kafka_topic_result_t **
rd_kafka_DeleteTopics_result_topics (
        const rd_kafka_DeleteTopics_result_t *result,
        size_t *cntp);
 
 
 
 
 
 
/*
 * CreatePartitions - add partitions to topic.
 *
 */
 
typedef struct rd_kafka_NewPartitions_s rd_kafka_NewPartitions_t;
 
RD_EXPORT rd_kafka_NewPartitions_t *
rd_kafka_NewPartitions_new (const char *topic, size_t new_total_cnt,
                            char *errstr, size_t errstr_size);
 
RD_EXPORT void
rd_kafka_NewPartitions_destroy (rd_kafka_NewPartitions_t *new_parts);
 
RD_EXPORT void
rd_kafka_NewPartitions_destroy_array (rd_kafka_NewPartitions_t **new_parts,
                                      size_t new_parts_cnt);
 
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);
 
 
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);
 
 
 
/*
 * CreatePartitions result type and methods
 */
 
RD_EXPORT const rd_kafka_topic_result_t **
rd_kafka_CreatePartitions_result_topics (
        const rd_kafka_CreatePartitions_result_t *result,
        size_t *cntp);
 
 
 
 
 
/*
 * Cluster, broker, topic configuration entries, sources, etc.
 *
 */
 
typedef 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,
} rd_kafka_ConfigSource_t;
 
 
RD_EXPORT const char *
rd_kafka_ConfigSource_name (rd_kafka_ConfigSource_t confsource);
 
 
typedef struct rd_kafka_ConfigEntry_s rd_kafka_ConfigEntry_t;
 
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);
 
 
 
 
typedef 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,        
} rd_kafka_ResourceType_t;
 
RD_EXPORT const char *
rd_kafka_ResourceType_name (rd_kafka_ResourceType_t restype);
 
typedef struct rd_kafka_ConfigResource_s rd_kafka_ConfigResource_t;
 
 
RD_EXPORT rd_kafka_ConfigResource_t *
rd_kafka_ConfigResource_new (rd_kafka_ResourceType_t restype,
                             const char *resname);
 
RD_EXPORT void
rd_kafka_ConfigResource_destroy (rd_kafka_ConfigResource_t *config);
 
 
RD_EXPORT void
rd_kafka_ConfigResource_destroy_array (rd_kafka_ConfigResource_t **config,
                                       size_t config_cnt);
 
 
RD_EXPORT rd_kafka_resp_err_t
rd_kafka_ConfigResource_set_config (rd_kafka_ConfigResource_t *config,
                                    const char *name, const char *value);
 
 
RD_EXPORT const rd_kafka_ConfigEntry_t **
rd_kafka_ConfigResource_configs (const rd_kafka_ConfigResource_t *config,
                                 size_t *cntp);
 
 
 
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);
 
 
/*
 * AlterConfigs - alter cluster configuration.
 *
 */
 
 
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);
 
 
/*
 * AlterConfigs result type and methods
 */
 
RD_EXPORT const rd_kafka_ConfigResource_t **
rd_kafka_AlterConfigs_result_resources (
        const rd_kafka_AlterConfigs_result_t *result,
        size_t *cntp);
 
 
 
 
 
 
/*
 * DescribeConfigs - retrieve cluster configuration.
 *
 */
 
 
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);
 
 
 
 
/*
 * DescribeConfigs result type and methods
 */
 
RD_EXPORT const rd_kafka_ConfigResource_t **
rd_kafka_DescribeConfigs_result_resources (
        const rd_kafka_DescribeConfigs_result_t *result,
        size_t *cntp);
 
 
/*
 * DeleteRecords - delete records (messages) from partitions
 *
 *
 */
 
typedef struct rd_kafka_DeleteRecords_s rd_kafka_DeleteRecords_t;
 
RD_EXPORT rd_kafka_DeleteRecords_t *
rd_kafka_DeleteRecords_new (const rd_kafka_topic_partition_list_t *
                            before_offsets);
 
RD_EXPORT void
rd_kafka_DeleteRecords_destroy (rd_kafka_DeleteRecords_t *del_records);
 
RD_EXPORT void
rd_kafka_DeleteRecords_destroy_array (rd_kafka_DeleteRecords_t **del_records,
                                      size_t del_record_cnt);
 
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);
 
 
/*
 * DeleteRecords result type and methods
 */
 
RD_EXPORT const rd_kafka_topic_partition_list_t *
rd_kafka_DeleteRecords_result_offsets (
    const rd_kafka_DeleteRecords_result_t *result);
 
/*
 * DeleteGroups - delete groups from cluster
 *
 *
 */
 
typedef struct rd_kafka_DeleteGroup_s rd_kafka_DeleteGroup_t;
 
RD_EXPORT rd_kafka_DeleteGroup_t *
rd_kafka_DeleteGroup_new (const char *group);
 
RD_EXPORT void
rd_kafka_DeleteGroup_destroy (rd_kafka_DeleteGroup_t *del_group);
 
RD_EXPORT void
rd_kafka_DeleteGroup_destroy_array (rd_kafka_DeleteGroup_t **del_groups,
                                    size_t del_group_cnt);
 
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);
 
 
 
/*
 * DeleteGroups result type and methods
 */
 
RD_EXPORT const rd_kafka_group_result_t **
rd_kafka_DeleteGroups_result_groups (
        const rd_kafka_DeleteGroups_result_t *result,
        size_t *cntp);
 
 
/*
 * DeleteConsumerGroupOffsets - delete groups from cluster
 *
 *
 */
 
typedef struct rd_kafka_DeleteConsumerGroupOffsets_s
rd_kafka_DeleteConsumerGroupOffsets_t;
 
RD_EXPORT rd_kafka_DeleteConsumerGroupOffsets_t *
rd_kafka_DeleteConsumerGroupOffsets_new (const char *group,
                                         const rd_kafka_topic_partition_list_t
                                         *partitions);
 
RD_EXPORT void
rd_kafka_DeleteConsumerGroupOffsets_destroy (
        rd_kafka_DeleteConsumerGroupOffsets_t *del_grpoffsets);
 
RD_EXPORT void
rd_kafka_DeleteConsumerGroupOffsets_destroy_array (
        rd_kafka_DeleteConsumerGroupOffsets_t **del_grpoffsets,
        size_t del_grpoffset_cnt);
 
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);
 
 
 
/*
 * DeleteConsumerGroupOffsets result type and methods
 */
 
RD_EXPORT const rd_kafka_group_result_t **
rd_kafka_DeleteConsumerGroupOffsets_result_groups (
        const rd_kafka_DeleteConsumerGroupOffsets_result_t *result,
        size_t *cntp);
 
 
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);
 
RD_EXPORT
rd_kafka_resp_err_t
rd_kafka_oauthbearer_set_token_failure (rd_kafka_t *rk, const char *errstr);
 
RD_EXPORT
rd_kafka_error_t *
rd_kafka_init_transactions (rd_kafka_t *rk, int timeout_ms);
 
 
 
RD_EXPORT
rd_kafka_error_t *rd_kafka_begin_transaction (rd_kafka_t *rk);
 
 
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);
 
 
RD_EXPORT
rd_kafka_error_t *
rd_kafka_commit_transaction (rd_kafka_t *rk, int timeout_ms);
 
 
RD_EXPORT
rd_kafka_error_t *
rd_kafka_abort_transaction (rd_kafka_t *rk, int timeout_ms);
 
 
/* @cond NO_DOC */
#ifdef __cplusplus
}
#endif
#endif /* _RDKAFKA_H_ */
/* @endcond NO_DOC */