librdkafka
The Apache Kafka C/C++ client library
Loading...
Searching...
No Matches
rdkafka.h
Go to the documentation of this file.
1/*
2 * librdkafka - Apache Kafka C library
3 *
4 * Copyright (c) 2012-2022, Magnus Edenhill
5 * 2023, Confluent Inc.
6 * All rights reserved.
7 *
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions are met:
10 *
11 * 1. Redistributions of source code must retain the above copyright notice,
12 * this list of conditions and the following disclaimer.
13 * 2. Redistributions in binary form must reproduce the above copyright notice,
14 * this list of conditions and the following disclaimer in the documentation
15 * and/or other materials provided with the distribution.
16 *
17 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
18 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
19 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
20 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
21 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
22 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
23 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
24 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
25 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
26 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
27 * POSSIBILITY OF SUCH DAMAGE.
28 */
29
44/* @cond NO_DOC */
45#ifndef _RDKAFKA_H_
46#define _RDKAFKA_H_
47
48#include <stdio.h>
49#include <inttypes.h>
50#include <sys/types.h>
51
52#ifdef __cplusplus
53extern "C" {
54#if 0
55} /* Restore indent */
56#endif
57#endif
58
59#ifdef _WIN32
60#include <basetsd.h>
61#ifndef WIN32_MEAN_AND_LEAN
62#define WIN32_MEAN_AND_LEAN
63#endif
64#include <winsock2.h> /* for sockaddr, .. */
65#ifndef _SSIZE_T_DEFINED
66#define _SSIZE_T_DEFINED
67typedef SSIZE_T ssize_t;
68#endif
69#define RD_UNUSED
70#define RD_INLINE __inline
71#define RD_DEPRECATED __declspec(deprecated)
72#define RD_FORMAT(...)
73#undef RD_EXPORT
74#ifdef LIBRDKAFKA_STATICLIB
75#define RD_EXPORT
76#else
77#ifdef LIBRDKAFKA_EXPORTS
78#define RD_EXPORT __declspec(dllexport)
79#else
80#define RD_EXPORT __declspec(dllimport)
81#endif
82#ifndef LIBRDKAFKA_TYPECHECKS
83#define LIBRDKAFKA_TYPECHECKS 0
84#endif
85#endif
86
87#else
88#include <sys/socket.h> /* for sockaddr, .. */
89
90#define RD_UNUSED __attribute__((unused))
91#define RD_INLINE inline
92#define RD_EXPORT
93#define RD_DEPRECATED __attribute__((deprecated))
94
95#if defined(__clang__) || defined(__GNUC__) || defined(__GNUG__)
96#define RD_HAS_STATEMENT_EXPRESSIONS
97#define RD_FORMAT(...) __attribute__((format(__VA_ARGS__)))
98#else
99#define RD_FORMAT(...)
100#endif
101
102#ifndef LIBRDKAFKA_TYPECHECKS
103#define LIBRDKAFKA_TYPECHECKS 1
104#endif
105#endif
106
107
113#if LIBRDKAFKA_TYPECHECKS
114#define _LRK_TYPECHECK(RET, TYPE, ARG) \
115 ({ \
116 if (0) { \
117 TYPE __t RD_UNUSED = (ARG); \
118 } \
119 RET; \
120 })
121
122#define _LRK_TYPECHECK2(RET, TYPE, ARG, TYPE2, ARG2) \
123 ({ \
124 if (0) { \
125 TYPE __t RD_UNUSED = (ARG); \
126 TYPE2 __t2 RD_UNUSED = (ARG2); \
127 } \
128 RET; \
129 })
130
131#define _LRK_TYPECHECK3(RET, TYPE, ARG, TYPE2, ARG2, TYPE3, ARG3) \
132 ({ \
133 if (0) { \
134 TYPE __t RD_UNUSED = (ARG); \
135 TYPE2 __t2 RD_UNUSED = (ARG2); \
136 TYPE3 __t3 RD_UNUSED = (ARG3); \
137 } \
138 RET; \
139 })
140#else
141#define _LRK_TYPECHECK(RET, TYPE, ARG) (RET)
142#define _LRK_TYPECHECK2(RET, TYPE, ARG, TYPE2, ARG2) (RET)
143#define _LRK_TYPECHECK3(RET, TYPE, ARG, TYPE2, ARG2, TYPE3, ARG3) (RET)
144#endif
145
146/* @endcond */
147
148
170#define RD_KAFKA_VERSION 0x020b00ff
171
180RD_EXPORT
182
188RD_EXPORT
189const char *rd_kafka_version_str(void);
190
213
214
225
226
227
234RD_EXPORT
236
244#define RD_KAFKA_DEBUG_CONTEXTS \
245 "all,generic,broker,topic,metadata,feature,queue,msg,protocol,cgrp," \
246 "security,fetch,interceptor,plugin,consumer,admin,eos,mock,assignor," \
247 "conf"
248
249
250/* @cond NO_DOC */
251/* Private types to provide ABI compatibility */
252typedef struct rd_kafka_s rd_kafka_t;
253typedef struct rd_kafka_topic_s rd_kafka_topic_t;
254typedef struct rd_kafka_conf_s rd_kafka_conf_t;
255typedef struct rd_kafka_topic_conf_s rd_kafka_topic_conf_t;
256typedef struct rd_kafka_queue_s rd_kafka_queue_t;
257typedef struct rd_kafka_op_s rd_kafka_event_t;
258typedef struct rd_kafka_topic_result_s rd_kafka_topic_result_t;
259typedef struct rd_kafka_consumer_group_metadata_s
260 rd_kafka_consumer_group_metadata_t;
261typedef struct rd_kafka_error_s rd_kafka_error_t;
262typedef struct rd_kafka_headers_s rd_kafka_headers_t;
263typedef struct rd_kafka_group_result_s rd_kafka_group_result_t;
264typedef struct rd_kafka_acl_result_s rd_kafka_acl_result_t;
265typedef struct rd_kafka_Uuid_s rd_kafka_Uuid_t;
266typedef struct rd_kafka_topic_partition_result_s
267 rd_kafka_topic_partition_result_t;
268/* @endcond */
269
270
283typedef enum {
284 /* Internal errors to rdkafka: */
417
420
421 /* Kafka broker errors: */
437#define RD_KAFKA_RESP_ERR_NOT_LEADER_OR_FOLLOWER \
438 RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION
439 RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION = 6,
457#define RD_KAFKA_RESP_ERR_GROUP_LOAD_IN_PROGRESS \
458 RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS
462#define RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE \
463 RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE
467#define RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP \
468 RD_KAFKA_RESP_ERR_NOT_COORDINATOR
662 RD_KAFKA_RESP_ERR_END_ALL,
664
665
673 const char *name;
674 const char *desc;
675};
676
677
681RD_EXPORT
682void rd_kafka_get_err_descs(const struct rd_kafka_err_desc **errdescs,
683 size_t *cntp);
684
685
686
692RD_EXPORT
694
695
696
702RD_EXPORT
704
705
731RD_EXPORT
733
734
759RD_EXPORT RD_DEPRECATED rd_kafka_resp_err_t rd_kafka_errno2err(int errnox);
760
761
774RD_EXPORT RD_DEPRECATED int rd_kafka_errno(void);
775
776
777
804RD_EXPORT
806rd_kafka_fatal_error(rd_kafka_t *rk, char *errstr, size_t errstr_size);
807
808
828 const char *reason);
829
830
835RD_EXPORT
836rd_kafka_resp_err_t rd_kafka_error_code(const rd_kafka_error_t *error);
837
846RD_EXPORT
847const char *rd_kafka_error_name(const rd_kafka_error_t *error);
848
855RD_EXPORT
856const char *rd_kafka_error_string(const rd_kafka_error_t *error);
857
858
863RD_EXPORT
864int rd_kafka_error_is_fatal(const rd_kafka_error_t *error);
865
866
871RD_EXPORT
872int rd_kafka_error_is_retriable(const rd_kafka_error_t *error);
873
874
885RD_EXPORT
886int rd_kafka_error_txn_requires_abort(const rd_kafka_error_t *error);
887
893RD_EXPORT
894void rd_kafka_error_destroy(rd_kafka_error_t *error);
895
896
905RD_EXPORT
907 const char *fmt,
908 ...) RD_FORMAT(printf, 2, 3);
909
910
926typedef struct rd_kafka_topic_partition_s {
927 char *topic;
928 int32_t partition;
929 int64_t offset;
930 void *metadata;
932 void *opaque;
934 void *_private;
939
944RD_EXPORT
946
947
956RD_EXPORT
959 int32_t leader_epoch);
960
969RD_EXPORT
971 const rd_kafka_topic_partition_t *rktpar);
972
977typedef struct rd_kafka_topic_partition_list_s {
978 int cnt;
979 int size;
982
997RD_EXPORT
999
1003RD_EXPORT
1006
1016RD_EXPORT
1019 const char *topic,
1020 int32_t partition);
1021
1022
1031RD_EXPORT
1034 const char *topic,
1035 int32_t start,
1036 int32_t stop);
1037
1038
1039
1051RD_EXPORT
1054 const char *topic,
1055 int32_t partition);
1056
1057
1065RD_EXPORT
1068 int idx);
1069
1070
1078RD_EXPORT
1081
1082
1083
1091RD_EXPORT
1094 const char *topic,
1095 int32_t partition,
1096 int64_t offset);
1097
1098
1099
1105RD_EXPORT
1107 const rd_kafka_topic_partition_list_t *rktparlist,
1108 const char *topic,
1109 int32_t partition);
1110
1111
1123 int (*cmp)(const void *a, const void *b, void *cmp_opaque),
1124 void *cmp_opaque);
1125
1126
1162
1163
1170typedef struct rd_kafka_vu_s {
1173 union {
1174 const char *cstr;
1175 rd_kafka_topic_t *rkt;
1176 int i;
1177 int32_t i32;
1178 int64_t i64;
1179 struct {
1180 void *ptr;
1181 size_t size;
1182 } mem;
1183 struct {
1184 const char *name;
1185 const void *val;
1186 ssize_t size;
1187 } header;
1188 rd_kafka_headers_t *headers;
1189 void *ptr;
1190 char _pad[64];
1191 } u;
1193
1202#define RD_KAFKA_V_END RD_KAFKA_VTYPE_END
1203
1209#define RD_KAFKA_V_TOPIC(topic) \
1210 _LRK_TYPECHECK(RD_KAFKA_VTYPE_TOPIC, const char *, topic), \
1211 (const char *)topic
1217#define RD_KAFKA_V_RKT(rkt) \
1218 _LRK_TYPECHECK(RD_KAFKA_VTYPE_RKT, rd_kafka_topic_t *, rkt), \
1219 (rd_kafka_topic_t *)rkt
1225#define RD_KAFKA_V_PARTITION(partition) \
1226 _LRK_TYPECHECK(RD_KAFKA_VTYPE_PARTITION, int32_t, partition), \
1227 (int32_t)partition
1233#define RD_KAFKA_V_VALUE(VALUE, LEN) \
1234 _LRK_TYPECHECK2(RD_KAFKA_VTYPE_VALUE, void *, VALUE, size_t, LEN), \
1235 (void *)VALUE, (size_t)LEN
1241#define RD_KAFKA_V_KEY(KEY, LEN) \
1242 _LRK_TYPECHECK2(RD_KAFKA_VTYPE_KEY, const void *, KEY, size_t, LEN), \
1243 (void *)KEY, (size_t)LEN
1251#define RD_KAFKA_V_OPAQUE(msg_opaque) \
1252 _LRK_TYPECHECK(RD_KAFKA_VTYPE_OPAQUE, void *, msg_opaque), \
1253 (void *)msg_opaque
1260#define RD_KAFKA_V_MSGFLAGS(msgflags) \
1261 _LRK_TYPECHECK(RD_KAFKA_VTYPE_MSGFLAGS, int, msgflags), (int)msgflags
1268#define RD_KAFKA_V_TIMESTAMP(timestamp) \
1269 _LRK_TYPECHECK(RD_KAFKA_VTYPE_TIMESTAMP, int64_t, timestamp), \
1270 (int64_t)timestamp
1279#define RD_KAFKA_V_HEADER(NAME, VALUE, LEN) \
1280 _LRK_TYPECHECK3(RD_KAFKA_VTYPE_HEADER, const char *, NAME, \
1281 const void *, VALUE, ssize_t, LEN), \
1282 (const char *)NAME, (const void *)VALUE, (ssize_t)LEN
1283
1295#define RD_KAFKA_V_HEADERS(HDRS) \
1296 _LRK_TYPECHECK(RD_KAFKA_VTYPE_HEADERS, rd_kafka_headers_t *, HDRS), \
1297 (rd_kafka_headers_t *)HDRS
1298
1299
1330RD_EXPORT rd_kafka_headers_t *rd_kafka_headers_new(size_t initial_count);
1331
1336RD_EXPORT void rd_kafka_headers_destroy(rd_kafka_headers_t *hdrs);
1337
1341RD_EXPORT rd_kafka_headers_t *
1342rd_kafka_headers_copy(const rd_kafka_headers_t *src);
1343
1361RD_EXPORT rd_kafka_resp_err_t rd_kafka_header_add(rd_kafka_headers_t *hdrs,
1362 const char *name,
1363 ssize_t name_size,
1364 const void *value,
1365 ssize_t value_size);
1366
1374RD_EXPORT rd_kafka_resp_err_t rd_kafka_header_remove(rd_kafka_headers_t *hdrs,
1375 const char *name);
1376
1377
1395RD_EXPORT rd_kafka_resp_err_t
1396rd_kafka_header_get_last(const rd_kafka_headers_t *hdrs,
1397 const char *name,
1398 const void **valuep,
1399 size_t *sizep);
1400
1414RD_EXPORT rd_kafka_resp_err_t
1415rd_kafka_header_get(const rd_kafka_headers_t *hdrs,
1416 size_t idx,
1417 const char *name,
1418 const void **valuep,
1419 size_t *sizep);
1420
1421
1429RD_EXPORT rd_kafka_resp_err_t
1430rd_kafka_header_get_all(const rd_kafka_headers_t *hdrs,
1431 size_t idx,
1432 const char **namep,
1433 const void **valuep,
1434 size_t *sizep);
1435
1436
1437
1450// FIXME: This doesn't show up in docs for some reason
1451// "Compound rd_kafka_message_t is not documented."
1452
1466typedef struct rd_kafka_message_s {
1468 rd_kafka_topic_t *rkt;
1469 int32_t partition;
1470 void *payload;
1474 size_t len;
1477 void *key;
1479 size_t key_len;
1481 int64_t offset;
1489 void *_private;
1497
1498
1502RD_EXPORT
1504
1505
1506
1513RD_EXPORT
1514const char *rd_kafka_message_errstr(const rd_kafka_message_t *rkmessage);
1515
1522RD_EXPORT
1523const char *
1525
1526
1538RD_EXPORT
1541
1542
1543
1550RD_EXPORT
1552
1553
1560RD_EXPORT
1562
1563
1580RD_EXPORT rd_kafka_resp_err_t
1582 rd_kafka_headers_t **hdrsp);
1583
1595RD_EXPORT rd_kafka_resp_err_t
1597 rd_kafka_headers_t **hdrsp);
1598
1599
1612RD_EXPORT
1614 rd_kafka_headers_t *hdrs);
1615
1616
1622RD_EXPORT size_t rd_kafka_header_cnt(const rd_kafka_headers_t *hdrs);
1623
1624
1646
1647
1654RD_EXPORT rd_kafka_msg_status_t
1656
1657
1665RD_EXPORT int32_t
1667
1668
1685RD_EXPORT const char *rd_kafka_Uuid_base64str(const rd_kafka_Uuid_t *uuid);
1686
1694RD_EXPORT int64_t
1695rd_kafka_Uuid_least_significant_bits(const rd_kafka_Uuid_t *uuid);
1696
1697
1705RD_EXPORT int64_t
1706rd_kafka_Uuid_most_significant_bits(const rd_kafka_Uuid_t *uuid);
1707
1708
1718RD_EXPORT rd_kafka_Uuid_t *rd_kafka_Uuid_new(int64_t most_significant_bits,
1719 int64_t least_significant_bits);
1720
1729RD_EXPORT rd_kafka_Uuid_t *rd_kafka_Uuid_copy(const rd_kafka_Uuid_t *uuid);
1730
1736RD_EXPORT void rd_kafka_Uuid_destroy(rd_kafka_Uuid_t *uuid);
1737
1760
1761
1795RD_EXPORT
1796rd_kafka_conf_t *rd_kafka_conf_new(void);
1797
1798
1802RD_EXPORT
1803void rd_kafka_conf_destroy(rd_kafka_conf_t *conf);
1804
1805
1812RD_EXPORT
1813rd_kafka_conf_t *rd_kafka_conf_dup(const rd_kafka_conf_t *conf);
1814
1815
1820RD_EXPORT
1821rd_kafka_conf_t *rd_kafka_conf_dup_filter(const rd_kafka_conf_t *conf,
1822 size_t filter_cnt,
1823 const char **filter);
1824
1825
1826
1835RD_EXPORT
1836const rd_kafka_conf_t *rd_kafka_conf(rd_kafka_t *rk);
1837
1838
1858RD_EXPORT
1860 const char *name,
1861 const char *value,
1862 char *errstr,
1863 size_t errstr_size);
1864
1865
1871RD_EXPORT
1872void rd_kafka_conf_set_events(rd_kafka_conf_t *conf, int events);
1873
1874
1911 rd_kafka_conf_t *conf,
1912 void (*event_cb)(rd_kafka_t *rk, rd_kafka_event_t *rkev, void *opaque));
1913
1914
1918RD_EXPORT
1919void rd_kafka_conf_set_dr_cb(rd_kafka_conf_t *conf,
1920 void (*dr_cb)(rd_kafka_t *rk,
1921 void *payload,
1922 size_t len,
1924 void *opaque,
1925 void *msg_opaque));
1926
1957RD_EXPORT
1959 rd_kafka_conf_t *conf,
1960 void (*dr_msg_cb)(rd_kafka_t *rk,
1961 const rd_kafka_message_t *rkmessage,
1962 void *opaque));
1963
1964
1972RD_EXPORT
1974 rd_kafka_conf_t *conf,
1975 void (*consume_cb)(rd_kafka_message_t *rkmessage, void *opaque));
1976
2081RD_EXPORT
2083 rd_kafka_conf_t *conf,
2084 void (*rebalance_cb)(rd_kafka_t *rk,
2087 void *opaque));
2088
2089
2090
2108RD_EXPORT
2110 rd_kafka_conf_t *conf,
2111 void (*offset_commit_cb)(rd_kafka_t *rk,
2114 void *opaque));
2115
2116
2139RD_EXPORT
2140void rd_kafka_conf_set_error_cb(rd_kafka_conf_t *conf,
2141 void (*error_cb)(rd_kafka_t *rk,
2142 int err,
2143 const char *reason,
2144 void *opaque));
2145
2163RD_EXPORT
2164void rd_kafka_conf_set_throttle_cb(rd_kafka_conf_t *conf,
2165 void (*throttle_cb)(rd_kafka_t *rk,
2166 const char *broker_name,
2167 int32_t broker_id,
2168 int throttle_time_ms,
2169 void *opaque));
2170
2171
2188RD_EXPORT
2189void rd_kafka_conf_set_log_cb(rd_kafka_conf_t *conf,
2190 void (*log_cb)(const rd_kafka_t *rk,
2191 int level,
2192 const char *fac,
2193 const char *buf));
2194
2195
2218RD_EXPORT
2220 rd_kafka_conf_t *conf,
2221 int (*stats_cb)(rd_kafka_t *rk, char *json, size_t json_len, void *opaque));
2222
2272RD_EXPORT
2274 rd_kafka_conf_t *conf,
2275 void (*oauthbearer_token_refresh_cb)(rd_kafka_t *rk,
2276 const char *oauthbearer_config,
2277 void *opaque));
2278
2306RD_EXPORT
2307void rd_kafka_conf_enable_sasl_queue(rd_kafka_conf_t *conf, int enable);
2308
2309
2327RD_EXPORT
2329 rd_kafka_conf_t *conf,
2330 int (*socket_cb)(int domain, int type, int protocol, void *opaque));
2331
2332
2333
2349RD_EXPORT void
2350rd_kafka_conf_set_connect_cb(rd_kafka_conf_t *conf,
2351 int (*connect_cb)(int sockfd,
2352 const struct sockaddr *addr,
2353 int addrlen,
2354 const char *id,
2355 void *opaque));
2356
2368 rd_kafka_conf_t *conf,
2369 int (*closesocket_cb)(int sockfd, void *opaque));
2370
2371
2372
2373#ifndef _WIN32
2391RD_EXPORT
2393 rd_kafka_conf_t *conf,
2394 int (*open_cb)(const char *pathname, int flags, mode_t mode, void *opaque));
2395#endif
2396
2398struct addrinfo;
2399
2419RD_EXPORT void
2420rd_kafka_conf_set_resolve_cb(rd_kafka_conf_t *conf,
2421 int (*resolve_cb)(const char *node,
2422 const char *service,
2423 const struct addrinfo *hints,
2424 struct addrinfo **res,
2425 void *opaque));
2426
2468RD_EXPORT
2470 rd_kafka_conf_t *conf,
2471 int (*ssl_cert_verify_cb)(rd_kafka_t *rk,
2472 const char *broker_name,
2473 int32_t broker_id,
2474 int *x509_error,
2475 int depth,
2476 const char *buf,
2477 size_t size,
2478 char *errstr,
2479 size_t errstr_size,
2480 void *opaque));
2481
2482
2496
2510
2511
2554RD_EXPORT rd_kafka_conf_res_t
2555rd_kafka_conf_set_ssl_cert(rd_kafka_conf_t *conf,
2556 rd_kafka_cert_type_t cert_type,
2557 rd_kafka_cert_enc_t cert_enc,
2558 const void *buffer,
2559 size_t size,
2560 char *errstr,
2561 size_t errstr_size);
2562
2563
2578RD_EXPORT
2580 void *callback_data);
2581
2582
2588RD_EXPORT
2589void rd_kafka_conf_set_opaque(rd_kafka_conf_t *conf, void *opaque);
2590
2595RD_EXPORT
2596void *rd_kafka_opaque(const rd_kafka_t *rk);
2597
2598
2599
2613RD_EXPORT
2614void rd_kafka_conf_set_default_topic_conf(rd_kafka_conf_t *conf,
2615 rd_kafka_topic_conf_t *tconf);
2616
2629RD_EXPORT rd_kafka_topic_conf_t *
2631
2632
2652RD_EXPORT
2653rd_kafka_conf_res_t rd_kafka_conf_get(const rd_kafka_conf_t *conf,
2654 const char *name,
2655 char *dest,
2656 size_t *dest_size);
2657
2658
2664RD_EXPORT
2665rd_kafka_conf_res_t rd_kafka_topic_conf_get(const rd_kafka_topic_conf_t *conf,
2666 const char *name,
2667 char *dest,
2668 size_t *dest_size);
2669
2670
2679RD_EXPORT
2680const char **rd_kafka_conf_dump(rd_kafka_conf_t *conf, size_t *cntp);
2681
2682
2691RD_EXPORT
2692const char **rd_kafka_topic_conf_dump(rd_kafka_topic_conf_t *conf,
2693 size_t *cntp);
2694
2699RD_EXPORT
2700void rd_kafka_conf_dump_free(const char **arr, size_t cnt);
2701
2709RD_EXPORT
2711
2728RD_EXPORT
2729rd_kafka_topic_conf_t *rd_kafka_topic_conf_new(void);
2730
2731
2735RD_EXPORT
2736rd_kafka_topic_conf_t *
2737rd_kafka_topic_conf_dup(const rd_kafka_topic_conf_t *conf);
2738
2743RD_EXPORT
2744rd_kafka_topic_conf_t *rd_kafka_default_topic_conf_dup(rd_kafka_t *rk);
2745
2746
2750RD_EXPORT
2751void rd_kafka_topic_conf_destroy(rd_kafka_topic_conf_t *topic_conf);
2752
2753
2762RD_EXPORT
2764 const char *name,
2765 const char *value,
2766 char *errstr,
2767 size_t errstr_size);
2768
2775RD_EXPORT
2776void rd_kafka_topic_conf_set_opaque(rd_kafka_topic_conf_t *conf,
2777 void *rkt_opaque);
2778
2779
2799RD_EXPORT
2801 rd_kafka_topic_conf_t *topic_conf,
2802 int32_t (*partitioner)(const rd_kafka_topic_t *rkt,
2803 const void *keydata,
2804 size_t keylen,
2805 int32_t partition_cnt,
2806 void *rkt_opaque,
2807 void *msg_opaque));
2808
2809
2836 rd_kafka_topic_conf_t *topic_conf,
2837 int (*msg_order_cmp)(const rd_kafka_message_t *a,
2838 const rd_kafka_message_t *b));
2839
2840
2848RD_EXPORT
2849int rd_kafka_topic_partition_available(const rd_kafka_topic_t *rkt,
2850 int32_t partition);
2851
2852
2853/*******************************************************************
2854 * *
2855 * Partitioners provided by rdkafka *
2856 * *
2857 *******************************************************************/
2858
2872RD_EXPORT
2873int32_t rd_kafka_msg_partitioner_random(const rd_kafka_topic_t *rkt,
2874 const void *key,
2875 size_t keylen,
2876 int32_t partition_cnt,
2877 void *rkt_opaque,
2878 void *msg_opaque);
2879
2893RD_EXPORT
2894int32_t rd_kafka_msg_partitioner_consistent(const rd_kafka_topic_t *rkt,
2895 const void *key,
2896 size_t keylen,
2897 int32_t partition_cnt,
2898 void *rkt_opaque,
2899 void *msg_opaque);
2900
2916RD_EXPORT
2917int32_t rd_kafka_msg_partitioner_consistent_random(const rd_kafka_topic_t *rkt,
2918 const void *key,
2919 size_t keylen,
2920 int32_t partition_cnt,
2921 void *rkt_opaque,
2922 void *msg_opaque);
2923
2924
2938RD_EXPORT
2939int32_t rd_kafka_msg_partitioner_murmur2(const rd_kafka_topic_t *rkt,
2940 const void *key,
2941 size_t keylen,
2942 int32_t partition_cnt,
2943 void *rkt_opaque,
2944 void *msg_opaque);
2945
2960RD_EXPORT
2961int32_t rd_kafka_msg_partitioner_murmur2_random(const rd_kafka_topic_t *rkt,
2962 const void *key,
2963 size_t keylen,
2964 int32_t partition_cnt,
2965 void *rkt_opaque,
2966 void *msg_opaque);
2967
2968
2982RD_EXPORT
2983int32_t rd_kafka_msg_partitioner_fnv1a(const rd_kafka_topic_t *rkt,
2984 const void *key,
2985 size_t keylen,
2986 int32_t partition_cnt,
2987 void *rkt_opaque,
2988 void *msg_opaque);
2989
2990
3005RD_EXPORT
3006int32_t rd_kafka_msg_partitioner_fnv1a_random(const rd_kafka_topic_t *rkt,
3007 const void *key,
3008 size_t keylen,
3009 int32_t partition_cnt,
3010 void *rkt_opaque,
3011 void *msg_opaque);
3012
3013
3053RD_EXPORT
3055 rd_kafka_conf_t *conf,
3056 char *errstr,
3057 size_t errstr_size);
3058
3059
3074RD_EXPORT
3075void rd_kafka_destroy(rd_kafka_t *rk);
3076
3077
3082RD_EXPORT
3083void rd_kafka_destroy_flags(rd_kafka_t *rk, int flags);
3084
3099#define RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE 0x8
3100
3101
3102
3106RD_EXPORT
3107const char *rd_kafka_name(const rd_kafka_t *rk);
3108
3109
3113RD_EXPORT
3114rd_kafka_type_t rd_kafka_type(const rd_kafka_t *rk);
3115
3116
3127RD_EXPORT
3128char *rd_kafka_memberid(const rd_kafka_t *rk);
3129
3130
3131
3150RD_EXPORT
3151char *rd_kafka_clusterid(rd_kafka_t *rk, int timeout_ms);
3152
3153
3169RD_EXPORT
3170int32_t rd_kafka_controllerid(rd_kafka_t *rk, int timeout_ms);
3171
3172
3194RD_EXPORT
3195rd_kafka_topic_t *rd_kafka_topic_new(rd_kafka_t *rk,
3196 const char *topic,
3197 rd_kafka_topic_conf_t *conf);
3198
3199
3200
3209RD_EXPORT
3210void rd_kafka_topic_destroy(rd_kafka_topic_t *rkt);
3211
3212
3216RD_EXPORT
3217const char *rd_kafka_topic_name(const rd_kafka_topic_t *rkt);
3218
3219
3224RD_EXPORT
3225void *rd_kafka_topic_opaque(const rd_kafka_topic_t *rkt);
3226
3227
3234#define RD_KAFKA_PARTITION_UA ((int32_t) - 1)
3235
3236
3264RD_EXPORT
3265int rd_kafka_poll(rd_kafka_t *rk, int timeout_ms);
3266
3267
3278RD_EXPORT
3279void rd_kafka_yield(rd_kafka_t *rk);
3280
3281
3282
3290RD_EXPORT rd_kafka_resp_err_t
3293
3294
3295
3303RD_EXPORT rd_kafka_resp_err_t
3306
3307
3308
3317RD_EXPORT rd_kafka_resp_err_t
3319 const char *topic,
3320 int32_t partition,
3321 int64_t *low,
3322 int64_t *high,
3323 int timeout_ms);
3324
3325
3343 const char *topic,
3344 int32_t partition,
3345 int64_t *low,
3346 int64_t *high);
3347
3348
3349
3375RD_EXPORT rd_kafka_resp_err_t
3378 int timeout_ms);
3379
3380
3381
3395RD_EXPORT
3396void *rd_kafka_mem_calloc(rd_kafka_t *rk, size_t num, size_t size);
3397
3398
3399
3413RD_EXPORT
3414void *rd_kafka_mem_malloc(rd_kafka_t *rk, size_t size);
3415
3416
3417
3434RD_EXPORT
3435void rd_kafka_mem_free(rd_kafka_t *rk, void *ptr);
3436
3437
3459RD_EXPORT
3460rd_kafka_queue_t *rd_kafka_queue_new(rd_kafka_t *rk);
3461
3465RD_EXPORT
3466void rd_kafka_queue_destroy(rd_kafka_queue_t *rkqu);
3467
3468
3475RD_EXPORT
3476rd_kafka_queue_t *rd_kafka_queue_get_main(rd_kafka_t *rk);
3477
3478
3479
3489RD_EXPORT
3490rd_kafka_queue_t *rd_kafka_queue_get_sasl(rd_kafka_t *rk);
3491
3492
3508RD_EXPORT
3509rd_kafka_error_t *rd_kafka_sasl_background_callbacks_enable(rd_kafka_t *rk);
3510
3511
3526RD_EXPORT
3527rd_kafka_error_t *rd_kafka_sasl_set_credentials(rd_kafka_t *rk,
3528 const char *username,
3529 const char *password);
3530
3546RD_EXPORT
3547rd_kafka_queue_t *rd_kafka_queue_get_consumer(rd_kafka_t *rk);
3548
3559RD_EXPORT
3560rd_kafka_queue_t *rd_kafka_queue_get_partition(rd_kafka_t *rk,
3561 const char *topic,
3562 int32_t partition);
3563
3586RD_EXPORT
3587rd_kafka_queue_t *rd_kafka_queue_get_background(rd_kafka_t *rk);
3588
3589
3600RD_EXPORT
3601void rd_kafka_queue_forward(rd_kafka_queue_t *src, rd_kafka_queue_t *dst);
3602
3621RD_EXPORT
3623 rd_kafka_queue_t *rkqu);
3624
3625
3629RD_EXPORT
3630size_t rd_kafka_queue_length(rd_kafka_queue_t *rkqu);
3631
3632
3650RD_EXPORT
3651void rd_kafka_queue_io_event_enable(rd_kafka_queue_t *rkqu,
3652 int fd,
3653 const void *payload,
3654 size_t size);
3655
3672RD_EXPORT
3673void rd_kafka_queue_cb_event_enable(rd_kafka_queue_t *rkqu,
3674 void (*event_cb)(rd_kafka_t *rk,
3675 void *qev_opaque),
3676 void *qev_opaque);
3677
3678
3686RD_EXPORT
3687void rd_kafka_queue_yield(rd_kafka_queue_t *rkqu);
3688
3689
3700#define RD_KAFKA_OFFSET_BEGINNING \
3701 -2
3703#define RD_KAFKA_OFFSET_END \
3704 -1
3706#define RD_KAFKA_OFFSET_STORED \
3707 -1000
3709#define RD_KAFKA_OFFSET_INVALID -1001
3713#define RD_KAFKA_OFFSET_TAIL_BASE -2000 /* internal: do not use */
3721#define RD_KAFKA_OFFSET_TAIL(CNT) (RD_KAFKA_OFFSET_TAIL_BASE - (CNT))
3722
3756RD_EXPORT
3757int rd_kafka_consume_start(rd_kafka_topic_t *rkt,
3758 int32_t partition,
3759 int64_t offset);
3760
3775RD_EXPORT
3776int rd_kafka_consume_start_queue(rd_kafka_topic_t *rkt,
3777 int32_t partition,
3778 int64_t offset,
3779 rd_kafka_queue_t *rkqu);
3780
3794RD_EXPORT
3795int rd_kafka_consume_stop(rd_kafka_topic_t *rkt, int32_t partition);
3796
3797
3798
3827RD_EXPORT
3829 int32_t partition,
3830 int64_t offset,
3831 int timeout_ms);
3832
3833
3834
3866RD_EXPORT rd_kafka_error_t *
3869 int timeout_ms);
3870
3871
3896RD_EXPORT
3898rd_kafka_consume(rd_kafka_topic_t *rkt, int32_t partition, int timeout_ms);
3899
3900
3901
3927RD_EXPORT
3928ssize_t rd_kafka_consume_batch(rd_kafka_topic_t *rkt,
3929 int32_t partition,
3930 int timeout_ms,
3931 rd_kafka_message_t **rkmessages,
3932 size_t rkmessages_size);
3933
3934
3935
3969RD_EXPORT
3970int rd_kafka_consume_callback(rd_kafka_topic_t *rkt,
3971 int32_t partition,
3972 int timeout_ms,
3973 void (*consume_cb)(rd_kafka_message_t *rkmessage,
3974 void *commit_opaque),
3975 void *commit_opaque);
3976
3977
3996RD_EXPORT
3998 int timeout_ms);
3999
4005RD_EXPORT
4006ssize_t rd_kafka_consume_batch_queue(rd_kafka_queue_t *rkqu,
4007 int timeout_ms,
4008 rd_kafka_message_t **rkmessages,
4009 size_t rkmessages_size);
4010
4020RD_EXPORT
4022 rd_kafka_queue_t *rkqu,
4023 int timeout_ms,
4024 void (*consume_cb)(rd_kafka_message_t *rkmessage, void *commit_opaque),
4025 void *commit_opaque);
4026
4027
4067RD_EXPORT
4069rd_kafka_offset_store(rd_kafka_topic_t *rkt, int32_t partition, int64_t offset);
4070
4071
4104RD_EXPORT rd_kafka_resp_err_t
4107
4108
4129RD_EXPORT
4131
4185RD_EXPORT rd_kafka_resp_err_t
4186rd_kafka_subscribe(rd_kafka_t *rk,
4187 const rd_kafka_topic_partition_list_t *topics);
4188
4189
4193RD_EXPORT
4195
4196
4206RD_EXPORT rd_kafka_resp_err_t
4208
4209
4210
4240RD_EXPORT
4241rd_kafka_message_t *rd_kafka_consumer_poll(rd_kafka_t *rk, int timeout_ms);
4242
4260RD_EXPORT
4262
4263
4281RD_EXPORT
4282rd_kafka_error_t *rd_kafka_consumer_close_queue(rd_kafka_t *rk,
4283 rd_kafka_queue_t *rkqu);
4284
4285
4294RD_EXPORT
4295int rd_kafka_consumer_closed(rd_kafka_t *rk);
4296
4297
4315RD_EXPORT rd_kafka_error_t *
4317 const rd_kafka_topic_partition_list_t *partitions);
4318
4319
4337RD_EXPORT rd_kafka_error_t *rd_kafka_incremental_unassign(
4338 rd_kafka_t *rk,
4339 const rd_kafka_topic_partition_list_t *partitions);
4340
4341
4355RD_EXPORT
4356const char *rd_kafka_rebalance_protocol(rd_kafka_t *rk);
4357
4358
4379RD_EXPORT rd_kafka_resp_err_t
4380rd_kafka_assign(rd_kafka_t *rk,
4381 const rd_kafka_topic_partition_list_t *partitions);
4382
4399RD_EXPORT rd_kafka_resp_err_t
4400rd_kafka_assignment(rd_kafka_t *rk,
4401 rd_kafka_topic_partition_list_t **partitions);
4402
4403
4420RD_EXPORT int rd_kafka_assignment_lost(rd_kafka_t *rk);
4421
4422
4458RD_EXPORT rd_kafka_resp_err_t
4459rd_kafka_commit(rd_kafka_t *rk,
4460 const rd_kafka_topic_partition_list_t *offsets,
4461 int async);
4462
4463
4470RD_EXPORT rd_kafka_resp_err_t
4472 const rd_kafka_message_t *rkmessage,
4473 int async);
4474
4475
4498RD_EXPORT rd_kafka_resp_err_t
4500 const rd_kafka_topic_partition_list_t *offsets,
4501 rd_kafka_queue_t *rkqu,
4502 void (*cb)(rd_kafka_t *rk,
4505 void *commit_opaque),
4506 void *commit_opaque);
4507
4508
4526RD_EXPORT rd_kafka_resp_err_t
4527rd_kafka_committed(rd_kafka_t *rk,
4529 int timeout_ms);
4530
4531
4532
4549RD_EXPORT rd_kafka_resp_err_t
4551
4552
4553
4565RD_EXPORT rd_kafka_consumer_group_metadata_t *
4567
4568
4578RD_EXPORT rd_kafka_consumer_group_metadata_t *
4580
4581
4594RD_EXPORT rd_kafka_consumer_group_metadata_t *
4596 int32_t generation_id,
4597 const char *member_id,
4598 const char *group_instance_id);
4599
4609RD_EXPORT
4611 const rd_kafka_consumer_group_metadata_t *group_metadata);
4612
4623RD_EXPORT
4625 const rd_kafka_consumer_group_metadata_t *group_metadata);
4626
4636RD_EXPORT
4638 const rd_kafka_consumer_group_metadata_t *group_metadata);
4639
4649RD_EXPORT
4651 const rd_kafka_consumer_group_metadata_t *group_metadata);
4652
4657RD_EXPORT void
4658rd_kafka_consumer_group_metadata_destroy(rd_kafka_consumer_group_metadata_t *);
4659
4660
4681RD_EXPORT rd_kafka_error_t *rd_kafka_consumer_group_metadata_write(
4682 const rd_kafka_consumer_group_metadata_t *cgmd,
4683 void **bufferp,
4684 size_t *sizep);
4685
4706RD_EXPORT rd_kafka_error_t *rd_kafka_consumer_group_metadata_read(
4707 rd_kafka_consumer_group_metadata_t **cgmdp,
4708 const void *buffer,
4709 size_t size);
4710
4726#define RD_KAFKA_MSG_F_FREE \
4727 0x1
4729#define RD_KAFKA_MSG_F_COPY \
4730 0x2
4732#define RD_KAFKA_MSG_F_BLOCK \
4733 0x4
4744#define RD_KAFKA_MSG_F_PARTITION \
4745 0x8
4850RD_EXPORT
4851int rd_kafka_produce(rd_kafka_topic_t *rkt,
4852 int32_t partition,
4853 int msgflags,
4854 void *payload,
4855 size_t len,
4856 const void *key,
4857 size_t keylen,
4858 void *msg_opaque);
4859
4860
4874RD_EXPORT
4876
4877
4889RD_EXPORT
4890rd_kafka_error_t *
4891rd_kafka_produceva(rd_kafka_t *rk, const rd_kafka_vu_t *vus, size_t cnt);
4892
4893
4923RD_EXPORT
4924int rd_kafka_produce_batch(rd_kafka_topic_t *rkt,
4925 int32_t partition,
4926 int msgflags,
4927 rd_kafka_message_t *rkmessages,
4928 int message_cnt);
4929
4930
4931
4956RD_EXPORT
4957rd_kafka_resp_err_t rd_kafka_flush(rd_kafka_t *rk, int timeout_ms);
4958
4959
4960
4991RD_EXPORT
4992rd_kafka_resp_err_t rd_kafka_purge(rd_kafka_t *rk, int purge_flags);
4993
4994
5002#define RD_KAFKA_PURGE_F_QUEUE 0x1
5003
5011#define RD_KAFKA_PURGE_F_INFLIGHT 0x2
5012
5013
5017#define RD_KAFKA_PURGE_F_NON_BLOCKING 0x4
5018
5019
5034typedef struct rd_kafka_metadata_broker {
5035 int32_t id;
5036 char *host;
5037 int port;
5039
5043typedef struct rd_kafka_metadata_partition {
5044 int32_t id;
5046 int32_t leader;
5048 int32_t *replicas;
5050 int32_t *isrs;
5052
5056typedef struct rd_kafka_metadata_topic {
5057 char *topic;
5059 struct rd_kafka_metadata_partition *partitions;
5062
5063
5067typedef struct rd_kafka_metadata {
5069 struct rd_kafka_metadata_broker *brokers;
5072 struct rd_kafka_metadata_topic *topics;
5077
5098RD_EXPORT
5100rd_kafka_metadata(rd_kafka_t *rk,
5101 int all_topics,
5102 rd_kafka_topic_t *only_rkt,
5103 const struct rd_kafka_metadata **metadatap,
5104 int timeout_ms);
5105
5109RD_EXPORT
5111
5115typedef struct rd_kafka_Node_s rd_kafka_Node_t;
5116
5124RD_EXPORT
5126
5137RD_EXPORT
5138const char *rd_kafka_Node_host(const rd_kafka_Node_t *node);
5139
5147RD_EXPORT
5149
5157RD_EXPORT
5158const char *rd_kafka_Node_rack(const rd_kafka_Node_t *node);
5159
5190
5196typedef enum {
5197 RD_KAFKA_CONSUMER_GROUP_STATE_UNKNOWN = 0,
5198 RD_KAFKA_CONSUMER_GROUP_STATE_PREPARING_REBALANCE = 1,
5199 RD_KAFKA_CONSUMER_GROUP_STATE_COMPLETING_REBALANCE = 2,
5200 RD_KAFKA_CONSUMER_GROUP_STATE_STABLE = 3,
5201 RD_KAFKA_CONSUMER_GROUP_STATE_DEAD = 4,
5202 RD_KAFKA_CONSUMER_GROUP_STATE_EMPTY = 5,
5203 RD_KAFKA_CONSUMER_GROUP_STATE__CNT
5205
5211typedef enum {
5212 RD_KAFKA_CONSUMER_GROUP_TYPE_UNKNOWN = 0,
5213 RD_KAFKA_CONSUMER_GROUP_TYPE_CONSUMER = 1,
5214 RD_KAFKA_CONSUMER_GROUP_TYPE_CLASSIC = 2,
5215 RD_KAFKA_CONSUMER_GROUP_TYPE__CNT
5217
5231
5241
5242
5273RD_EXPORT
5276 const char *group,
5277 const struct rd_kafka_group_list **grplistp,
5278 int timeout_ms);
5279
5287RD_EXPORT
5288const char *
5290
5298RD_EXPORT
5301
5309RD_EXPORT
5310const char *
5312
5322RD_EXPORT
5325
5329RD_EXPORT
5331
5332
5376RD_EXPORT
5377int rd_kafka_brokers_add(rd_kafka_t *rk, const char *brokerlist);
5378
5389RD_EXPORT
5390int32_t *rd_kafka_brokers_learned_ids(rd_kafka_t *rk, size_t *cntp);
5391
5392
5405RD_EXPORT RD_DEPRECATED void
5406rd_kafka_set_logger(rd_kafka_t *rk,
5407 void (*func)(const rd_kafka_t *rk,
5408 int level,
5409 const char *fac,
5410 const char *buf));
5411
5412
5422RD_EXPORT
5423void rd_kafka_set_log_level(rd_kafka_t *rk, int level);
5424
5425
5429RD_EXPORT
5430void rd_kafka_log_print(const rd_kafka_t *rk,
5431 int level,
5432 const char *fac,
5433 const char *buf);
5434
5435
5441RD_EXPORT
5442void rd_kafka_log_syslog(const rd_kafka_t *rk,
5443 int level,
5444 const char *fac,
5445 const char *buf);
5446
5447
5470RD_EXPORT
5471int rd_kafka_outq_len(rd_kafka_t *rk);
5472
5473
5474
5481RD_EXPORT
5482void rd_kafka_dump(FILE *fp, rd_kafka_t *rk);
5483
5484
5485
5491RD_EXPORT
5493
5494
5507
5508
5517RD_EXPORT
5518int rd_kafka_wait_destroyed(int timeout_ms);
5519
5520
5526RD_EXPORT
5528
5529
5546RD_EXPORT
5548
5549
5566#define RD_KAFKA_EVENT_NONE 0x0
5567#define RD_KAFKA_EVENT_DR 0x1
5568#define RD_KAFKA_EVENT_FETCH 0x2
5569#define RD_KAFKA_EVENT_LOG 0x4
5570#define RD_KAFKA_EVENT_ERROR 0x8
5571#define RD_KAFKA_EVENT_REBALANCE 0x10
5572#define RD_KAFKA_EVENT_OFFSET_COMMIT 0x20
5573#define RD_KAFKA_EVENT_STATS 0x40
5574#define RD_KAFKA_EVENT_CREATETOPICS_RESULT 100
5575#define RD_KAFKA_EVENT_DELETETOPICS_RESULT 101
5576#define RD_KAFKA_EVENT_CREATEPARTITIONS_RESULT \
5577 102
5578#define RD_KAFKA_EVENT_ALTERCONFIGS_RESULT 103
5579#define RD_KAFKA_EVENT_DESCRIBECONFIGS_RESULT \
5580 104
5581#define RD_KAFKA_EVENT_DELETERECORDS_RESULT 105
5582#define RD_KAFKA_EVENT_DELETEGROUPS_RESULT 106
5584#define RD_KAFKA_EVENT_DELETECONSUMERGROUPOFFSETS_RESULT 107
5586#define RD_KAFKA_EVENT_OAUTHBEARER_TOKEN_REFRESH 0x100
5587#define RD_KAFKA_EVENT_BACKGROUND 0x200
5588#define RD_KAFKA_EVENT_CREATEACLS_RESULT 0x400
5589#define RD_KAFKA_EVENT_DESCRIBEACLS_RESULT 0x800
5590#define RD_KAFKA_EVENT_DELETEACLS_RESULT 0x1000
5592#define RD_KAFKA_EVENT_LISTCONSUMERGROUPS_RESULT 0x2000
5594#define RD_KAFKA_EVENT_DESCRIBECONSUMERGROUPS_RESULT 0x4000
5596#define RD_KAFKA_EVENT_LISTCONSUMERGROUPOFFSETS_RESULT 0x8000
5598#define RD_KAFKA_EVENT_ALTERCONSUMERGROUPOFFSETS_RESULT 0x10000
5600#define RD_KAFKA_EVENT_INCREMENTALALTERCONFIGS_RESULT 0x20000
5602#define RD_KAFKA_EVENT_DESCRIBEUSERSCRAMCREDENTIALS_RESULT 0x40000
5604#define RD_KAFKA_EVENT_ALTERUSERSCRAMCREDENTIALS_RESULT 0x80000
5606#define RD_KAFKA_EVENT_DESCRIBETOPICS_RESULT 0x100000
5608#define RD_KAFKA_EVENT_DESCRIBECLUSTER_RESULT 0x200000
5610#define RD_KAFKA_EVENT_LISTOFFSETS_RESULT 0x400000
5612#define RD_KAFKA_EVENT_ELECTLEADERS_RESULT 0x800000
5613
5620RD_EXPORT
5621rd_kafka_event_type_t rd_kafka_event_type(const rd_kafka_event_t *rkev);
5622
5629RD_EXPORT
5630const char *rd_kafka_event_name(const rd_kafka_event_t *rkev);
5631
5632
5642RD_EXPORT
5643void rd_kafka_event_destroy(rd_kafka_event_t *rkev);
5644
5645
5661RD_EXPORT
5663
5664
5678RD_EXPORT
5679size_t rd_kafka_event_message_array(rd_kafka_event_t *rkev,
5680 const rd_kafka_message_t **rkmessages,
5681 size_t size);
5682
5683
5691RD_EXPORT
5692size_t rd_kafka_event_message_count(rd_kafka_event_t *rkev);
5693
5694
5706RD_EXPORT
5707const char *rd_kafka_event_config_string(rd_kafka_event_t *rkev);
5708
5709
5718RD_EXPORT
5720
5721
5730RD_EXPORT
5731const char *rd_kafka_event_error_string(rd_kafka_event_t *rkev);
5732
5733
5742RD_EXPORT
5743int rd_kafka_event_error_is_fatal(rd_kafka_event_t *rkev);
5744
5745
5773RD_EXPORT
5774void *rd_kafka_event_opaque(rd_kafka_event_t *rkev);
5775
5776
5785RD_EXPORT
5786int rd_kafka_event_log(rd_kafka_event_t *rkev,
5787 const char **fac,
5788 const char **str,
5789 int *level);
5790
5791
5803RD_EXPORT
5804int rd_kafka_event_debug_contexts(rd_kafka_event_t *rkev,
5805 char *dst,
5806 size_t dstsize);
5807
5808
5821RD_EXPORT
5822const char *rd_kafka_event_stats(rd_kafka_event_t *rkev);
5823
5824
5837
5838
5850rd_kafka_event_topic_partition(rd_kafka_event_t *rkev);
5851
5852
5854typedef rd_kafka_event_t rd_kafka_CreateTopics_result_t;
5856typedef rd_kafka_event_t rd_kafka_DeleteTopics_result_t;
5858typedef rd_kafka_event_t rd_kafka_CreateAcls_result_t;
5860typedef rd_kafka_event_t rd_kafka_DescribeAcls_result_t;
5862typedef rd_kafka_event_t rd_kafka_DeleteAcls_result_t;
5864typedef rd_kafka_event_t rd_kafka_CreatePartitions_result_t;
5866typedef rd_kafka_event_t rd_kafka_AlterConfigs_result_t;
5870typedef rd_kafka_event_t rd_kafka_DescribeConfigs_result_t;
5872typedef rd_kafka_event_t rd_kafka_DeleteRecords_result_t;
5878typedef rd_kafka_event_t rd_kafka_DeleteGroups_result_t;
5886typedef rd_kafka_event_t rd_kafka_DescribeTopics_result_t;
5888typedef rd_kafka_event_t rd_kafka_DescribeCluster_result_t;
5894typedef rd_kafka_event_t rd_kafka_ListOffsets_result_t;
5896typedef rd_kafka_event_t rd_kafka_ElectLeaders_result_t;
5897
5907RD_EXPORT const rd_kafka_CreateTopics_result_t *
5909
5919RD_EXPORT const rd_kafka_DeleteTopics_result_t *
5921
5931RD_EXPORT const rd_kafka_CreatePartitions_result_t *
5933
5943RD_EXPORT const rd_kafka_AlterConfigs_result_t *
5945
5957
5967RD_EXPORT const rd_kafka_DescribeConfigs_result_t *
5969
5977RD_EXPORT const rd_kafka_DeleteRecords_result_t *
5979
5994
6009
6022RD_EXPORT const rd_kafka_DescribeTopics_result_t *
6024
6037RD_EXPORT const rd_kafka_DescribeCluster_result_t *
6048RD_EXPORT const rd_kafka_DeleteGroups_result_t *
6050
6062
6070RD_EXPORT const rd_kafka_CreateAcls_result_t *
6071rd_kafka_event_CreateAcls_result(rd_kafka_event_t *rkev);
6072
6080RD_EXPORT const rd_kafka_DescribeAcls_result_t *
6082
6090RD_EXPORT const rd_kafka_DeleteAcls_result_t *
6091rd_kafka_event_DeleteAcls_result(rd_kafka_event_t *rkev);
6092
6107
6122
6135RD_EXPORT const rd_kafka_ListOffsets_result_t *
6137
6138
6153
6168
6181RD_EXPORT const rd_kafka_ElectLeaders_result_t *
6183
6193RD_EXPORT
6194rd_kafka_event_t *rd_kafka_queue_poll(rd_kafka_queue_t *rkqu, int timeout_ms);
6195
6210RD_EXPORT
6211int rd_kafka_queue_poll_callback(rd_kafka_queue_t *rkqu, int timeout_ms);
6212
6213
6258 rd_kafka_conf_t *conf,
6259 void **plug_opaquep,
6260 char *errstr,
6261 size_t errstr_size);
6262
6344 rd_kafka_conf_t *conf,
6345 const char *name,
6346 const char *val,
6347 char *errstr,
6348 size_t errstr_size,
6349 void *ic_opaque);
6350
6351
6374 rd_kafka_conf_t *new_conf,
6375 const rd_kafka_conf_t *old_conf,
6376 size_t filter_cnt,
6377 const char **filter,
6378 void *ic_opaque);
6379
6380
6388 void *ic_opaque);
6389
6390
6409 rd_kafka_t *rk,
6410 const rd_kafka_conf_t *conf,
6411 void *ic_opaque,
6412 char *errstr,
6413 size_t errstr_size);
6414
6415
6424 rd_kafka_interceptor_f_on_destroy_t)(rd_kafka_t *rk, void *ic_opaque);
6425
6426
6427
6449 rd_kafka_t *rk,
6450 rd_kafka_message_t *rkmessage,
6451 void *ic_opaque);
6452
6476 rd_kafka_t *rk,
6477 rd_kafka_message_t *rkmessage,
6478 void *ic_opaque);
6479
6480
6498 rd_kafka_t *rk,
6499 rd_kafka_message_t *rkmessage,
6500 void *ic_opaque);
6501
6524 rd_kafka_t *rk,
6525 const rd_kafka_topic_partition_list_t *offsets,
6527 void *ic_opaque);
6528
6529
6552 rd_kafka_t *rk,
6553 int sockfd,
6554 const char *brokername,
6555 int32_t brokerid,
6556 int16_t ApiKey,
6557 int16_t ApiVersion,
6558 int32_t CorrId,
6559 size_t size,
6560 void *ic_opaque);
6561
6562
6589 rd_kafka_t *rk,
6590 int sockfd,
6591 const char *brokername,
6592 int32_t brokerid,
6593 int16_t ApiKey,
6594 int16_t ApiVersion,
6595 int32_t CorrId,
6596 size_t size,
6597 int64_t rtt,
6599 void *ic_opaque);
6600
6601
6619 rd_kafka_t *rk,
6620 rd_kafka_thread_type_t thread_type,
6621 const char *thread_name,
6622 void *ic_opaque);
6623
6624
6645 rd_kafka_t *rk,
6646 rd_kafka_thread_type_t thread_type,
6647 const char *thread_name,
6648 void *ic_opaque);
6649
6650
6666 rd_kafka_t *rk,
6667 int32_t broker_id,
6668 const char *secproto,
6669 const char *name,
6670 int port,
6671 const char *state,
6672 void *ic_opaque);
6673
6674
6688 rd_kafka_conf_t *conf,
6689 const char *ic_name,
6691 void *ic_opaque);
6692
6693
6707 rd_kafka_conf_t *conf,
6708 const char *ic_name,
6710 void *ic_opaque);
6711
6726 rd_kafka_conf_t *conf,
6727 const char *ic_name,
6729 void *ic_opaque);
6730
6731
6753RD_EXPORT rd_kafka_resp_err_t
6755 const char *ic_name,
6757 void *ic_opaque);
6758
6759
6760
6774 rd_kafka_t *rk,
6775 const char *ic_name,
6777 void *ic_opaque);
6778
6779
6792RD_EXPORT rd_kafka_resp_err_t
6794 const char *ic_name,
6796 void *ic_opaque);
6797
6811 rd_kafka_t *rk,
6812 const char *ic_name,
6814 void *ic_opaque);
6815
6816
6830 rd_kafka_t *rk,
6831 const char *ic_name,
6833 void *ic_opaque);
6834
6835
6849 rd_kafka_t *rk,
6850 const char *ic_name,
6852 void *ic_opaque);
6853
6854
6868 rd_kafka_t *rk,
6869 const char *ic_name,
6871 void *ic_opaque);
6872
6873
6887 rd_kafka_t *rk,
6888 const char *ic_name,
6890 void *ic_opaque);
6891
6892
6906 rd_kafka_t *rk,
6907 const char *ic_name,
6909 void *ic_opaque);
6910
6911
6925 rd_kafka_t *rk,
6926 const char *ic_name,
6928 void *ic_opaque);
6929
6930
6943RD_EXPORT
6945 rd_kafka_t *rk,
6946 const char *ic_name,
6948 void *ic_opaque);
6949
6950
6951
6972RD_EXPORT rd_kafka_resp_err_t
6973rd_kafka_topic_result_error(const rd_kafka_topic_result_t *topicres);
6974
6981RD_EXPORT const char *
6982rd_kafka_topic_result_error_string(const rd_kafka_topic_result_t *topicres);
6983
6989RD_EXPORT const char *
6990rd_kafka_topic_result_name(const rd_kafka_topic_result_t *topicres);
6991
7001RD_EXPORT const rd_kafka_error_t *
7002rd_kafka_group_result_error(const rd_kafka_group_result_t *groupres);
7003
7009RD_EXPORT const char *
7010rd_kafka_group_result_name(const rd_kafka_group_result_t *groupres);
7011
7017RD_EXPORT const rd_kafka_topic_partition_list_t *
7018rd_kafka_group_result_partitions(const rd_kafka_group_result_t *groupres);
7019
7033RD_EXPORT const rd_kafka_topic_partition_t *
7035 const rd_kafka_topic_partition_result_t *partition_result);
7036
7042RD_EXPORT const rd_kafka_error_t *rd_kafka_topic_partition_result_error(
7043 const rd_kafka_topic_partition_result_t *partition_result);
7044
7123
7136typedef struct rd_kafka_AdminOptions_s rd_kafka_AdminOptions_t;
7137
7146 RD_KAFKA_ISOLATION_LEVEL_READ_UNCOMMITTED = 0,
7147 RD_KAFKA_ISOLATION_LEVEL_READ_COMMITTED = 1
7149
7169RD_EXPORT rd_kafka_AdminOptions_t *
7171
7172
7177
7178
7196RD_EXPORT rd_kafka_resp_err_t
7198 int timeout_ms,
7199 char *errstr,
7200 size_t errstr_size);
7201
7202
7229RD_EXPORT rd_kafka_resp_err_t
7231 int timeout_ms,
7232 char *errstr,
7233 size_t errstr_size);
7234
7235
7254RD_EXPORT rd_kafka_resp_err_t
7256 int true_or_false,
7257 char *errstr,
7258 size_t errstr_size);
7259
7260
7288RD_EXPORT rd_kafka_resp_err_t
7290 int32_t broker_id,
7291 char *errstr,
7292 size_t errstr_size);
7293
7294
7307RD_EXPORT
7309 rd_kafka_AdminOptions_t *options,
7310 int true_or_false);
7311
7326RD_EXPORT
7328 rd_kafka_AdminOptions_t *options,
7329 int true_or_false);
7330
7343RD_EXPORT
7345 rd_kafka_AdminOptions_t *options,
7346 const rd_kafka_consumer_group_state_t *consumer_group_states,
7347 size_t consumer_group_states_cnt);
7348
7361RD_EXPORT
7363 rd_kafka_AdminOptions_t *options,
7364 const rd_kafka_consumer_group_type_t *consumer_group_types,
7365 size_t consumer_group_types_cnt);
7366
7370RD_EXPORT
7371rd_kafka_error_t *
7374
7379RD_EXPORT void
7381 void *ev_opaque);
7382
7383
7384
7411
7423typedef struct rd_kafka_NewTopic_s rd_kafka_NewTopic_t;
7424
7446RD_EXPORT rd_kafka_NewTopic_t *rd_kafka_NewTopic_new(const char *topic,
7447 int num_partitions,
7448 int replication_factor,
7449 char *errstr,
7450 size_t errstr_size);
7451
7457
7458
7465 size_t new_topic_cnt);
7466
7467
7489RD_EXPORT rd_kafka_resp_err_t
7491 int32_t partition,
7492 int32_t *broker_ids,
7493 size_t broker_id_cnt,
7494 char *errstr,
7495 size_t errstr_size);
7496
7509RD_EXPORT rd_kafka_resp_err_t
7511 const char *name,
7512 const char *value);
7513
7514
7533RD_EXPORT void rd_kafka_CreateTopics(rd_kafka_t *rk,
7534 rd_kafka_NewTopic_t **new_topics,
7535 size_t new_topic_cnt,
7536 const rd_kafka_AdminOptions_t *options,
7537 rd_kafka_queue_t *rkqu);
7538
7539
7540/*
7541 * CreateTopics result type and methods
7542 */
7543
7552RD_EXPORT const rd_kafka_topic_result_t **rd_kafka_CreateTopics_result_topics(
7553 const rd_kafka_CreateTopics_result_t *result,
7554 size_t *cntp);
7555
7556
7557
7558/*
7559 * DeleteTopics - delete topics from cluster
7560 *
7561 */
7562
7564typedef struct rd_kafka_DeleteTopic_s rd_kafka_DeleteTopic_t;
7565
7576
7582
7588RD_EXPORT void
7590 size_t del_topic_cnt);
7591
7605RD_EXPORT
7606void rd_kafka_DeleteTopics(rd_kafka_t *rk,
7607 rd_kafka_DeleteTopic_t **del_topics,
7608 size_t del_topic_cnt,
7609 const rd_kafka_AdminOptions_t *options,
7610 rd_kafka_queue_t *rkqu);
7611
7612
7613
7614/*
7615 * DeleteTopics result type and methods
7616 */
7617
7626RD_EXPORT const rd_kafka_topic_result_t **rd_kafka_DeleteTopics_result_topics(
7627 const rd_kafka_DeleteTopics_result_t *result,
7628 size_t *cntp);
7629
7630
7641typedef struct rd_kafka_NewPartitions_s rd_kafka_NewPartitions_t;
7642
7659RD_EXPORT rd_kafka_NewPartitions_t *
7661 size_t new_total_cnt,
7662 char *errstr,
7663 size_t errstr_size);
7664
7669RD_EXPORT void
7671
7677RD_EXPORT void
7679 size_t new_parts_cnt);
7680
7704 rd_kafka_NewPartitions_t *new_parts,
7705 int32_t new_partition_idx,
7706 int32_t *broker_ids,
7707 size_t broker_id_cnt,
7708 char *errstr,
7709 size_t errstr_size);
7710
7711
7730RD_EXPORT void rd_kafka_CreatePartitions(rd_kafka_t *rk,
7731 rd_kafka_NewPartitions_t **new_parts,
7732 size_t new_parts_cnt,
7733 const rd_kafka_AdminOptions_t *options,
7734 rd_kafka_queue_t *rkqu);
7735
7736
7737
7738/*
7739 * CreatePartitions result type and methods
7740 */
7741
7750RD_EXPORT const rd_kafka_topic_result_t **
7753 size_t *cntp);
7754
7796
7797
7801RD_EXPORT const char *
7803
7804
7806typedef struct rd_kafka_ConfigEntry_s rd_kafka_ConfigEntry_t;
7807
7811RD_EXPORT const char *
7813
7818RD_EXPORT const char *
7820
7826
7831RD_EXPORT int
7833
7839RD_EXPORT int
7841
7849RD_EXPORT int
7851
7855RD_EXPORT int
7857
7858
7869RD_EXPORT const rd_kafka_ConfigEntry_t **
7871 size_t *cntp);
7872
7873
7874
7888
7906
7912 RD_KAFKA_ALTER_CONFIG_OP_TYPE_SET = 0,
7913 RD_KAFKA_ALTER_CONFIG_OP_TYPE_DELETE = 1,
7914 RD_KAFKA_ALTER_CONFIG_OP_TYPE_APPEND = 2,
7915 RD_KAFKA_ALTER_CONFIG_OP_TYPE_SUBTRACT = 3,
7916 RD_KAFKA_ALTER_CONFIG_OP_TYPE__CNT,
7918
7923 rd_kafka_ResourcePatternType_t resource_pattern_type);
7924
7928RD_EXPORT const char *
7930
7932typedef struct rd_kafka_ConfigResource_s rd_kafka_ConfigResource_t;
7933
7934
7943RD_EXPORT rd_kafka_ConfigResource_t *
7945 const char *resname);
7946
7951RD_EXPORT void
7953
7954
7960RD_EXPORT void
7962 size_t config_cnt);
7963
7964
7978RD_EXPORT rd_kafka_resp_err_t
7980 const char *name,
7981 const char *value);
7982
7983
8004 const char *name,
8006 const char *value);
8007
8008
8017RD_EXPORT const rd_kafka_ConfigEntry_t **
8019 size_t *cntp);
8020
8021
8022
8028
8032RD_EXPORT const char *
8034
8038RD_EXPORT rd_kafka_resp_err_t
8040
8045RD_EXPORT const char *
8047
8048
8049/*
8050 * AlterConfigs - alter cluster configuration.
8051 *
8052 */
8053
8054
8077RD_EXPORT
8078void rd_kafka_AlterConfigs(rd_kafka_t *rk,
8079 rd_kafka_ConfigResource_t **configs,
8080 size_t config_cnt,
8081 const rd_kafka_AdminOptions_t *options,
8082 rd_kafka_queue_t *rkqu);
8083
8084
8085/*
8086 * AlterConfigs result type and methods
8087 */
8088
8103RD_EXPORT const rd_kafka_ConfigResource_t **
8105 const rd_kafka_AlterConfigs_result_t *result,
8106 size_t *cntp);
8107
8108
8109
8110/*
8111 * IncrementalAlterConfigs - alter cluster configuration incrementally.
8112 *
8113 */
8114
8115
8136RD_EXPORT
8138 rd_kafka_ConfigResource_t **configs,
8139 size_t config_cnt,
8140 const rd_kafka_AdminOptions_t *options,
8141 rd_kafka_queue_t *rkqu);
8142
8143
8144/*
8145 * IncrementalAlterConfigs result type and methods
8146 */
8147
8163RD_EXPORT const rd_kafka_ConfigResource_t **
8166 size_t *cntp);
8167
8168
8169
8170/*
8171 * DescribeConfigs - retrieve cluster configuration.
8172 *
8173 */
8174
8175
8200RD_EXPORT
8201void rd_kafka_DescribeConfigs(rd_kafka_t *rk,
8202 rd_kafka_ConfigResource_t **configs,
8203 size_t config_cnt,
8204 const rd_kafka_AdminOptions_t *options,
8205 rd_kafka_queue_t *rkqu);
8206
8207
8208
8209/*
8210 * DescribeConfigs result type and methods
8211 */
8212
8221RD_EXPORT const rd_kafka_ConfigResource_t **
8224 size_t *cntp);
8225
8226
8237typedef struct rd_kafka_DeleteRecords_s rd_kafka_DeleteRecords_t;
8238
8256 const rd_kafka_topic_partition_list_t *before_offsets);
8257
8262RD_EXPORT void
8264
8270RD_EXPORT void
8272 size_t del_record_cnt);
8273
8295RD_EXPORT void rd_kafka_DeleteRecords(rd_kafka_t *rk,
8296 rd_kafka_DeleteRecords_t **del_records,
8297 size_t del_record_cnt,
8298 const rd_kafka_AdminOptions_t *options,
8299 rd_kafka_queue_t *rkqu);
8300
8301
8302/*
8303 * DeleteRecords result type and methods
8304 */
8305
8315RD_EXPORT const rd_kafka_topic_partition_list_t *
8317 const rd_kafka_DeleteRecords_result_t *result);
8318
8330typedef struct rd_kafka_TopicCollection_s rd_kafka_TopicCollection_t;
8331
8336typedef struct rd_kafka_TopicPartitionInfo_s rd_kafka_TopicPartitionInfo_t;
8337
8342typedef struct rd_kafka_TopicDescription_s rd_kafka_TopicDescription_t;
8343
8353RD_EXPORT
8355rd_kafka_TopicCollection_of_topic_names(const char **topics, size_t topics_cnt);
8356
8361RD_EXPORT void
8363
8378RD_EXPORT
8379void rd_kafka_DescribeTopics(rd_kafka_t *rk,
8380 const rd_kafka_TopicCollection_t *topics,
8381 const rd_kafka_AdminOptions_t *options,
8382 rd_kafka_queue_t *rkqu);
8383
8393RD_EXPORT
8396 size_t *cntp);
8397
8398
8409RD_EXPORT
8411 const rd_kafka_TopicDescription_t *topicdesc,
8412 size_t *cntp);
8413
8414
8422RD_EXPORT
8424 const rd_kafka_TopicPartitionInfo_t *partition);
8425
8426
8437RD_EXPORT
8439 const rd_kafka_TopicPartitionInfo_t *partition);
8440
8452RD_EXPORT
8453const rd_kafka_Node_t **
8455 size_t *cntp);
8456
8468RD_EXPORT
8470 const rd_kafka_TopicPartitionInfo_t *partition,
8471 size_t *cntp);
8472
8485RD_EXPORT
8487 const rd_kafka_TopicDescription_t *topicdesc,
8488 size_t *cntp);
8489
8500RD_EXPORT
8501const char *
8503
8513RD_EXPORT const rd_kafka_Uuid_t *rd_kafka_TopicDescription_topic_id(
8514 const rd_kafka_TopicDescription_t *topicdesc);
8515
8523RD_EXPORT
8525 const rd_kafka_TopicDescription_t *topicdesc);
8526
8537RD_EXPORT
8538const rd_kafka_error_t *
8540
8541
8561RD_EXPORT
8562void rd_kafka_DescribeCluster(rd_kafka_t *rk,
8563 const rd_kafka_AdminOptions_t *options,
8564 rd_kafka_queue_t *rkqu);
8565
8576RD_EXPORT
8579 size_t *cntp);
8580
8592RD_EXPORT
8596 size_t *cntp);
8597
8605RD_EXPORT
8607 const rd_kafka_DescribeCluster_result_t *result);
8608
8618RD_EXPORT
8620 const rd_kafka_DescribeCluster_result_t *result);
8621
8636typedef struct rd_kafka_ConsumerGroupListing_s rd_kafka_ConsumerGroupListing_t;
8637
8639typedef struct rd_kafka_ListConsumerGroupsResult_s
8641
8652RD_EXPORT
8654 const rd_kafka_AdminOptions_t *options,
8655 rd_kafka_queue_t *rkqu);
8656
8667RD_EXPORT
8669 const rd_kafka_ConsumerGroupListing_t *grplist);
8670
8679RD_EXPORT
8681 const rd_kafka_ConsumerGroupListing_t *grplist);
8682
8690RD_EXPORT
8692 const rd_kafka_ConsumerGroupListing_t *grplist);
8693
8701RD_EXPORT
8703 const rd_kafka_ConsumerGroupListing_t *grplist);
8704
8716RD_EXPORT
8720 size_t *cntp);
8721
8735RD_EXPORT
8738 size_t *cntp);
8739
8751typedef struct rd_kafka_ConsumerGroupDescription_s
8753
8758typedef struct rd_kafka_MemberDescription_s rd_kafka_MemberDescription_t;
8759
8764typedef struct rd_kafka_MemberAssignment_s rd_kafka_MemberAssignment_t;
8765
8781RD_EXPORT
8783 const char **groups,
8784 size_t groups_cnt,
8785 const rd_kafka_AdminOptions_t *options,
8786 rd_kafka_queue_t *rkqu);
8787
8799RD_EXPORT
8803 size_t *cntp);
8804
8805
8816RD_EXPORT
8819
8830RD_EXPORT
8833
8841RD_EXPORT
8844
8845
8856RD_EXPORT
8859
8872RD_EXPORT
8876 size_t *cntp);
8877
8885RD_EXPORT
8888
8899RD_EXPORT
8902
8910RD_EXPORT
8913
8921RD_EXPORT
8924
8937RD_EXPORT
8940 size_t idx);
8941
8952RD_EXPORT
8954 const rd_kafka_MemberDescription_t *member);
8955
8966RD_EXPORT
8968 const rd_kafka_MemberDescription_t *member);
8969
8980RD_EXPORT
8982 const rd_kafka_MemberDescription_t *member);
8983
8994RD_EXPORT
8995const char *
8997
9008RD_EXPORT
9010 const rd_kafka_MemberDescription_t *member);
9011
9022RD_EXPORT
9024 const rd_kafka_MemberAssignment_t *assignment);
9025
9037RD_EXPORT
9039 const rd_kafka_MemberDescription_t *member);
9040
9052typedef struct rd_kafka_DeleteGroup_s rd_kafka_DeleteGroup_t;
9053
9063RD_EXPORT
9065
9070RD_EXPORT
9072
9078RD_EXPORT void
9080 size_t del_group_cnt);
9081
9097RD_EXPORT
9098void rd_kafka_DeleteGroups(rd_kafka_t *rk,
9099 rd_kafka_DeleteGroup_t **del_groups,
9100 size_t del_group_cnt,
9101 const rd_kafka_AdminOptions_t *options,
9102 rd_kafka_queue_t *rkqu);
9103
9104
9105
9106/*
9107 * DeleteGroups result type and methods
9108 */
9109
9118RD_EXPORT const rd_kafka_group_result_t **rd_kafka_DeleteGroups_result_groups(
9119 const rd_kafka_DeleteGroups_result_t *result,
9120 size_t *cntp);
9121
9132typedef struct rd_kafka_ListConsumerGroupOffsets_s
9134
9149 const char *group_id,
9150 const rd_kafka_topic_partition_list_t *partitions);
9151
9157 rd_kafka_ListConsumerGroupOffsets_t *list_grpoffsets);
9158
9165 rd_kafka_ListConsumerGroupOffsets_t **list_grpoffsets,
9166 size_t list_grpoffset_cnt);
9167
9185RD_EXPORT
9187 rd_kafka_t *rk,
9188 rd_kafka_ListConsumerGroupOffsets_t **list_grpoffsets,
9189 size_t list_grpoffsets_cnt,
9190 const rd_kafka_AdminOptions_t *options,
9191 rd_kafka_queue_t *rkqu);
9192
9193
9194
9195/*
9196 * ListConsumerGroupOffsets result type and methods
9197 */
9198
9210RD_EXPORT const rd_kafka_group_result_t **
9213 size_t *cntp);
9214
9215
9216
9227typedef struct rd_kafka_AlterConsumerGroupOffsets_s
9229
9244 const char *group_id,
9245 const rd_kafka_topic_partition_list_t *partitions);
9246
9252 rd_kafka_AlterConsumerGroupOffsets_t *alter_grpoffsets);
9253
9260 rd_kafka_AlterConsumerGroupOffsets_t **alter_grpoffsets,
9261 size_t alter_grpoffset_cnt);
9262
9281RD_EXPORT
9283 rd_kafka_t *rk,
9284 rd_kafka_AlterConsumerGroupOffsets_t **alter_grpoffsets,
9285 size_t alter_grpoffsets_cnt,
9286 const rd_kafka_AdminOptions_t *options,
9287 rd_kafka_queue_t *rkqu);
9288
9289
9290
9291/*
9292 * AlterConsumerGroupOffsets result type and methods
9293 */
9294
9306RD_EXPORT const rd_kafka_group_result_t **
9309 size_t *cntp);
9310
9311
9312
9323typedef struct rd_kafka_DeleteConsumerGroupOffsets_s
9325
9340 const char *group,
9341 const rd_kafka_topic_partition_list_t *partitions);
9342
9349
9357 size_t del_grpoffset_cnt);
9358
9377RD_EXPORT
9379 rd_kafka_t *rk,
9381 size_t del_grpoffsets_cnt,
9382 const rd_kafka_AdminOptions_t *options,
9383 rd_kafka_queue_t *rkqu);
9384
9385
9386
9387/*
9388 * DeleteConsumerGroupOffsets result type and methods
9389 */
9390
9399RD_EXPORT const rd_kafka_group_result_t **
9402 size_t *cntp);
9403
9417 /* Used to retrieve the offset with the largest timestamp of a partition
9418 * as message timestamps can be specified client side this may not match
9419 * the log end offset returned by SPEC_LATEST.
9420 */
9421 RD_KAFKA_OFFSET_SPEC_MAX_TIMESTAMP = -3,
9422 /* Used to retrieve the offset with the earliest timestamp of a
9423 partition. */
9424 RD_KAFKA_OFFSET_SPEC_EARLIEST = -2,
9425 /* Used to retrieve the offset with the latest timestamp of a partition.
9426 */
9427 RD_KAFKA_OFFSET_SPEC_LATEST = -1,
9429
9434typedef struct rd_kafka_ListOffsetsResultInfo_s
9436
9440RD_EXPORT
9443 const rd_kafka_ListOffsetsResultInfo_t *result_info);
9444
9448RD_EXPORT
9450 const rd_kafka_ListOffsetsResultInfo_t *result_info);
9451
9456RD_EXPORT
9459 size_t *cntp);
9460
9485RD_EXPORT
9486void rd_kafka_ListOffsets(rd_kafka_t *rk,
9487 rd_kafka_topic_partition_list_t *topic_partitions,
9488 const rd_kafka_AdminOptions_t *options,
9489 rd_kafka_queue_t *rkqu);
9490
9503 RD_KAFKA_SCRAM_MECHANISM_UNKNOWN = 0,
9504 RD_KAFKA_SCRAM_MECHANISM_SHA_256 = 1,
9505 RD_KAFKA_SCRAM_MECHANISM_SHA_512 = 2,
9506 RD_KAFKA_SCRAM_MECHANISM__CNT
9508
9514typedef struct rd_kafka_ScramCredentialInfo_s rd_kafka_ScramCredentialInfo_t;
9515
9519RD_EXPORT
9521 const rd_kafka_ScramCredentialInfo_t *scram_credential_info);
9522
9526RD_EXPORT
9528 const rd_kafka_ScramCredentialInfo_t *scram_credential_info);
9529
9536typedef struct rd_kafka_UserScramCredentialsDescription_s
9538
9542RD_EXPORT
9545
9549RD_EXPORT
9552
9557RD_EXPORT
9560
9565RD_EXPORT
9569 size_t idx);
9570
9580RD_EXPORT
9584 size_t *cntp);
9585
9597RD_EXPORT
9599 rd_kafka_t *rk,
9600 const char **users,
9601 size_t user_cnt,
9602 const rd_kafka_AdminOptions_t *options,
9603 rd_kafka_queue_t *rkqu);
9604
9608typedef struct rd_kafka_UserScramCredentialAlteration_s
9610
9631RD_EXPORT
9634 rd_kafka_ScramMechanism_t mechanism,
9635 int32_t iterations,
9636 const unsigned char *password,
9637 size_t password_size,
9638 const unsigned char *salt,
9639 size_t salt_size);
9640
9650RD_EXPORT
9653 rd_kafka_ScramMechanism_t mechanism);
9654
9655
9659RD_EXPORT
9662
9666RD_EXPORT
9669 size_t alteration_cnt);
9670
9674typedef struct rd_kafka_AlterUserScramCredentials_result_response_s
9676
9681RD_EXPORT
9684
9689RD_EXPORT
9690const rd_kafka_error_t *
9693
9702RD_EXPORT
9706 size_t *cntp);
9707
9721RD_EXPORT
9723 rd_kafka_t *rk,
9725 size_t alteration_cnt,
9726 const rd_kafka_AdminOptions_t *options,
9727 rd_kafka_queue_t *rkqu);
9728
9741typedef struct rd_kafka_AclBinding_s rd_kafka_AclBinding_t;
9742
9748
9752RD_EXPORT const rd_kafka_error_t *
9753rd_kafka_acl_result_error(const rd_kafka_acl_result_t *aclres);
9754
9755
9759RD_EXPORT const char *
9761
9774
9779 rd_kafka_AclPermissionType_t acl_permission_type);
9780
9799RD_EXPORT rd_kafka_AclBinding_t *
9801 const char *name,
9802 rd_kafka_ResourcePatternType_t resource_pattern_type,
9803 const char *principal,
9804 const char *host,
9805 rd_kafka_AclOperation_t operation,
9806 rd_kafka_AclPermissionType_t permission_type,
9807 char *errstr,
9808 size_t errstr_size);
9809
9837 const char *name,
9838 rd_kafka_ResourcePatternType_t resource_pattern_type,
9839 const char *principal,
9840 const char *host,
9841 rd_kafka_AclOperation_t operation,
9842 rd_kafka_AclPermissionType_t permission_type,
9843 char *errstr,
9844 size_t errstr_size);
9845
9851
9857RD_EXPORT const char *
9859
9865RD_EXPORT const char *
9867
9873RD_EXPORT const char *
9875
9881
9887
9893
9897RD_EXPORT const rd_kafka_error_t *
9899
9900
9906
9907
9913RD_EXPORT void
9915 size_t acl_bindings_cnt);
9916
9924RD_EXPORT const rd_kafka_acl_result_t **
9926 size_t *cntp);
9927
9944RD_EXPORT void rd_kafka_CreateAcls(rd_kafka_t *rk,
9945 rd_kafka_AclBinding_t **new_acls,
9946 size_t new_acls_cnt,
9947 const rd_kafka_AdminOptions_t *options,
9948 rd_kafka_queue_t *rkqu);
9949
9963RD_EXPORT const rd_kafka_AclBinding_t **
9965 size_t *cntp);
9966
9981RD_EXPORT void rd_kafka_DescribeAcls(rd_kafka_t *rk,
9982 rd_kafka_AclBindingFilter_t *acl_filter,
9983 const rd_kafka_AdminOptions_t *options,
9984 rd_kafka_queue_t *rkqu);
9985
9992typedef struct rd_kafka_DeleteAcls_result_response_s
9994
10004 size_t *cntp);
10005
10010RD_EXPORT const rd_kafka_error_t *rd_kafka_DeleteAcls_result_response_error(
10011 const rd_kafka_DeleteAcls_result_response_t *result_response);
10012
10013
10020RD_EXPORT const rd_kafka_AclBinding_t **
10022 const rd_kafka_DeleteAcls_result_response_t *result_response,
10023 size_t *matching_acls_cntp);
10024
10041RD_EXPORT void rd_kafka_DeleteAcls(rd_kafka_t *rk,
10042 rd_kafka_AclBindingFilter_t **del_acls,
10043 size_t del_acls_cnt,
10044 const rd_kafka_AdminOptions_t *options,
10045 rd_kafka_queue_t *rkqu);
10046
10060typedef struct rd_kafka_ElectLeaders_s rd_kafka_ElectLeaders_t;
10061
10070
10084RD_EXPORT rd_kafka_ElectLeaders_t *
10087
10094RD_EXPORT void
10096
10120RD_EXPORT void rd_kafka_ElectLeaders(rd_kafka_t *rk,
10121 rd_kafka_ElectLeaders_t *elect_leaders,
10122 const rd_kafka_AdminOptions_t *options,
10123 rd_kafka_queue_t *rkqu);
10124
10136RD_EXPORT const rd_kafka_topic_partition_result_t **
10138 const rd_kafka_ElectLeaders_result_t *result,
10139 size_t *cntp);
10140
10193RD_EXPORT
10196 const char *token_value,
10197 int64_t md_lifetime_ms,
10198 const char *md_principal_name,
10199 const char **extensions,
10200 size_t extension_size,
10201 char *errstr,
10202 size_t errstr_size);
10203
10224RD_EXPORT
10226 const char *errstr);
10227
10414RD_EXPORT
10415rd_kafka_error_t *rd_kafka_init_transactions(rd_kafka_t *rk, int timeout_ms);
10416
10417
10418
10463RD_EXPORT
10464rd_kafka_error_t *rd_kafka_begin_transaction(rd_kafka_t *rk);
10465
10466
10538RD_EXPORT
10540 rd_kafka_t *rk,
10541 const rd_kafka_topic_partition_list_t *offsets,
10542 const rd_kafka_consumer_group_metadata_t *cgmetadata,
10543 int timeout_ms);
10544
10545
10613RD_EXPORT
10614rd_kafka_error_t *rd_kafka_commit_transaction(rd_kafka_t *rk, int timeout_ms);
10615
10616
10675RD_EXPORT
10676rd_kafka_error_t *rd_kafka_abort_transaction(rd_kafka_t *rk, int timeout_ms);
10677
10678
10681/* @cond NO_DOC */
10682#ifdef __cplusplus
10683}
10684#endif
10685#endif /* _RDKAFKA_H_ */
10686/* @endcond NO_DOC */
RD_EXPORT const rd_kafka_topic_result_t ** rd_kafka_CreateTopics_result_topics(const rd_kafka_CreateTopics_result_t *result, size_t *cntp)
Get an array of topic results from a CreateTopics result.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_conf_interceptor_add_on_new(rd_kafka_conf_t *conf, const char *ic_name, rd_kafka_interceptor_f_on_new_t *on_new, void *ic_opaque)
Append an on_new() interceptor.
RD_EXPORT int32_t rd_kafka_msg_partitioner_murmur2(const rd_kafka_topic_t *rkt, const void *key, size_t keylen, int32_t partition_cnt, void *rkt_opaque, void *msg_opaque)
Murmur2 partitioner (Java compatible).
struct rd_kafka_Node_s rd_kafka_Node_t
Node (broker) information.
Definition rdkafka.h:5115
RD_EXPORT void rd_kafka_CreateAcls(rd_kafka_t *rk, rd_kafka_AclBinding_t **new_acls, size_t new_acls_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Create acls as specified by the new_acls array of size new_topic_cnt elements.
RD_EXPORT void rd_kafka_ElectLeaders_destroy(rd_kafka_ElectLeaders_t *elect_leaders)
Destroy and free a rd_kafka_ElectLeaders_t object previously created with rd_kafka_ElectLeaders_new()
rd_kafka_resp_err_t
Error codes.
Definition rdkafka.h:283
@ RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_EXPIRED
Definition rdkafka.h:574
@ RD_KAFKA_RESP_ERR_POLICY_VIOLATION
Definition rdkafka.h:524
@ RD_KAFKA_RESP_ERR_TRANSACTION_COORDINATOR_FENCED
Definition rdkafka.h:545
@ RD_KAFKA_RESP_ERR__AUTO_OFFSET_RESET
Definition rdkafka.h:409
@ RD_KAFKA_RESP_ERR_RESOURCE_NOT_FOUND
Definition rdkafka.h:627
@ RD_KAFKA_RESP_ERR__PURGE_QUEUE
Definition rdkafka.h:385
@ RD_KAFKA_RESP_ERR_UNSUPPORTED_FOR_MESSAGE_FORMAT
Definition rdkafka.h:522
@ RD_KAFKA_RESP_ERR_OPERATION_NOT_ATTEMPTED
Definition rdkafka.h:551
@ RD_KAFKA_RESP_ERR_INVALID_TRANSACTION_TIMEOUT
Definition rdkafka.h:538
@ RD_KAFKA_RESP_ERR__INVALID_DIFFERENT_RECORD
Definition rdkafka.h:414
@ RD_KAFKA_RESP_ERR_INVALID_PRINCIPAL_TYPE
Definition rdkafka.h:576
@ RD_KAFKA_RESP_ERR_PREFERRED_LEADER_NOT_AVAILABLE
Definition rdkafka.h:602
@ RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION
Definition rdkafka.h:309
@ RD_KAFKA_RESP_ERR__FS
Definition rdkafka.h:311
@ RD_KAFKA_RESP_ERR__TIMED_OUT
Definition rdkafka.h:319
@ RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT
Definition rdkafka.h:441
@ RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE
Definition rdkafka.h:435
@ RD_KAFKA_RESP_ERR_INVALID_REQUIRED_ACKS
Definition rdkafka.h:478
@ RD_KAFKA_RESP_ERR__FATAL
Definition rdkafka.h:389
@ RD_KAFKA_RESP_ERR_STALE_MEMBER_EPOCH
Definition rdkafka.h:652
@ RD_KAFKA_RESP_ERR__UNDERFLOW
Definition rdkafka.h:379
@ RD_KAFKA_RESP_ERR__BEGIN
Definition rdkafka.h:286
@ RD_KAFKA_RESP_ERR__NOENT
Definition rdkafka.h:377
@ RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART
Definition rdkafka.h:431
@ RD_KAFKA_RESP_ERR_RECORD_LIST_TOO_LARGE
Definition rdkafka.h:472
@ RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH
Definition rdkafka.h:530
@ RD_KAFKA_RESP_ERR_NETWORK_EXCEPTION
Definition rdkafka.h:453
@ RD_KAFKA_RESP_ERR_KAFKA_STORAGE_ERROR
Definition rdkafka.h:553
@ RD_KAFKA_RESP_ERR_GROUP_ID_NOT_FOUND
Definition rdkafka.h:580
@ RD_KAFKA_RESP_ERR_TOPIC_ALREADY_EXISTS
Definition rdkafka.h:508
@ RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE
Definition rdkafka.h:460
@ RD_KAFKA_RESP_ERR_MSG_SIZE_TOO_LARGE
Definition rdkafka.h:447
@ RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_AUTH_DISABLED
Definition rdkafka.h:564
@ RD_KAFKA_RESP_ERR_DUPLICATE_RESOURCE
Definition rdkafka.h:629
@ RD_KAFKA_RESP_ERR_INVALID_CONFIG
Definition rdkafka.h:516
@ RD_KAFKA_RESP_ERR_FEATURE_UPDATE_FAILED
Definition rdkafka.h:638
@ RD_KAFKA_RESP_ERR__MAX_POLL_EXCEEDED
Definition rdkafka.h:395
@ RD_KAFKA_RESP_ERR_INVALID_UPDATE_VERSION
Definition rdkafka.h:636
@ RD_KAFKA_RESP_ERR__UNKNOWN_GROUP
Definition rdkafka.h:331
@ RD_KAFKA_RESP_ERR__ISR_INSUFF
Definition rdkafka.h:323
@ RD_KAFKA_RESP_ERR_SASL_AUTHENTICATION_FAILED
Definition rdkafka.h:558
@ RD_KAFKA_RESP_ERR_UNKNOWN_SUBSCRIPTION_ID
Definition rdkafka.h:655
@ RD_KAFKA_RESP_ERR_TELEMETRY_TOO_LARGE
Definition rdkafka.h:658
@ RD_KAFKA_RESP_ERR_INVALID_PRODUCER_ID_MAPPING
Definition rdkafka.h:535
@ RD_KAFKA_RESP_ERR__IN_PROGRESS
Definition rdkafka.h:333
@ RD_KAFKA_RESP_ERR__READ_ONLY
Definition rdkafka.h:375
@ RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS_AFTER_APPEND
Definition rdkafka.h:476
@ RD_KAFKA_RESP_ERR__UNKNOWN_BROKER
Definition rdkafka.h:397
@ RD_KAFKA_RESP_ERR_INCONSISTENT_GROUP_PROTOCOL
Definition rdkafka.h:482
@ RD_KAFKA_RESP_ERR_NON_EMPTY_GROUP
Definition rdkafka.h:578
@ RD_KAFKA_RESP_ERR_SECURITY_DISABLED
Definition rdkafka.h:549
@ RD_KAFKA_RESP_ERR_UNSUPPORTED_ASSIGNOR
Definition rdkafka.h:650
@ RD_KAFKA_RESP_ERR_TOPIC_DELETION_DISABLED
Definition rdkafka.h:588
@ RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_OWNER_MISMATCH
Definition rdkafka.h:568
@ RD_KAFKA_RESP_ERR__INVALID_ARG
Definition rdkafka.h:317
@ RD_KAFKA_RESP_ERR_UNSUPPORTED_VERSION
Definition rdkafka.h:506
@ RD_KAFKA_RESP_ERR__OUTDATED
Definition rdkafka.h:355
@ RD_KAFKA_RESP_ERR__FAIL
Definition rdkafka.h:294
@ RD_KAFKA_RESP_ERR_INVALID_MSG_SIZE
Definition rdkafka.h:433
@ RD_KAFKA_RESP_ERR_STALE_BROKER_EPOCH
Definition rdkafka.h:596
@ RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS
Definition rdkafka.h:455
@ RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS
Definition rdkafka.h:339
@ RD_KAFKA_RESP_ERR__VALUE_SERIALIZATION
Definition rdkafka.h:367
@ RD_KAFKA_RESP_ERR_FETCH_SESSION_ID_NOT_FOUND
Definition rdkafka.h:582
@ RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS
Definition rdkafka.h:341
@ RD_KAFKA_RESP_ERR__APPLICATION
Definition rdkafka.h:403
@ RD_KAFKA_RESP_ERR_REPLICA_NOT_AVAILABLE
Definition rdkafka.h:445
@ RD_KAFKA_RESP_ERR__EXISTING_SUBSCRIPTION
Definition rdkafka.h:337
@ RD_KAFKA_RESP_ERR_NOT_CONTROLLER
Definition rdkafka.h:518
@ RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_NOT_FOUND
Definition rdkafka.h:566
@ RD_KAFKA_RESP_ERR_REBOOTSTRAP_REQUIRED
Definition rdkafka.h:661
@ RD_KAFKA_RESP_ERR_LISTENER_NOT_FOUND
Definition rdkafka.h:586
@ RD_KAFKA_RESP_ERR_INVALID_GROUP_ID
Definition rdkafka.h:484
@ RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT
Definition rdkafka.h:620
@ RD_KAFKA_RESP_ERR__GAPLESS_GUARANTEE
Definition rdkafka.h:393
@ RD_KAFKA_RESP_ERR__INVALID_TYPE
Definition rdkafka.h:381
@ RD_KAFKA_RESP_ERR_OFFSET_NOT_AVAILABLE
Definition rdkafka.h:598
@ RD_KAFKA_RESP_ERR__PURGE_INFLIGHT
Definition rdkafka.h:387
@ RD_KAFKA_RESP_ERR__STATE
Definition rdkafka.h:345
@ RD_KAFKA_RESP_ERR_UNKNOWN_PRODUCER_ID
Definition rdkafka.h:560
@ RD_KAFKA_RESP_ERR_UNKNOWN_LEADER_EPOCH
Definition rdkafka.h:592
@ RD_KAFKA_RESP_ERR_UNACCEPTABLE_CREDENTIAL
Definition rdkafka.h:631
@ RD_KAFKA_RESP_ERR_BROKER_NOT_AVAILABLE
Definition rdkafka.h:443
@ RD_KAFKA_RESP_ERR_INCONSISTENT_VOTER_SET
Definition rdkafka.h:634
@ RD_KAFKA_RESP_ERR_ELIGIBLE_LEADERS_NOT_AVAILABLE
Definition rdkafka.h:609
@ RD_KAFKA_RESP_ERR__RETRY
Definition rdkafka.h:383
@ RD_KAFKA_RESP_ERR_INVALID_REPLICATION_FACTOR
Definition rdkafka.h:512
@ RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_ID
Definition rdkafka.h:642
@ RD_KAFKA_RESP_ERR_NOT_COORDINATOR
Definition rdkafka.h:465
@ RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS
Definition rdkafka.h:335
@ RD_KAFKA_RESP_ERR_ILLEGAL_GENERATION
Definition rdkafka.h:480
@ RD_KAFKA_RESP_ERR__BAD_MSG
Definition rdkafka.h:288
@ RD_KAFKA_RESP_ERR__BAD_COMPRESSION
Definition rdkafka.h:290
@ RD_KAFKA_RESP_ERR__QUEUE_FULL
Definition rdkafka.h:321
@ RD_KAFKA_RESP_ERR__WAIT_CACHE
Definition rdkafka.h:361
@ RD_KAFKA_RESP_ERR__RESOLVE
Definition rdkafka.h:300
@ RD_KAFKA_RESP_ERR_PRODUCER_FENCED
Definition rdkafka.h:625
@ RD_KAFKA_RESP_ERR__AUTHENTICATION
Definition rdkafka.h:351
@ RD_KAFKA_RESP_ERR__NOT_IMPLEMENTED
Definition rdkafka.h:349
@ RD_KAFKA_RESP_ERR_INVALID_RECORD
Definition rdkafka.h:618
@ RD_KAFKA_RESP_ERR__UNKNOWN_PROTOCOL
Definition rdkafka.h:347
@ RD_KAFKA_RESP_ERR__PARTIAL
Definition rdkafka.h:373
@ RD_KAFKA_RESP_ERR_STALE_CTRL_EPOCH
Definition rdkafka.h:449
@ RD_KAFKA_RESP_ERR_CLUSTER_AUTHORIZATION_FAILED
Definition rdkafka.h:498
@ RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_AUTHORIZATION_FAILED
Definition rdkafka.h:572
@ RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS
Definition rdkafka.h:474
@ RD_KAFKA_RESP_ERR_GROUP_MAX_SIZE_REACHED
Definition rdkafka.h:604
@ RD_KAFKA_RESP_ERR_INVALID_TXN_STATE
Definition rdkafka.h:532
@ RD_KAFKA_RESP_ERR__DESTROY
Definition rdkafka.h:292
@ RD_KAFKA_RESP_ERR__INCONSISTENT
Definition rdkafka.h:391
@ RD_KAFKA_RESP_ERR__KEY_SERIALIZATION
Definition rdkafka.h:365
@ RD_KAFKA_RESP_ERR_UNKNOWN_MEMBER_ID
Definition rdkafka.h:486
@ RD_KAFKA_RESP_ERR_MEMBER_ID_REQUIRED
Definition rdkafka.h:600
@ RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS
Definition rdkafka.h:541
@ RD_KAFKA_RESP_ERR_UNRELEASED_INSTANCE_ID
Definition rdkafka.h:647
@ RD_KAFKA_RESP_ERR__TRANSPORT
Definition rdkafka.h:296
@ RD_KAFKA_RESP_ERR_INVALID_TIMESTAMP
Definition rdkafka.h:500
@ RD_KAFKA_RESP_ERR_INVALID_REQUEST
Definition rdkafka.h:520
@ RD_KAFKA_RESP_ERR_OFFSET_OUT_OF_RANGE
Definition rdkafka.h:427
@ RD_KAFKA_RESP_ERR__LOG_TRUNCATION
Definition rdkafka.h:411
@ RD_KAFKA_RESP_ERR_FENCED_INSTANCE_ID
Definition rdkafka.h:607
@ RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_REQUEST_NOT_ALLOWED
Definition rdkafka.h:570
@ RD_KAFKA_RESP_ERR_UNSUPPORTED_SASL_MECHANISM
Definition rdkafka.h:502
@ RD_KAFKA_RESP_ERR__CONFLICT
Definition rdkafka.h:343
@ RD_KAFKA_RESP_ERR_REASSIGNMENT_IN_PROGRESS
Definition rdkafka.h:562
@ RD_KAFKA_RESP_ERR_INVALID_COMMIT_OFFSET_SIZE
Definition rdkafka.h:492
@ RD_KAFKA_RESP_ERR__NOOP
Definition rdkafka.h:407
@ RD_KAFKA_RESP_ERR_THROTTLING_QUOTA_EXCEEDED
Definition rdkafka.h:622
@ RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE
Definition rdkafka.h:357
@ RD_KAFKA_RESP_ERR_DUPLICATE_SEQUENCE_NUMBER
Definition rdkafka.h:528
@ RD_KAFKA_RESP_ERR_ELECTION_NOT_NEEDED
Definition rdkafka.h:611
@ RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED
Definition rdkafka.h:494
@ RD_KAFKA_RESP_ERR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED
Definition rdkafka.h:547
@ RD_KAFKA_RESP_ERR_GROUP_SUBSCRIBED_TO_TOPIC
Definition rdkafka.h:616
@ RD_KAFKA_RESP_ERR__ASSIGNMENT_LOST
Definition rdkafka.h:405
@ RD_KAFKA_RESP_ERR__MSG_TIMED_OUT
Definition rdkafka.h:302
@ RD_KAFKA_RESP_ERR_INVALID_SESSION_TIMEOUT
Definition rdkafka.h:488
@ RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC
Definition rdkafka.h:313
@ RD_KAFKA_RESP_ERR_PRINCIPAL_DESERIALIZATION_FAILURE
Definition rdkafka.h:640
@ RD_KAFKA_RESP_ERR_OFFSET_METADATA_TOO_LARGE
Definition rdkafka.h:451
@ RD_KAFKA_RESP_ERR_FENCED_MEMBER_EPOCH
Definition rdkafka.h:644
@ RD_KAFKA_RESP_ERR__NODE_UPDATE
Definition rdkafka.h:325
@ RD_KAFKA_RESP_ERR_NO_REASSIGNMENT_IN_PROGRESS
Definition rdkafka.h:613
@ RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE
Definition rdkafka.h:298
@ RD_KAFKA_RESP_ERR_INVALID_FETCH_SESSION_EPOCH
Definition rdkafka.h:584
@ RD_KAFKA_RESP_ERR__VALUE_DESERIALIZATION
Definition rdkafka.h:371
@ RD_KAFKA_RESP_ERR__NO_OFFSET
Definition rdkafka.h:353
@ RD_KAFKA_RESP_ERR_INVALID_MSG
Definition rdkafka.h:429
@ RD_KAFKA_RESP_ERR_LOG_DIR_NOT_FOUND
Definition rdkafka.h:556
@ RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN
Definition rdkafka.h:315
@ RD_KAFKA_RESP_ERR__DESTROY_BROKER
Definition rdkafka.h:416
@ RD_KAFKA_RESP_ERR_INVALID_REPLICA_ASSIGNMENT
Definition rdkafka.h:514
@ RD_KAFKA_RESP_ERR__FENCED
Definition rdkafka.h:401
@ RD_KAFKA_RESP_ERR_INVALID_PARTITIONS
Definition rdkafka.h:510
@ RD_KAFKA_RESP_ERR__NOT_CONFIGURED
Definition rdkafka.h:399
@ RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE
Definition rdkafka.h:359
@ RD_KAFKA_RESP_ERR_NO_ERROR
Definition rdkafka.h:425
@ RD_KAFKA_RESP_ERR__END
Definition rdkafka.h:419
@ RD_KAFKA_RESP_ERR__INTR
Definition rdkafka.h:363
@ RD_KAFKA_RESP_ERR_ILLEGAL_SASL_STATE
Definition rdkafka.h:504
@ RD_KAFKA_RESP_ERR_GROUP_AUTHORIZATION_FAILED
Definition rdkafka.h:496
@ RD_KAFKA_RESP_ERR_UNSUPPORTED_COMPRESSION_TYPE
Definition rdkafka.h:594
@ RD_KAFKA_RESP_ERR_OUT_OF_ORDER_SEQUENCE_NUMBER
Definition rdkafka.h:526
@ RD_KAFKA_RESP_ERR__KEY_DESERIALIZATION
Definition rdkafka.h:369
@ RD_KAFKA_RESP_ERR__SSL
Definition rdkafka.h:327
@ RD_KAFKA_RESP_ERR_REBALANCE_IN_PROGRESS
Definition rdkafka.h:490
@ RD_KAFKA_RESP_ERR_UNKNOWN
Definition rdkafka.h:423
@ RD_KAFKA_RESP_ERR_FENCED_LEADER_EPOCH
Definition rdkafka.h:590
@ RD_KAFKA_RESP_ERR__WAIT_COORD
Definition rdkafka.h:329
@ RD_KAFKA_RESP_ERR__PARTITION_EOF
Definition rdkafka.h:307
@ RD_KAFKA_RESP_ERR_TOPIC_EXCEPTION
Definition rdkafka.h:470
RD_EXPORT void rd_kafka_TopicCollection_destroy(rd_kafka_TopicCollection_t *topics)
Destroy and free a TopicCollection object created with rd_kafka_TopicCollection_new_* methods.
RD_EXPORT int32_t * rd_kafka_brokers_learned_ids(rd_kafka_t *rk, size_t *cntp)
Retrieve and return the learned broker ids.
RD_EXPORT void rd_kafka_conf_set_throttle_cb(rd_kafka_conf_t *conf, void(*throttle_cb)(rd_kafka_t *rk, const char *broker_name, int32_t broker_id, int throttle_time_ms, void *opaque))
Set throttle callback.
RD_EXPORT void rd_kafka_NewTopic_destroy(rd_kafka_NewTopic_t *new_topic)
Destroy and free a NewTopic object previously created with rd_kafka_NewTopic_new()
RD_EXPORT void rd_kafka_conf_set_socket_cb(rd_kafka_conf_t *conf, int(*socket_cb)(int domain, int type, int protocol, void *opaque))
Set socket callback.
RD_EXPORT void rd_kafka_get_err_descs(const struct rd_kafka_err_desc **errdescs, size_t *cntp)
Returns the full list of error codes.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_offsets_store(rd_kafka_t *rk, rd_kafka_topic_partition_list_t *offsets)
Store offsets for next auto-commit for one or more partitions.
RD_EXPORT void rd_kafka_Uuid_destroy(rd_kafka_Uuid_t *uuid)
Destroy the provided uuid.
RD_EXPORT rd_kafka_consumer_group_state_t rd_kafka_ConsumerGroupListing_state(const rd_kafka_ConsumerGroupListing_t *grplist)
Gets state for the grplist group.
RD_EXPORT rd_kafka_queue_t * rd_kafka_queue_new(rd_kafka_t *rk)
Create a new message queue.
RD_EXPORT void rd_kafka_DeleteRecords_destroy_array(rd_kafka_DeleteRecords_t **del_records, size_t del_record_cnt)
Helper function to destroy all DeleteRecords objects in the del_groups array (of del_group_cnt elemen...
RD_EXPORT rd_kafka_resp_err_t rd_kafka_assign(rd_kafka_t *rk, const rd_kafka_topic_partition_list_t *partitions)
Atomic assignment of partitions to consume.
struct rd_kafka_NewTopic_s rd_kafka_NewTopic_t
Definition rdkafka.h:7423
struct rd_kafka_NewPartitions_s rd_kafka_NewPartitions_t
Definition rdkafka.h:7641
RD_EXPORT void rd_kafka_conf_set_log_cb(rd_kafka_conf_t *conf, void(*log_cb)(const rd_kafka_t *rk, int level, const char *fac, const char *buf))
Set logger callback.
RD_EXPORT rd_kafka_conf_t * rd_kafka_conf_dup(const rd_kafka_conf_t *conf)
Creates a copy/duplicate of configuration object conf.
RD_EXPORT const char * rd_kafka_AclPermissionType_name(rd_kafka_AclPermissionType_t acl_permission_type)
struct rd_kafka_ListOffsetsResultInfo_s rd_kafka_ListOffsetsResultInfo_t
Information returned from a ListOffsets call for a specific rd_kafka_topic_partition_t.
Definition rdkafka.h:9434
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_engine_callback_data(rd_kafka_conf_t *conf, void *callback_data)
Set callback_data for OpenSSL engine.
RD_EXPORT void rd_kafka_conf_set_opaque(rd_kafka_conf_t *conf, void *opaque)
Sets the application's opaque pointer that will be passed to callbacks.
struct rd_kafka_DeleteRecords_s rd_kafka_DeleteRecords_t
Definition rdkafka.h:8237
RD_EXPORT int rd_kafka_thread_cnt(void)
Retrieve the current number of threads in use by librdkafka.
RD_EXPORT const rd_kafka_acl_result_t ** rd_kafka_CreateAcls_result_acls(const rd_kafka_CreateAcls_result_t *result, size_t *cntp)
Get an array of acl results from a CreateAcls result.
struct rd_kafka_TopicPartitionInfo_s rd_kafka_TopicPartitionInfo_t
TopicPartition represents a partition in the DescribeTopics result.
Definition rdkafka.h:8336
struct rd_kafka_ElectLeaders_s rd_kafka_ElectLeaders_t
Represents elect leaders request.
Definition rdkafka.h:10060
RD_EXPORT int32_t rd_kafka_ScramCredentialInfo_iterations(const rd_kafka_ScramCredentialInfo_t *scram_credential_info)
Returns the iterations of a given ScramCredentialInfo.
struct rd_kafka_TopicDescription_s rd_kafka_TopicDescription_t
DescribeTopics result type.
Definition rdkafka.h:8342
RD_EXPORT void rd_kafka_metadata_destroy(const struct rd_kafka_metadata *metadata)
Release metadata memory.
rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_request_sent_t(rd_kafka_t *rk, int sockfd, const char *brokername, int32_t brokerid, int16_t ApiKey, int16_t ApiVersion, int32_t CorrId, size_t size, void *ic_opaque)
on_request_sent() is called when a request has been fully written to a broker TCP connections socket.
Definition rdkafka.h:6551
RD_EXPORT void rd_kafka_topic_partition_list_destroy(rd_kafka_topic_partition_list_t *rkparlist)
Free all resources used by the list and the list itself.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_interceptor_add_on_commit(rd_kafka_t *rk, const char *ic_name, rd_kafka_interceptor_f_on_commit_t *on_commit, void *ic_opaque)
Append an on_commit() interceptor.
RD_EXPORT const char * rd_kafka_ConfigResource_name(const rd_kafka_ConfigResource_t *config)
rd_kafka_event_t rd_kafka_DeleteRecords_result_t
Definition rdkafka.h:5872
RD_EXPORT const rd_kafka_UserScramCredentialsDescription_t ** rd_kafka_DescribeUserScramCredentials_result_descriptions(const rd_kafka_DescribeUserScramCredentials_result_t *result, size_t *cntp)
Get an array of descriptions from a DescribeUserScramCredentials result.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_interceptor_add_on_consume(rd_kafka_t *rk, const char *ic_name, rd_kafka_interceptor_f_on_consume_t *on_consume, void *ic_opaque)
Append an on_consume() interceptor.
RD_EXPORT rd_kafka_error_t * rd_kafka_begin_transaction(rd_kafka_t *rk)
Begin a new transaction.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_subscribe(rd_kafka_t *rk, const rd_kafka_topic_partition_list_t *topics)
Subscribe to topic set using balanced consumer groups.
struct rd_kafka_DeleteAcls_result_response_s rd_kafka_DeleteAcls_result_response_t
Definition rdkafka.h:9992
RD_EXPORT const rd_kafka_DeleteRecords_result_t * rd_kafka_event_DeleteRecords_result(rd_kafka_event_t *rkev)
RD_EXPORT rd_kafka_error_t * rd_kafka_AdminOptions_set_require_stable_offsets(rd_kafka_AdminOptions_t *options, int true_or_false)
Whether broker should return stable offsets (transaction-committed).
RD_EXPORT rd_kafka_resp_err_t rd_kafka_topic_partition_list_set_offset(rd_kafka_topic_partition_list_t *rktparlist, const char *topic, int32_t partition, int64_t offset)
Set offset to offset for topic and partition.
RD_EXPORT const rd_kafka_ConsumerGroupListing_t ** rd_kafka_ListConsumerGroups_result_valid(const rd_kafka_ListConsumerGroups_result_t *result, size_t *cntp)
Get an array of valid list groups from a ListConsumerGroups result.
RD_EXPORT rd_kafka_event_type_t rd_kafka_event_type(const rd_kafka_event_t *rkev)
RD_EXPORT rd_kafka_resp_err_t rd_kafka_assignment(rd_kafka_t *rk, rd_kafka_topic_partition_list_t **partitions)
Returns the current partition assignment as set by rd_kafka_assign() or rd_kafka_incremental_assign()...
RD_EXPORT void rd_kafka_conf_set_rebalance_cb(rd_kafka_conf_t *conf, void(*rebalance_cb)(rd_kafka_t *rk, rd_kafka_resp_err_t err, rd_kafka_topic_partition_list_t *partitions, void *opaque))
Consumer: Set rebalance callback for use with coordinated consumer group balancing.
RD_EXPORT void rd_kafka_yield(rd_kafka_t *rk)
Cancels the current callback dispatcher (rd_kafka_poll(), rd_kafka_consume_callback(),...
RD_EXPORT rd_kafka_topic_conf_t * rd_kafka_conf_get_default_topic_conf(rd_kafka_conf_t *conf)
Gets the default topic configuration as previously set with rd_kafka_conf_set_default_topic_conf() or...
RD_EXPORT rd_kafka_resp_err_t rd_kafka_interceptor_add_on_destroy(rd_kafka_t *rk, const char *ic_name, rd_kafka_interceptor_f_on_destroy_t *on_destroy, void *ic_opaque)
Append an on_destroy() interceptor.
RD_EXPORT void rd_kafka_conf_set_resolve_cb(rd_kafka_conf_t *conf, int(*resolve_cb)(const char *node, const char *service, const struct addrinfo *hints, struct addrinfo **res, void *opaque))
Set address resolution callback.
RD_EXPORT void rd_kafka_conf_set_default_topic_conf(rd_kafka_conf_t *conf, rd_kafka_topic_conf_t *tconf)
Sets the default topic configuration to use for automatically subscribed topics (e....
RD_EXPORT const rd_kafka_AclOperation_t * rd_kafka_ConsumerGroupDescription_authorized_operations(const rd_kafka_ConsumerGroupDescription_t *grpdesc, size_t *cntp)
Gets the authorized ACL operations for the grpdesc group.
RD_EXPORT int rd_kafka_queue_poll_callback(rd_kafka_queue_t *rkqu, int timeout_ms)
Poll a queue for events served through callbacks for max timeout_ms.
RD_EXPORT rd_kafka_ElectLeaders_t * rd_kafka_ElectLeaders_new(rd_kafka_ElectionType_t election_type, rd_kafka_topic_partition_list_t *partitions)
Create a new rd_kafka_ElectLeaders_t object. This object is later passed to rd_kafka_ElectLeaders().
rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_conf_dup_t(rd_kafka_conf_t *new_conf, const rd_kafka_conf_t *old_conf, size_t filter_cnt, const char **filter, void *ic_opaque)
on_conf_dup() is called from rd_kafka_conf_dup() in the order the interceptors were added and is used...
Definition rdkafka.h:6373
RD_EXPORT rd_kafka_resp_err_t rd_kafka_oauthbearer_set_token(rd_kafka_t *rk, const char *token_value, int64_t md_lifetime_ms, const char *md_principal_name, const char **extensions, size_t extension_size, char *errstr, size_t errstr_size)
Set SASL/OAUTHBEARER token and metadata.
RD_EXPORT const char * rd_kafka_ConsumerGroupDescription_group_id(const rd_kafka_ConsumerGroupDescription_t *grpdesc)
Gets the group id for the grpdesc group.
RD_EXPORT rd_kafka_ListConsumerGroupOffsets_t * rd_kafka_ListConsumerGroupOffsets_new(const char *group_id, const rd_kafka_topic_partition_list_t *partitions)
Create a new ListConsumerGroupOffsets object. This object is later passed to rd_kafka_ListConsumerGro...
RD_EXPORT rd_kafka_AclBindingFilter_t * rd_kafka_AclBindingFilter_new(rd_kafka_ResourceType_t restype, const char *name, rd_kafka_ResourcePatternType_t resource_pattern_type, const char *principal, const char *host, rd_kafka_AclOperation_t operation, rd_kafka_AclPermissionType_t permission_type, char *errstr, size_t errstr_size)
Create a new AclBindingFilter object. This object is later passed to rd_kafka_DescribeAcls() or rd_ka...
RD_EXPORT const char * rd_kafka_MemberDescription_client_id(const rd_kafka_MemberDescription_t *member)
Gets client id of member.
rd_kafka_IsolationLevel_t
IsolationLevel enum name for use with rd_kafka_AdminOptions_new()
Definition rdkafka.h:7145
RD_EXPORT size_t rd_kafka_event_message_array(rd_kafka_event_t *rkev, const rd_kafka_message_t **rkmessages, size_t size)
Extacts size message(s) from the event into the pre-allocated array rkmessages.
RD_EXPORT int rd_kafka_ConfigEntry_is_synonym(const rd_kafka_ConfigEntry_t *entry)
RD_EXPORT void rd_kafka_conf_set_offset_commit_cb(rd_kafka_conf_t *conf, void(*offset_commit_cb)(rd_kafka_t *rk, rd_kafka_resp_err_t err, rd_kafka_topic_partition_list_t *offsets, void *opaque))
Consumer: Set offset commit callback for use with consumer groups.
RD_EXPORT void rd_kafka_CreatePartitions(rd_kafka_t *rk, rd_kafka_NewPartitions_t **new_parts, size_t new_parts_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Create additional partitions for the given topics, as specified by the new_parts array of size new_pa...
RD_EXPORT const char * rd_kafka_AlterUserScramCredentials_result_response_user(const rd_kafka_AlterUserScramCredentials_result_response_t *response)
Returns the username for a rd_kafka_AlterUserScramCredentials_result_response.
RD_EXPORT rd_kafka_topic_partition_t * rd_kafka_topic_partition_list_find(const rd_kafka_topic_partition_list_t *rktparlist, const char *topic, int32_t partition)
Find element by topic and partition.
rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_new_t(rd_kafka_t *rk, const rd_kafka_conf_t *conf, void *ic_opaque, char *errstr, size_t errstr_size)
on_new() is called from rd_kafka_new() prior toreturning the newly created client instance to the app...
Definition rdkafka.h:6408
RD_EXPORT const char * rd_kafka_consumer_group_metadata_member_id(const rd_kafka_consumer_group_metadata_t *group_metadata)
Get member id of a group metadata.
RD_EXPORT rd_kafka_topic_t * rd_kafka_topic_new(rd_kafka_t *rk, const char *topic, rd_kafka_topic_conf_t *conf)
Creates a new topic handle for topic named topic.
RD_EXPORT int rd_kafka_event_error_is_fatal(rd_kafka_event_t *rkev)
RD_EXPORT const rd_kafka_topic_partition_list_t * rd_kafka_DeleteRecords_result_offsets(const rd_kafka_DeleteRecords_result_t *result)
Get a list of topic and partition results from a DeleteRecords result. The returned objects will cont...
RD_EXPORT void rd_kafka_AlterConsumerGroupOffsets_destroy_array(rd_kafka_AlterConsumerGroupOffsets_t **alter_grpoffsets, size_t alter_grpoffset_cnt)
Helper function to destroy all AlterConsumerGroupOffsets objects in the alter_grpoffsets array (of al...
rd_kafka_event_t rd_kafka_CreatePartitions_result_t
Definition rdkafka.h:5864
RD_EXPORT void rd_kafka_topic_conf_set_opaque(rd_kafka_topic_conf_t *conf, void *rkt_opaque)
Sets the application's opaque pointer that will be passed to all topic callbacks as the rkt_opaque ar...
RD_EXPORT const rd_kafka_ConfigResource_t ** rd_kafka_IncrementalAlterConfigs_result_resources(const rd_kafka_IncrementalAlterConfigs_result_t *result, size_t *cntp)
Get an array of resource results from a IncrementalAlterConfigs result.
struct rd_kafka_AlterUserScramCredentials_result_response_s rd_kafka_AlterUserScramCredentials_result_response_t
Result of a single user SCRAM alteration.
Definition rdkafka.h:9674
RD_EXPORT const char * rd_kafka_ConfigEntry_name(const rd_kafka_ConfigEntry_t *entry)
RD_EXPORT void rd_kafka_ListOffsets(rd_kafka_t *rk, rd_kafka_topic_partition_list_t *topic_partitions, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
List offsets for the specified topic_partitions. This operation enables to find the beginning offset,...
RD_EXPORT const char * rd_kafka_rebalance_protocol(rd_kafka_t *rk)
The rebalance protocol currently in use. This will be "NONE" if the consumer has not (yet) joined a g...
RD_EXPORT void rd_kafka_dump(FILE *fp, rd_kafka_t *rk)
Dumps rdkafka's internal state for handle rk to stream fp.
RD_EXPORT const rd_kafka_topic_partition_list_t * rd_kafka_group_result_partitions(const rd_kafka_group_result_t *groupres)
RD_EXPORT rd_kafka_resp_err_t rd_kafka_pause_partitions(rd_kafka_t *rk, rd_kafka_topic_partition_list_t *partitions)
Pause producing or consumption for the provided list of partitions.
RD_EXPORT const char * rd_kafka_TopicDescription_name(const rd_kafka_TopicDescription_t *topicdesc)
Gets the topic name for the topicdesc topic.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_NewTopic_set_config(rd_kafka_NewTopic_t *new_topic, const char *name, const char *value)
Set (broker-side) topic configuration name/value pair.
RD_EXPORT void * rd_kafka_event_opaque(rd_kafka_event_t *rkev)
RD_EXPORT const char * rd_kafka_consumer_group_state_name(rd_kafka_consumer_group_state_t state)
Returns a name for a state code.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_interceptor_add_on_broker_state_change(rd_kafka_t *rk, const char *ic_name, rd_kafka_interceptor_f_on_broker_state_change_t *on_broker_state_change, void *ic_opaque)
Append an on_broker_state_change() interceptor.
RD_EXPORT const char ** rd_kafka_topic_conf_dump(rd_kafka_topic_conf_t *conf, size_t *cntp)
Dump the topic configuration properties and values of conf to an array with "key",...
rd_kafka_event_t rd_kafka_DescribeConsumerGroups_result_t
Definition rdkafka.h:5876
RD_EXPORT const rd_kafka_DescribeTopics_result_t * rd_kafka_event_DescribeTopics_result(rd_kafka_event_t *rkev)
Get DescribeTopics result.
RD_EXPORT rd_kafka_error_t * rd_kafka_AdminOptions_set_include_authorized_operations(rd_kafka_AdminOptions_t *options, int true_or_false)
Whether broker should return authorized operations for the given resource in the DescribeConsumerGrou...
RD_EXPORT rd_kafka_resp_err_t rd_kafka_test_fatal_error(rd_kafka_t *rk, rd_kafka_resp_err_t err, const char *reason)
Trigger a fatal error for testing purposes.
RD_EXPORT rd_kafka_queue_t * rd_kafka_queue_get_sasl(rd_kafka_t *rk)
RD_EXPORT void rd_kafka_AclBinding_destroy(rd_kafka_AclBinding_t *acl_binding)
Destroy and free an AclBinding object previously created with rd_kafka_AclBinding_new()
RD_EXPORT void * rd_kafka_mem_calloc(rd_kafka_t *rk, size_t num, size_t size)
Allocate and zero memory using the same allocator librdkafka uses.
struct rd_kafka_UserScramCredentialsDescription_s rd_kafka_UserScramCredentialsDescription_t
Representation of all SASL/SCRAM credentials associated with a user that can be retrieved,...
Definition rdkafka.h:9536
RD_EXPORT const rd_kafka_Uuid_t * rd_kafka_TopicDescription_topic_id(const rd_kafka_TopicDescription_t *topicdesc)
Gets the topic id for the topicdesc topic.
RD_EXPORT void rd_kafka_error_destroy(rd_kafka_error_t *error)
Free and destroy an error object.
rd_kafka_event_t rd_kafka_DeleteTopics_result_t
Definition rdkafka.h:5856
RD_EXPORT const char * rd_kafka_Uuid_base64str(const rd_kafka_Uuid_t *uuid)
Computes base64 encoding for the given uuid string.
RD_EXPORT const rd_kafka_ListOffsets_result_t * rd_kafka_event_ListOffsets_result(rd_kafka_event_t *rkev)
Get ListOffsets result.
RD_EXPORT void rd_kafka_message_destroy(rd_kafka_message_t *rkmessage)
Frees resources for rkmessage and hands ownership back to rdkafka.
RD_EXPORT const rd_kafka_topic_partition_list_t * rd_kafka_MemberAssignment_partitions(const rd_kafka_MemberAssignment_t *assignment)
Gets assigned partitions of a member assignment.
RD_EXPORT rd_kafka_error_t * rd_kafka_consumer_group_metadata_write(const rd_kafka_consumer_group_metadata_t *cgmd, void **bufferp, size_t *sizep)
Serialize the consumer group metadata to a binary format. This is mainly for client binding use and n...
RD_EXPORT void rd_kafka_IncrementalAlterConfigs(rd_kafka_t *rk, rd_kafka_ConfigResource_t **configs, size_t config_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Incrementally update the configuration for the specified resources. Updates are not transactional so ...
rd_kafka_event_t rd_kafka_ListConsumerGroupOffsets_result_t
Definition rdkafka.h:5884
RD_EXPORT rd_kafka_consumer_group_metadata_t * rd_kafka_consumer_group_metadata_new_with_genid(const char *group_id, int32_t generation_id, const char *member_id, const char *group_instance_id)
Create a new consumer group metadata object. This is typically only used for writing tests.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_conf_interceptor_add_on_conf_dup(rd_kafka_conf_t *conf, const char *ic_name, rd_kafka_interceptor_f_on_conf_dup_t *on_conf_dup, void *ic_opaque)
Append an on_conf_dup() interceptor.
RD_EXPORT const rd_kafka_ElectLeaders_result_t * rd_kafka_event_ElectLeaders_result(rd_kafka_event_t *rkev)
Get ElectLeaders result.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_consumer_close(rd_kafka_t *rk)
Close the consumer.
RD_EXPORT const rd_kafka_AlterUserScramCredentials_result_t * rd_kafka_event_AlterUserScramCredentials_result(rd_kafka_event_t *rkev)
Get AlterUserScramCredentials result.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_committed(rd_kafka_t *rk, rd_kafka_topic_partition_list_t *partitions, int timeout_ms)
Retrieve committed offsets for topics+partitions.
rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_send_t(rd_kafka_t *rk, rd_kafka_message_t *rkmessage, void *ic_opaque)
on_send() is called from rd_kafka_produce*() (et.al) prior to the partitioner being called.
Definition rdkafka.h:6448
RD_EXPORT const rd_kafka_topic_result_t ** rd_kafka_CreatePartitions_result_topics(const rd_kafka_CreatePartitions_result_t *result, size_t *cntp)
Get an array of topic results from a CreatePartitions result.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_header_get_all(const rd_kafka_headers_t *hdrs, size_t idx, const char **namep, const void **valuep, size_t *sizep)
Iterator for all headers.
RD_EXPORT const rd_kafka_TopicPartitionInfo_t ** rd_kafka_TopicDescription_partitions(const rd_kafka_TopicDescription_t *topicdesc, size_t *cntp)
Gets an array of partitions for the topicdesc topic.
struct rd_kafka_TopicCollection_s rd_kafka_TopicCollection_t
Represents a collection of topics, to be passed to DescribeTopics.
Definition rdkafka.h:8330
RD_EXPORT rd_kafka_DeleteRecords_t * rd_kafka_DeleteRecords_new(const rd_kafka_topic_partition_list_t *before_offsets)
Create a new DeleteRecords object. This object is later passed to rd_kafka_DeleteRecords().
RD_EXPORT void rd_kafka_AlterConsumerGroupOffsets(rd_kafka_t *rk, rd_kafka_AlterConsumerGroupOffsets_t **alter_grpoffsets, size_t alter_grpoffsets_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Alter committed offsets for a set of partitions in a consumer group. This will succeed at the partiti...
RD_EXPORT void rd_kafka_DescribeConsumerGroups(rd_kafka_t *rk, const char **groups, size_t groups_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Describe groups from cluster as specified by the groups array of size groups_cnt elements.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_unsubscribe(rd_kafka_t *rk)
Unsubscribe from the current subscription set.
RD_EXPORT void rd_kafka_log_print(const rd_kafka_t *rk, int level, const char *fac, const char *buf)
Builtin (default) log sink: print to stderr.
RD_EXPORT ssize_t rd_kafka_consume_batch_queue(rd_kafka_queue_t *rkqu, int timeout_ms, rd_kafka_message_t **rkmessages, size_t rkmessages_size)
Consume batch of messages from queue.
rd_kafka_OffsetSpec_t
Allows to specify the desired offsets when using ListOffsets.
Definition rdkafka.h:9416
RD_EXPORT rd_kafka_topic_conf_t * rd_kafka_topic_conf_dup(const rd_kafka_topic_conf_t *conf)
Creates a copy/duplicate of topic configuration object conf.
RD_EXPORT const rd_kafka_MemberDescription_t * rd_kafka_ConsumerGroupDescription_member(const rd_kafka_ConsumerGroupDescription_t *grpdesc, size_t idx)
Gets a member of grpdesc group.
RD_EXPORT rd_kafka_topic_partition_list_t * rd_kafka_topic_partition_list_copy(const rd_kafka_topic_partition_list_t *src)
Make a copy of an existing list.
RD_EXPORT void rd_kafka_queue_cb_event_enable(rd_kafka_queue_t *rkqu, void(*event_cb)(rd_kafka_t *rk, void *qev_opaque), void *qev_opaque)
Enable callback event triggering for queue.
RD_EXPORT const rd_kafka_ConfigResource_t ** rd_kafka_DescribeConfigs_result_resources(const rd_kafka_DescribeConfigs_result_t *result, size_t *cntp)
Get an array of resource results from a DescribeConfigs result.
RD_EXPORT const rd_kafka_DeleteConsumerGroupOffsets_result_t * rd_kafka_event_DeleteConsumerGroupOffsets_result(rd_kafka_event_t *rkev)
Get DeleteConsumerGroupOffsets result.
RD_EXPORT const rd_kafka_CreateTopics_result_t * rd_kafka_event_CreateTopics_result(rd_kafka_event_t *rkev)
Get CreateTopics result.
RD_EXPORT rd_kafka_queue_t * rd_kafka_queue_get_partition(rd_kafka_t *rk, const char *topic, int32_t partition)
RD_EXPORT int64_t rd_kafka_message_timestamp(const rd_kafka_message_t *rkmessage, rd_kafka_timestamp_type_t *tstype)
Returns the message timestamp for a consumed message.
RD_EXPORT int64_t rd_kafka_Uuid_most_significant_bits(const rd_kafka_Uuid_t *uuid)
Gets most significant 64 bits for the given UUID.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_interceptor_add_on_request_sent(rd_kafka_t *rk, const char *ic_name, rd_kafka_interceptor_f_on_request_sent_t *on_request_sent, void *ic_opaque)
Append an on_request_sent() interceptor.
RD_EXPORT const rd_kafka_CreateAcls_result_t * rd_kafka_event_CreateAcls_result(rd_kafka_event_t *rkev)
RD_EXPORT rd_kafka_error_t * rd_kafka_seek_partitions(rd_kafka_t *rk, rd_kafka_topic_partition_list_t *partitions, int timeout_ms)
Seek consumer for partitions in partitions to the per-partition offset in the .offset field of partit...
RD_EXPORT rd_kafka_resp_err_t rd_kafka_fatal_error(rd_kafka_t *rk, char *errstr, size_t errstr_size)
Returns the first fatal error set on this client instance, or RD_KAFKA_RESP_ERR_NO_ERROR if no fatal ...
rd_kafka_event_t rd_kafka_CreateAcls_result_t
Definition rdkafka.h:5858
RD_EXPORT rd_kafka_resp_err_t rd_kafka_query_watermark_offsets(rd_kafka_t *rk, const char *topic, int32_t partition, int64_t *low, int64_t *high, int timeout_ms)
Query broker for low (oldest/beginning) and high (newest/end) offsets for partition.
rd_kafka_event_t rd_kafka_CreateTopics_result_t
Definition rdkafka.h:5854
RD_EXPORT rd_kafka_topic_partition_t * rd_kafka_topic_partition_list_add(rd_kafka_topic_partition_list_t *rktparlist, const char *topic, int32_t partition)
Add topic+partition to list.
RD_EXPORT const rd_kafka_error_t * rd_kafka_group_result_error(const rd_kafka_group_result_t *groupres)
Group result provides per-group operation result information.
RD_EXPORT int rd_kafka_consume_callback_queue(rd_kafka_queue_t *rkqu, int timeout_ms, void(*consume_cb)(rd_kafka_message_t *rkmessage, void *commit_opaque), void *commit_opaque)
Consume multiple messages from queue with callback.
RD_EXPORT void rd_kafka_conf_set_open_cb(rd_kafka_conf_t *conf, int(*open_cb)(const char *pathname, int flags, mode_t mode, void *opaque))
Set open callback.
rd_kafka_event_t rd_kafka_DeleteAcls_result_t
Definition rdkafka.h:5862
RD_EXPORT rd_kafka_error_t * rd_kafka_incremental_unassign(rd_kafka_t *rk, const rd_kafka_topic_partition_list_t *partitions)
Incrementally remove partitions from the current assignment.
RD_EXPORT const char * rd_kafka_event_error_string(rd_kafka_event_t *rkev)
RD_EXPORT rd_kafka_error_t * rd_kafka_AdminOptions_set_match_consumer_group_states(rd_kafka_AdminOptions_t *options, const rd_kafka_consumer_group_state_t *consumer_group_states, size_t consumer_group_states_cnt)
Set consumer groups states to query for.
RD_EXPORT const rd_kafka_AclBinding_t ** rd_kafka_DeleteAcls_result_response_matching_acls(const rd_kafka_DeleteAcls_result_response_t *result_response, size_t *matching_acls_cntp)
RD_EXPORT rd_kafka_t * rd_kafka_new(rd_kafka_type_t type, rd_kafka_conf_t *conf, char *errstr, size_t errstr_size)
Creates a new Kafka handle and starts its operation according to the specified type (RD_KAFKA_CONSUME...
RD_EXPORT const rd_kafka_DeleteTopics_result_t * rd_kafka_event_DeleteTopics_result(rd_kafka_event_t *rkev)
Get DeleteTopics result.
rd_kafka_event_t rd_kafka_AlterUserScramCredentials_result_t
Definition rdkafka.h:5892
RD_EXPORT const rd_kafka_error_t * rd_kafka_AlterUserScramCredentials_result_response_error(const rd_kafka_AlterUserScramCredentials_result_response_t *response)
Returns the error of a rd_kafka_AlterUserScramCredentials_result_response.
RD_EXPORT rd_kafka_queue_t * rd_kafka_queue_get_consumer(rd_kafka_t *rk)
RD_EXPORT void rd_kafka_NewTopic_destroy_array(rd_kafka_NewTopic_t **new_topics, size_t new_topic_cnt)
Helper function to destroy all NewTopic objects in the new_topics array (of new_topic_cnt elements)....
rd_kafka_event_t rd_kafka_DescribeAcls_result_t
Definition rdkafka.h:5860
RD_EXPORT rd_kafka_DeleteGroup_t * rd_kafka_DeleteGroup_new(const char *group)
Create a new DeleteGroup object. This object is later passed to rd_kafka_DeleteGroups().
RD_EXPORT const rd_kafka_DescribeAcls_result_t * rd_kafka_event_DescribeAcls_result(rd_kafka_event_t *rkev)
RD_EXPORT rd_kafka_UserScramCredentialAlteration_t * rd_kafka_UserScramCredentialDeletion_new(const char *username, rd_kafka_ScramMechanism_t mechanism)
Allocates a new UserScramCredentialDeletion given its fields.
RD_EXPORT const char * rd_kafka_topic_name(const rd_kafka_topic_t *rkt)
Returns the topic name.
RD_EXPORT const char * rd_kafka_get_debug_contexts(void)
Retrieve supported debug contexts for use with the "debug" configuration property....
RD_EXPORT const rd_kafka_AclOperation_t * rd_kafka_DescribeCluster_result_authorized_operations(const rd_kafka_DescribeCluster_result_t *result, size_t *cntp)
Gets the authorized ACL operations for the result cluster.
RD_EXPORT rd_kafka_ConfigResource_t * rd_kafka_ConfigResource_new(rd_kafka_ResourceType_t restype, const char *resname)
Create new ConfigResource object.
RD_EXPORT void rd_kafka_mem_free(rd_kafka_t *rk, void *ptr)
Free pointer returned by librdkafka.
rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_broker_state_change_t(rd_kafka_t *rk, int32_t broker_id, const char *secproto, const char *name, int port, const char *state, void *ic_opaque)
on_broker_state_change() is called just after a broker has been created or its state has been changed...
Definition rdkafka.h:6665
RD_EXPORT const char * rd_kafka_MemberDescription_host(const rd_kafka_MemberDescription_t *member)
Gets host of member.
RD_EXPORT const char * rd_kafka_consumer_group_metadata_group_id(const rd_kafka_consumer_group_metadata_t *group_metadata)
Get group id of a group metadata.
RD_EXPORT rd_kafka_message_t * rd_kafka_consume_queue(rd_kafka_queue_t *rkqu, int timeout_ms)
Consume from queue.
rd_kafka_AclOperation_t
Apache Kafka ACL operation types. Common type for multiple Admin API functions.
Definition rdkafka.h:7390
@ RD_KAFKA_ACL_OPERATION_ALL
Definition rdkafka.h:7394
@ RD_KAFKA_ACL_OPERATION_ALTER
Definition rdkafka.h:7399
@ RD_KAFKA_ACL_OPERATION_CLUSTER_ACTION
Definition rdkafka.h:7401
@ RD_KAFKA_ACL_OPERATION_IDEMPOTENT_WRITE
Definition rdkafka.h:7407
@ RD_KAFKA_ACL_OPERATION_DELETE
Definition rdkafka.h:7398
@ RD_KAFKA_ACL_OPERATION_WRITE
Definition rdkafka.h:7396
@ RD_KAFKA_ACL_OPERATION_ANY
Definition rdkafka.h:7392
@ RD_KAFKA_ACL_OPERATION_UNKNOWN
Definition rdkafka.h:7391
@ RD_KAFKA_ACL_OPERATION_CREATE
Definition rdkafka.h:7397
@ RD_KAFKA_ACL_OPERATION_ALTER_CONFIGS
Definition rdkafka.h:7405
@ RD_KAFKA_ACL_OPERATION_DESCRIBE_CONFIGS
Definition rdkafka.h:7403
@ RD_KAFKA_ACL_OPERATION_READ
Definition rdkafka.h:7395
@ RD_KAFKA_ACL_OPERATION_DESCRIBE
Definition rdkafka.h:7400
RD_EXPORT ssize_t rd_kafka_consume_batch(rd_kafka_topic_t *rkt, int32_t partition, int timeout_ms, rd_kafka_message_t **rkmessages, size_t rkmessages_size)
Consume up to rkmessages_size from topic rkt and partition putting a pointer to each message in the a...
RD_EXPORT int rd_kafka_event_log(rd_kafka_event_t *rkev, const char **fac, const char **str, int *level)
Extract log message from the event.
RD_EXPORT int rd_kafka_assignment_lost(rd_kafka_t *rk)
Check whether the consumer considers the current assignment to have been lost involuntarily....
RD_EXPORT void rd_kafka_conf_set_connect_cb(rd_kafka_conf_t *conf, int(*connect_cb)(int sockfd, const struct sockaddr *addr, int addrlen, const char *id, void *opaque))
Set connect callback.
RD_EXPORT void rd_kafka_UserScramCredentialAlteration_destroy_array(rd_kafka_UserScramCredentialAlteration_t **alterations, size_t alteration_cnt)
Destroys an array of UserScramCredentialAlteration.
RD_EXPORT const rd_kafka_error_t * rd_kafka_acl_result_error(const rd_kafka_acl_result_t *aclres)
RD_EXPORT const rd_kafka_DescribeConfigs_result_t * rd_kafka_event_DescribeConfigs_result(rd_kafka_event_t *rkev)
Get DescribeConfigs result.
RD_EXPORT rd_kafka_conf_res_t rd_kafka_conf_set_ssl_cert(rd_kafka_conf_t *conf, rd_kafka_cert_type_t cert_type, rd_kafka_cert_enc_t cert_enc, const void *buffer, size_t size, char *errstr, size_t errstr_size)
Set certificate/key cert_type from the cert_enc encoded memory at buffer of size bytes.
RD_EXPORT const char * rd_kafka_err2name(rd_kafka_resp_err_t err)
Returns the error code name (enum name).
RD_EXPORT rd_kafka_conf_res_t rd_kafka_topic_conf_get(const rd_kafka_topic_conf_t *conf, const char *name, char *dest, size_t *dest_size)
Retrieve topic configuration value for property name.
RD_EXPORT int rd_kafka_consume_callback(rd_kafka_topic_t *rkt, int32_t partition, int timeout_ms, void(*consume_cb)(rd_kafka_message_t *rkmessage, void *commit_opaque), void *commit_opaque)
Consumes messages from topic rkt and partition, calling the provided callback for each consumed messs...
RD_EXPORT int32_t rd_kafka_msg_partitioner_random(const rd_kafka_topic_t *rkt, const void *key, size_t keylen, int32_t partition_cnt, void *rkt_opaque, void *msg_opaque)
Random partitioner.
RD_EXPORT void rd_kafka_conf_set_stats_cb(rd_kafka_conf_t *conf, int(*stats_cb)(rd_kafka_t *rk, char *json, size_t json_len, void *opaque))
Set statistics callback in provided conf object.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_purge(rd_kafka_t *rk, int purge_flags)
Purge messages currently handled by the producer instance.
rd_kafka_event_t rd_kafka_DescribeCluster_result_t
Definition rdkafka.h:5888
RD_EXPORT rd_kafka_resp_err_t rd_kafka_oauthbearer_set_token_failure(rd_kafka_t *rk, const char *errstr)
SASL/OAUTHBEARER token refresh failure indicator.
RD_EXPORT char * rd_kafka_clusterid(rd_kafka_t *rk, int timeout_ms)
Returns the ClusterId as reported in broker metadata.
RD_EXPORT int rd_kafka_event_debug_contexts(rd_kafka_event_t *rkev, char *dst, size_t dstsize)
Extract log debug context from event.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_set_log_queue(rd_kafka_t *rk, rd_kafka_queue_t *rkqu)
Forward librdkafka logs (and debug) to the specified queue for serving with one of the ....
RD_EXPORT const char ** rd_kafka_conf_dump(rd_kafka_conf_t *conf, size_t *cntp)
Dump the configuration properties and values of conf to an array with "key", "value" pairs.
RD_EXPORT void rd_kafka_conf_set_background_event_cb(rd_kafka_conf_t *conf, void(*event_cb)(rd_kafka_t *rk, rd_kafka_event_t *rkev, void *opaque))
Generic event callback to be used with the event API to trigger callbacks for rd_kafka_event_t object...
RD_EXPORT rd_kafka_resp_err_t rd_kafka_header_get(const rd_kafka_headers_t *hdrs, size_t idx, const char *name, const void **valuep, size_t *sizep)
Iterator for headers matching name.
RD_EXPORT rd_kafka_consumer_group_state_t rd_kafka_consumer_group_state_code(const char *name)
Returns a code for a state name.
RD_EXPORT void rd_kafka_queue_forward(rd_kafka_queue_t *src, rd_kafka_queue_t *dst)
Forward/re-route queue src to dst. If dst is NULL the forwarding is removed.
RD_EXPORT rd_kafka_AclBinding_t * rd_kafka_AclBinding_new(rd_kafka_ResourceType_t restype, const char *name, rd_kafka_ResourcePatternType_t resource_pattern_type, const char *principal, const char *host, rd_kafka_AclOperation_t operation, rd_kafka_AclPermissionType_t permission_type, char *errstr, size_t errstr_size)
Create a new AclBinding object. This object is later passed to rd_kafka_CreateAcls().
RD_EXPORT void rd_kafka_DeleteConsumerGroupOffsets_destroy_array(rd_kafka_DeleteConsumerGroupOffsets_t **del_grpoffsets, size_t del_grpoffset_cnt)
Helper function to destroy all DeleteConsumerGroupOffsets objects in the del_grpoffsets array (of del...
rd_kafka_cert_type_t
SSL certificate type.
Definition rdkafka.h:2490
@ RD_KAFKA_CERT_PUBLIC_KEY
Definition rdkafka.h:2491
@ RD_KAFKA_CERT_PRIVATE_KEY
Definition rdkafka.h:2492
@ RD_KAFKA_CERT_CA
Definition rdkafka.h:2493
RD_EXPORT rd_kafka_conf_t * rd_kafka_conf_dup_filter(const rd_kafka_conf_t *conf, size_t filter_cnt, const char **filter)
Same as rd_kafka_conf_dup() but with an array of property name prefixes to filter out (ignore) when c...
RD_EXPORT rd_kafka_error_t * rd_kafka_sasl_set_credentials(rd_kafka_t *rk, const char *username, const char *password)
Sets SASL credentials used for SASL PLAIN and SCRAM mechanisms by this Kafka client.
RD_EXPORT rd_kafka_DeleteConsumerGroupOffsets_t * rd_kafka_DeleteConsumerGroupOffsets_new(const char *group, const rd_kafka_topic_partition_list_t *partitions)
Create a new DeleteConsumerGroupOffsets object. This object is later passed to rd_kafka_DeleteConsume...
rd_kafka_event_t rd_kafka_ListOffsets_result_t
Definition rdkafka.h:5894
rd_kafka_event_t rd_kafka_DescribeTopics_result_t
Definition rdkafka.h:5886
RD_EXPORT void rd_kafka_DeleteConsumerGroupOffsets_destroy(rd_kafka_DeleteConsumerGroupOffsets_t *del_grpoffsets)
Destroy and free a DeleteConsumerGroupOffsets object previously created with rd_kafka_DeleteConsumerG...
RD_EXPORT void rd_kafka_DeleteAcls(rd_kafka_t *rk, rd_kafka_AclBindingFilter_t **del_acls, size_t del_acls_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Delete acls matching the filteres provided in del_acls array of size del_acls_cnt.
RD_EXPORT const char * rd_kafka_ConsumerGroupDescription_partition_assignor(const rd_kafka_ConsumerGroupDescription_t *grpdesc)
Gets the partition assignor for the grpdesc group.
RD_EXPORT size_t rd_kafka_event_message_count(rd_kafka_event_t *rkev)
RD_EXPORT const rd_kafka_error_t * rd_kafka_UserScramCredentialsDescription_error(const rd_kafka_UserScramCredentialsDescription_t *description)
Returns the error associated with a UserScramCredentialsDescription.
RD_EXPORT void rd_kafka_topic_partition_list_add_range(rd_kafka_topic_partition_list_t *rktparlist, const char *topic, int32_t start, int32_t stop)
Add range of partitions from start to stop inclusive.
RD_EXPORT rd_kafka_error_t * rd_kafka_AdminOptions_set_match_consumer_group_types(rd_kafka_AdminOptions_t *options, const rd_kafka_consumer_group_type_t *consumer_group_types, size_t consumer_group_types_cnt)
Set consumer groups types to query for.
RD_EXPORT const rd_kafka_topic_partition_result_t ** rd_kafka_ElectLeaders_result_partitions(const rd_kafka_ElectLeaders_result_t *result, size_t *cntp)
Get the array of topic partition result objects from the elect leaders result event and populates the...
RD_EXPORT const char * rd_kafka_error_string(const rd_kafka_error_t *error)
RD_EXPORT const rd_kafka_CreatePartitions_result_t * rd_kafka_event_CreatePartitions_result(rd_kafka_event_t *rkev)
Get CreatePartitions result.
struct rd_kafka_ListConsumerGroupsResult_s rd_kafka_ListConsumerGroupsResult_t
Definition rdkafka.h:8639
RD_EXPORT rd_kafka_error_t * rd_kafka_ConfigResource_add_incremental_config(rd_kafka_ConfigResource_t *config, const char *name, rd_kafka_AlterConfigOpType_t op_type, const char *value)
Add the value of the configuration entry for a subsequent incremental alter config operation....
RD_EXPORT rd_kafka_headers_t * rd_kafka_headers_copy(const rd_kafka_headers_t *src)
Make a copy of headers list src.
RD_EXPORT void rd_kafka_conf_set_consume_cb(rd_kafka_conf_t *conf, void(*consume_cb)(rd_kafka_message_t *rkmessage, void *opaque))
Consumer: Set consume callback for use with rd_kafka_consumer_poll()
RD_EXPORT const rd_kafka_error_t * rd_kafka_topic_partition_result_error(const rd_kafka_topic_partition_result_t *partition_result)
RD_EXPORT const char * rd_kafka_topic_result_error_string(const rd_kafka_topic_result_t *topicres)
RD_EXPORT rd_kafka_resp_err_t rd_kafka_seek(rd_kafka_topic_t *rkt, int32_t partition, int64_t offset, int timeout_ms)
Seek consumer for topic+partition to offset which is either an absolute or logical offset.
RD_EXPORT const rd_kafka_DescribeConsumerGroups_result_t * rd_kafka_event_DescribeConsumerGroups_result(rd_kafka_event_t *rkev)
Get DescribeConsumerGroups result.
RD_EXPORT const char * rd_kafka_DescribeCluster_result_cluster_id(const rd_kafka_DescribeCluster_result_t *result)
Gets the cluster id for the result cluster.
RD_EXPORT void rd_kafka_DeleteRecords_destroy(rd_kafka_DeleteRecords_t *del_records)
Destroy and free a DeleteRecords object previously created with rd_kafka_DeleteRecords_new()
RD_EXPORT const rd_kafka_topic_result_t ** rd_kafka_DeleteTopics_result_topics(const rd_kafka_DeleteTopics_result_t *result, size_t *cntp)
Get an array of topic results from a DeleteTopics result.
RD_EXPORT const rd_kafka_AlterConfigs_result_t * rd_kafka_event_AlterConfigs_result(rd_kafka_event_t *rkev)
Get AlterConfigs result.
RD_EXPORT const rd_kafka_Node_t * rd_kafka_TopicPartitionInfo_leader(const rd_kafka_TopicPartitionInfo_t *partition)
Gets the partition leader for partition.
RD_EXPORT void rd_kafka_CreateTopics(rd_kafka_t *rk, rd_kafka_NewTopic_t **new_topics, size_t new_topic_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Create topics in cluster as specified by the new_topics array of size new_topic_cnt elements.
RD_EXPORT rd_kafka_error_t * rd_kafka_incremental_assign(rd_kafka_t *rk, const rd_kafka_topic_partition_list_t *partitions)
Incrementally add partitions to the current assignment.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_ConfigResource_error(const rd_kafka_ConfigResource_t *config)
RD_EXPORT rd_kafka_resp_err_t rd_kafka_NewPartitions_set_replica_assignment(rd_kafka_NewPartitions_t *new_parts, int32_t new_partition_idx, int32_t *broker_ids, size_t broker_id_cnt, char *errstr, size_t errstr_size)
Set the replica (broker id) assignment for new_partition_idx to the replica set in broker_ids (of bro...
RD_EXPORT rd_kafka_resp_err_t rd_kafka_error_code(const rd_kafka_error_t *error)
RD_EXPORT const rd_kafka_DescribeUserScramCredentials_result_t * rd_kafka_event_DescribeUserScramCredentials_result(rd_kafka_event_t *rkev)
Get DescribeUserScramCredentials result.
rd_kafka_conf_res_t() rd_kafka_interceptor_f_on_conf_set_t(rd_kafka_conf_t *conf, const char *name, const char *val, char *errstr, size_t errstr_size, void *ic_opaque)
on_conf_set() is called from rd_kafka_*_conf_set() in the order the interceptors were added.
Definition rdkafka.h:6343
RD_EXPORT rd_kafka_error_t * rd_kafka_commit_transaction(rd_kafka_t *rk, int timeout_ms)
Commit the current transaction (as started with rd_kafka_begin_transaction()).
RD_EXPORT rd_kafka_error_t * rd_kafka_error_new(rd_kafka_resp_err_t code, const char *fmt,...) RD_FORMAT(printf
Create a new error object with error code and optional human readable error string in fmt.
RD_EXPORT rd_kafka_error_t * rd_kafka_AdminOptions_set_isolation_level(rd_kafka_AdminOptions_t *options, rd_kafka_IsolationLevel_t value)
Set Isolation Level to an allowed rd_kafka_IsolationLevel_t value.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_header_get_last(const rd_kafka_headers_t *hdrs, const char *name, const void **valuep, size_t *sizep)
Find last header in list hdrs matching name.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_list_groups(rd_kafka_t *rk, const char *group, const struct rd_kafka_group_list **grplistp, int timeout_ms)
List and describe client groups in cluster.
rd_kafka_resp_err_t() rd_kafka_plugin_f_conf_init_t(rd_kafka_conf_t *conf, void **plug_opaquep, char *errstr, size_t errstr_size)
Plugin's configuration initializer method called each time the library is referenced from configurati...
Definition rdkafka.h:6257
RD_EXPORT rd_kafka_resp_err_t rd_kafka_position(rd_kafka_t *rk, rd_kafka_topic_partition_list_t *partitions)
Retrieve current positions (offsets) for topics+partitions.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_AdminOptions_set_operation_timeout(rd_kafka_AdminOptions_t *options, int timeout_ms, char *errstr, size_t errstr_size)
Sets the broker's operation timeout, such as the timeout for CreateTopics to complete the creation of...
RD_EXPORT void rd_kafka_DescribeTopics(rd_kafka_t *rk, const rd_kafka_TopicCollection_t *topics, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Describe topics as specified by the topics array of size topics_cnt elements.
RD_EXPORT const rd_kafka_Node_t * rd_kafka_ConsumerGroupDescription_coordinator(const rd_kafka_ConsumerGroupDescription_t *grpdesc)
Gets the coordinator for the grpdesc group.
RD_EXPORT const char * rd_kafka_AclBinding_principal(const rd_kafka_AclBinding_t *acl)
RD_EXPORT rd_kafka_TopicCollection_t * rd_kafka_TopicCollection_of_topic_names(const char **topics, size_t topics_cnt)
Creates a new TopicCollection for passing to rd_kafka_DescribeTopics.
RD_EXPORT void rd_kafka_DeleteGroup_destroy(rd_kafka_DeleteGroup_t *del_group)
Destroy and free a DeleteGroup object previously created with rd_kafka_DeleteGroup_new()
RD_EXPORT void rd_kafka_log_syslog(const rd_kafka_t *rk, int level, const char *fac, const char *buf)
Builtin log sink: print to syslog.
struct rd_kafka_DeleteTopic_s rd_kafka_DeleteTopic_t
Definition rdkafka.h:7564
rd_kafka_consumer_group_type_t
Consumer group type.
Definition rdkafka.h:5211
struct rd_kafka_ScramCredentialInfo_s rd_kafka_ScramCredentialInfo_t
Scram credential info. Mechanism and iterations for a SASL/SCRAM credential associated with a user.
Definition rdkafka.h:9514
RD_EXPORT void rd_kafka_topic_conf_set_msg_order_cmp(rd_kafka_topic_conf_t *topic_conf, int(*msg_order_cmp)(const rd_kafka_message_t *a, const rd_kafka_message_t *b))
Producer: Set message queueing order comparator callback.
RD_EXPORT const rd_kafka_error_t * rd_kafka_TopicDescription_error(const rd_kafka_TopicDescription_t *topicdesc)
Gets the error for the topicdesc topic.
RD_EXPORT rd_kafka_error_t * rd_kafka_consumer_group_metadata_read(rd_kafka_consumer_group_metadata_t **cgmdp, const void *buffer, size_t size)
Reads serialized consumer group metadata and returns a consumer group metadata object....
RD_EXPORT rd_kafka_resp_err_t rd_kafka_offset_store(rd_kafka_topic_t *rkt, int32_t partition, int64_t offset)
Store offset offset + 1 for topic rkt partition partition.
RD_EXPORT rd_kafka_topic_partition_list_t * rd_kafka_event_topic_partition_list(rd_kafka_event_t *rkev)
rd_kafka_AlterConfigOpType_t
Incremental alter configs operations.
Definition rdkafka.h:7911
RD_EXPORT void rd_kafka_DeleteGroup_destroy_array(rd_kafka_DeleteGroup_t **del_groups, size_t del_group_cnt)
Helper function to destroy all DeleteGroup objects in the del_groups array (of del_group_cnt elements...
RD_EXPORT const rd_kafka_group_result_t ** rd_kafka_DeleteConsumerGroupOffsets_result_groups(const rd_kafka_DeleteConsumerGroupOffsets_result_t *result, size_t *cntp)
Get an array of results from a DeleteConsumerGroupOffsets result.
RD_EXPORT const rd_kafka_error_t * rd_kafka_ConsumerGroupDescription_error(const rd_kafka_ConsumerGroupDescription_t *grpdesc)
Gets the error for the grpdesc group.
RD_EXPORT const char * rd_kafka_event_config_string(rd_kafka_event_t *rkev)
RD_EXPORT int rd_kafka_produce_batch(rd_kafka_topic_t *rkt, int32_t partition, int msgflags, rd_kafka_message_t *rkmessages, int message_cnt)
Produce multiple messages.
rd_kafka_event_t rd_kafka_ListConsumerGroups_result_t
Definition rdkafka.h:5874
RD_EXPORT const rd_kafka_ConfigResource_t ** rd_kafka_AlterConfigs_result_resources(const rd_kafka_AlterConfigs_result_t *result, size_t *cntp)
Get an array of resource results from a AlterConfigs result.
RD_EXPORT rd_kafka_ResourcePatternType_t rd_kafka_AclBinding_resource_pattern_type(const rd_kafka_AclBinding_t *acl)
RD_EXPORT int rd_kafka_ConfigEntry_is_sensitive(const rd_kafka_ConfigEntry_t *entry)
RD_EXPORT const rd_kafka_Node_t ** rd_kafka_TopicPartitionInfo_replicas(const rd_kafka_TopicPartitionInfo_t *partition, size_t *cntp)
Gets the partition replicas for partition.
RD_EXPORT void rd_kafka_ListConsumerGroupOffsets_destroy(rd_kafka_ListConsumerGroupOffsets_t *list_grpoffsets)
Destroy and free a ListConsumerGroupOffsets object previously created with rd_kafka_ListConsumerGroup...
RD_EXPORT void rd_kafka_DescribeAcls(rd_kafka_t *rk, rd_kafka_AclBindingFilter_t *acl_filter, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Describe acls matching the filter provided in acl_filter.
RD_EXPORT rd_kafka_error_t * rd_kafka_abort_transaction(rd_kafka_t *rk, int timeout_ms)
Aborts the ongoing transaction.
rd_kafka_admin_op_t
Admin operation enum name for use with rd_kafka_AdminOptions_new()
Definition rdkafka.h:7091
@ RD_KAFKA_ADMIN_OP_DESCRIBEUSERSCRAMCREDENTIALS
Definition rdkafka.h:7114
@ RD_KAFKA_ADMIN_OP_DELETERECORDS
Definition rdkafka.h:7098
@ RD_KAFKA_ADMIN_OP_CREATETOPICS
Definition rdkafka.h:7093
@ RD_KAFKA_ADMIN_OP_CREATEACLS
Definition rdkafka.h:7102
@ RD_KAFKA_ADMIN_OP_INCREMENTALALTERCONFIGS
Definition rdkafka.h:7112
@ RD_KAFKA_ADMIN_OP_DESCRIBEACLS
Definition rdkafka.h:7103
@ RD_KAFKA_ADMIN_OP_DESCRIBECLUSTER
Definition rdkafka.h:7118
@ RD_KAFKA_ADMIN_OP_DESCRIBECONSUMERGROUPS
Definition rdkafka.h:7106
@ RD_KAFKA_ADMIN_OP_DELETEGROUPS
Definition rdkafka.h:7099
@ RD_KAFKA_ADMIN_OP_ELECTLEADERS
Definition rdkafka.h:7120
@ RD_KAFKA_ADMIN_OP_ANY
Definition rdkafka.h:7092
@ RD_KAFKA_ADMIN_OP_DELETEACLS
Definition rdkafka.h:7104
@ RD_KAFKA_ADMIN_OP_DELETECONSUMERGROUPOFFSETS
Definition rdkafka.h:7101
@ RD_KAFKA_ADMIN_OP_DELETETOPICS
Definition rdkafka.h:7094
@ RD_KAFKA_ADMIN_OP_LISTCONSUMERGROUPOFFSETS
Definition rdkafka.h:7108
@ RD_KAFKA_ADMIN_OP_LISTCONSUMERGROUPS
Definition rdkafka.h:7105
@ RD_KAFKA_ADMIN_OP_LISTOFFSETS
Definition rdkafka.h:7119
@ RD_KAFKA_ADMIN_OP__CNT
Definition rdkafka.h:7121
@ RD_KAFKA_ADMIN_OP_ALTERCONSUMERGROUPOFFSETS
Definition rdkafka.h:7110
@ RD_KAFKA_ADMIN_OP_ALTERCONFIGS
Definition rdkafka.h:7096
@ RD_KAFKA_ADMIN_OP_CREATEPARTITIONS
Definition rdkafka.h:7095
@ RD_KAFKA_ADMIN_OP_ALTERUSERSCRAMCREDENTIALS
Definition rdkafka.h:7116
@ RD_KAFKA_ADMIN_OP_DESCRIBETOPICS
Definition rdkafka.h:7117
@ RD_KAFKA_ADMIN_OP_DESCRIBECONFIGS
Definition rdkafka.h:7097
struct rd_kafka_AlterConsumerGroupOffsets_s rd_kafka_AlterConsumerGroupOffsets_t
Definition rdkafka.h:9227
RD_EXPORT const char * rd_kafka_ResourceType_name(rd_kafka_ResourceType_t restype)
RD_EXPORT rd_kafka_resp_err_t rd_kafka_interceptor_add_on_response_received(rd_kafka_t *rk, const char *ic_name, rd_kafka_interceptor_f_on_response_received_t *on_response_received, void *ic_opaque)
Append an on_response_received() interceptor.
RD_EXPORT rd_kafka_message_t * rd_kafka_consume(rd_kafka_topic_t *rkt, int32_t partition, int timeout_ms)
Consume a single message from topic rkt and partition.
RD_EXPORT const rd_kafka_ScramCredentialInfo_t * rd_kafka_UserScramCredentialsDescription_scramcredentialinfo(const rd_kafka_UserScramCredentialsDescription_t *description, size_t idx)
Returns the ScramCredentialInfo at index idx of UserScramCredentialsDescription.
RD_EXPORT void rd_kafka_NewPartitions_destroy_array(rd_kafka_NewPartitions_t **new_parts, size_t new_parts_cnt)
Helper function to destroy all NewPartitions objects in the new_parts array (of new_parts_cnt element...
RD_EXPORT int rd_kafka_version(void)
Returns the librdkafka version as integer.
RD_EXPORT rd_kafka_Uuid_t * rd_kafka_Uuid_copy(const rd_kafka_Uuid_t *uuid)
Copies the given UUID.
RD_EXPORT int rd_kafka_topic_partition_list_del_by_idx(rd_kafka_topic_partition_list_t *rktparlist, int idx)
Delete partition from list by elems[] index.
RD_EXPORT size_t rd_kafka_UserScramCredentialsDescription_scramcredentialinfo_count(const rd_kafka_UserScramCredentialsDescription_t *description)
Returns the count of ScramCredentialInfos of a UserScramCredentialsDescription.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_metadata(rd_kafka_t *rk, int all_topics, rd_kafka_topic_t *only_rkt, const struct rd_kafka_metadata **metadatap, int timeout_ms)
Request Metadata from broker.
RD_EXPORT int32_t rd_kafka_consumer_group_metadata_generation_id(const rd_kafka_consumer_group_metadata_t *group_metadata)
Get the generation id (classic protocol) or member epoch (consumer protocol) of a group metadata.
struct rd_kafka_UserScramCredentialAlteration_s rd_kafka_UserScramCredentialAlteration_t
A request to alter a user's SASL/SCRAM credentials.
Definition rdkafka.h:9608
RD_EXPORT rd_kafka_UserScramCredentialAlteration_t * rd_kafka_UserScramCredentialUpsertion_new(const char *username, rd_kafka_ScramMechanism_t mechanism, int32_t iterations, const unsigned char *password, size_t password_size, const unsigned char *salt, size_t salt_size)
Allocates a new UserScramCredentialUpsertion given its fields. If salt isn't given a 64 B salt is gen...
RD_EXPORT rd_kafka_resp_err_t rd_kafka_interceptor_add_on_thread_exit(rd_kafka_t *rk, const char *ic_name, rd_kafka_interceptor_f_on_thread_exit_t *on_thread_exit, void *ic_opaque)
Append an on_thread_exit() interceptor.
RD_EXPORT int rd_kafka_consumer_closed(rd_kafka_t *rk)
rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_commit_t(rd_kafka_t *rk, const rd_kafka_topic_partition_list_t *offsets, rd_kafka_resp_err_t err, void *ic_opaque)
on_commit() is called on completed or failed offset commit. It is called from internal librdkafka thr...
Definition rdkafka.h:6523
RD_EXPORT rd_kafka_headers_t * rd_kafka_headers_new(size_t initial_count)
Create a new headers list.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_header_remove(rd_kafka_headers_t *hdrs, const char *name)
Remove all headers for the given key (if any).
RD_EXPORT RD_DEPRECATED int rd_kafka_errno(void)
Returns the thread-local system errno.
RD_EXPORT rd_kafka_topic_partition_list_t * rd_kafka_topic_partition_list_new(int size)
Create a new list/vector Topic+Partition container.
RD_EXPORT rd_kafka_topic_conf_t * rd_kafka_topic_conf_new(void)
Create topic configuration object.
RD_EXPORT void rd_kafka_DeleteTopic_destroy(rd_kafka_DeleteTopic_t *del_topic)
Destroy and free a DeleteTopic object previously created with rd_kafka_DeleteTopic_new()
RD_EXPORT const char * rd_kafka_Node_host(const rd_kafka_Node_t *node)
Get the host of node.
rd_kafka_consumer_group_state_t
Consumer group state.
Definition rdkafka.h:5196
rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_response_received_t(rd_kafka_t *rk, int sockfd, const char *brokername, int32_t brokerid, int16_t ApiKey, int16_t ApiVersion, int32_t CorrId, size_t size, int64_t rtt, rd_kafka_resp_err_t err, void *ic_opaque)
on_response_received() is called when a protocol response has been fully received from a broker TCP c...
Definition rdkafka.h:6588
RD_EXPORT int rd_kafka_consume_start_queue(rd_kafka_topic_t *rkt, int32_t partition, int64_t offset, rd_kafka_queue_t *rkqu)
Same as rd_kafka_consume_start() but re-routes incoming messages to the provided queue rkqu (which mu...
int rd_kafka_event_type_t
Event types.
Definition rdkafka.h:5565
RD_EXPORT size_t rd_kafka_queue_length(rd_kafka_queue_t *rkqu)
struct rd_kafka_ConsumerGroupListing_s rd_kafka_ConsumerGroupListing_t
ListConsumerGroups result for a single group.
Definition rdkafka.h:8636
RD_EXPORT void rd_kafka_conf_dump_free(const char **arr, size_t cnt)
Frees a configuration dump returned from rd_kafka_conf_dump() or `rd_kafka_topic_conf_dump().
struct rd_kafka_ConfigEntry_s rd_kafka_ConfigEntry_t
Definition rdkafka.h:7806
RD_EXPORT rd_kafka_resp_err_t rd_kafka_conf_interceptor_add_on_conf_destroy(rd_kafka_conf_t *conf, const char *ic_name, rd_kafka_interceptor_f_on_conf_destroy_t *on_conf_destroy, void *ic_opaque)
Append an on_conf_destroy() interceptor.
RD_EXPORT rd_kafka_consumer_group_type_t rd_kafka_ConsumerGroupDescription_type(const rd_kafka_ConsumerGroupDescription_t *grpdesc)
Gets type for the grpdesc group.
RD_EXPORT void rd_kafka_topic_destroy(rd_kafka_topic_t *rkt)
Loose application's topic handle refcount as previously created with rd_kafka_topic_new().
RD_EXPORT void rd_kafka_queue_destroy(rd_kafka_queue_t *rkqu)
RD_EXPORT void rd_kafka_ListConsumerGroupOffsets_destroy_array(rd_kafka_ListConsumerGroupOffsets_t **list_grpoffsets, size_t list_grpoffset_cnt)
Helper function to destroy all ListConsumerGroupOffsets objects in the list_grpoffsets array (of list...
struct rd_kafka_AdminOptions_s rd_kafka_AdminOptions_t
AdminOptions provides a generic mechanism for setting optional parameters for the Admin API requests.
Definition rdkafka.h:7136
RD_EXPORT int rd_kafka_topic_partition_list_del(rd_kafka_topic_partition_list_t *rktparlist, const char *topic, int32_t partition)
Delete partition from list.
RD_EXPORT const rd_kafka_group_result_t ** rd_kafka_AlterConsumerGroupOffsets_result_groups(const rd_kafka_AlterConsumerGroupOffsets_result_t *result, size_t *cntp)
Get an array of results from a AlterConsumerGroupOffsets result.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_topic_result_error(const rd_kafka_topic_result_t *topicres)
Topic result provides per-topic operation result information.
RD_EXPORT int rd_kafka_error_is_fatal(const rd_kafka_error_t *error)
RD_EXPORT rd_kafka_resp_err_t rd_kafka_conf_interceptor_add_on_conf_set(rd_kafka_conf_t *conf, const char *ic_name, rd_kafka_interceptor_f_on_conf_set_t *on_conf_set, void *ic_opaque)
Append an on_conf_set() interceptor.
RD_EXPORT char * rd_kafka_memberid(const rd_kafka_t *rk)
Returns this client's broker-assigned group member id.
RD_EXPORT void rd_kafka_conf_set_oauthbearer_token_refresh_cb(rd_kafka_conf_t *conf, void(*oauthbearer_token_refresh_cb)(rd_kafka_t *rk, const char *oauthbearer_config, void *opaque))
Set SASL/OAUTHBEARER token refresh callback in provided conf object.
RD_EXPORT void rd_kafka_AlterConsumerGroupOffsets_destroy(rd_kafka_AlterConsumerGroupOffsets_t *alter_grpoffsets)
Destroy and free a AlterConsumerGroupOffsets object previously created with rd_kafka_AlterConsumerGro...
RD_EXPORT rd_kafka_resp_err_t rd_kafka_AdminOptions_set_request_timeout(rd_kafka_AdminOptions_t *options, int timeout_ms, char *errstr, size_t errstr_size)
Sets the overall request timeout, including broker lookup, request transmission, operation time on br...
RD_EXPORT void rd_kafka_message_set_headers(rd_kafka_message_t *rkmessage, rd_kafka_headers_t *hdrs)
Replace the message's current headers with a new list.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_commit_message(rd_kafka_t *rk, const rd_kafka_message_t *rkmessage, int async)
Commit message's offset on broker for the message's partition. The committed offset is the message's ...
RD_EXPORT const char * rd_kafka_ResourcePatternType_name(rd_kafka_ResourcePatternType_t resource_pattern_type)
rd_kafka_vtype_t
Var-arg tag types.
Definition rdkafka.h:1144
@ RD_KAFKA_VTYPE_PARTITION
Definition rdkafka.h:1148
@ RD_KAFKA_VTYPE_HEADERS
Definition rdkafka.h:1160
@ RD_KAFKA_VTYPE_TOPIC
Definition rdkafka.h:1146
@ RD_KAFKA_VTYPE_MSGFLAGS
Definition rdkafka.h:1156
@ RD_KAFKA_VTYPE_RKT
Definition rdkafka.h:1147
@ RD_KAFKA_VTYPE_HEADER
Definition rdkafka.h:1158
@ RD_KAFKA_VTYPE_KEY
Definition rdkafka.h:1150
@ RD_KAFKA_VTYPE_END
Definition rdkafka.h:1145
@ RD_KAFKA_VTYPE_OPAQUE
Definition rdkafka.h:1151
@ RD_KAFKA_VTYPE_VALUE
Definition rdkafka.h:1149
@ RD_KAFKA_VTYPE_TIMESTAMP
Definition rdkafka.h:1157
struct rd_kafka_ConfigResource_s rd_kafka_ConfigResource_t
Definition rdkafka.h:7932
struct rd_kafka_DeleteGroup_s rd_kafka_DeleteGroup_t
Definition rdkafka.h:9052
RD_EXPORT rd_kafka_resp_err_t rd_kafka_poll_set_consumer(rd_kafka_t *rk)
Redirect the main (rd_kafka_poll()) queue to the KafkaConsumer's queue (rd_kafka_consumer_poll()).
RD_EXPORT const rd_kafka_ConfigEntry_t ** rd_kafka_ConfigResource_configs(const rd_kafka_ConfigResource_t *config, size_t *cntp)
Get an array of config entries from a ConfigResource object.
RD_EXPORT const char * rd_kafka_topic_result_name(const rd_kafka_topic_result_t *topicres)
RD_EXPORT void rd_kafka_AdminOptions_set_opaque(rd_kafka_AdminOptions_t *options, void *ev_opaque)
Set application opaque value that can be extracted from the result event using rd_kafka_event_opaque(...
RD_EXPORT rd_kafka_error_t * rd_kafka_consumer_close_queue(rd_kafka_t *rk, rd_kafka_queue_t *rkqu)
Asynchronously close the consumer.
RD_EXPORT RD_DEPRECATED void rd_kafka_set_logger(rd_kafka_t *rk, void(*func)(const rd_kafka_t *rk, int level, const char *fac, const char *buf))
Set logger function.
RD_EXPORT const char * rd_kafka_AclBinding_host(const rd_kafka_AclBinding_t *acl)
RD_EXPORT int rd_kafka_ConsumerGroupDescription_is_simple_consumer_group(const rd_kafka_ConsumerGroupDescription_t *grpdesc)
Is the grpdesc group a simple consumer group.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_event_error(rd_kafka_event_t *rkev)
RD_EXPORT void rd_kafka_headers_destroy(rd_kafka_headers_t *hdrs)
Destroy the headers list. The object and any returned value pointers are not usable after this call.
RD_EXPORT int32_t rd_kafka_message_broker_id(const rd_kafka_message_t *rkmessage)
Returns the broker id of the broker the message was produced to or fetched from.
rd_kafka_event_t rd_kafka_AlterConsumerGroupOffsets_result_t
Definition rdkafka.h:5882
RD_EXPORT void rd_kafka_ConfigResource_destroy_array(rd_kafka_ConfigResource_t **config, size_t config_cnt)
Helper function to destroy all ConfigResource objects in the configs array (of config_cnt elements)....
RD_EXPORT rd_kafka_resp_err_t rd_kafka_interceptor_add_on_thread_start(rd_kafka_t *rk, const char *ic_name, rd_kafka_interceptor_f_on_thread_start_t *on_thread_start, void *ic_opaque)
Append an on_thread_start() interceptor.
RD_EXPORT size_t rd_kafka_ConsumerGroupDescription_member_count(const rd_kafka_ConsumerGroupDescription_t *grpdesc)
Gets the members count of grpdesc group.
RD_EXPORT void rd_kafka_conf_properties_show(FILE *fp)
Prints a table to fp of all supported configuration properties, their default values as well as a des...
RD_EXPORT const char * rd_kafka_ConfigEntry_value(const rd_kafka_ConfigEntry_t *entry)
RD_EXPORT const rd_kafka_AlterUserScramCredentials_result_response_t ** rd_kafka_AlterUserScramCredentials_result_responses(const rd_kafka_AlterUserScramCredentials_result_t *result, size_t *cntp)
Get an array of responses from a AlterUserScramCredentials result.
rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_acknowledgement_t(rd_kafka_t *rk, rd_kafka_message_t *rkmessage, void *ic_opaque)
on_acknowledgement() is called to inform interceptors that a message was succesfully delivered or per...
Definition rdkafka.h:6475
RD_EXPORT const rd_kafka_conf_t * rd_kafka_conf(rd_kafka_t *rk)
struct rd_kafka_DeleteConsumerGroupOffsets_s rd_kafka_DeleteConsumerGroupOffsets_t
Definition rdkafka.h:9323
RD_EXPORT int rd_kafka_ConfigEntry_is_read_only(const rd_kafka_ConfigEntry_t *entry)
RD_EXPORT rd_kafka_AdminOptions_t * rd_kafka_AdminOptions_new(rd_kafka_t *rk, rd_kafka_admin_op_t for_api)
Create a new AdminOptions object.
rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_thread_start_t(rd_kafka_t *rk, rd_kafka_thread_type_t thread_type, const char *thread_name, void *ic_opaque)
on_thread_start() is called from a newly created librdkafka-managed thread.
Definition rdkafka.h:6618
RD_EXPORT int rd_kafka_wait_destroyed(int timeout_ms)
Wait for all rd_kafka_t objects to be destroyed.
RD_EXPORT const rd_kafka_error_t * rd_kafka_AclBinding_error(const rd_kafka_AclBinding_t *acl)
rd_kafka_ConfigSource_t
Apache Kafka config sources.
Definition rdkafka.h:7773
@ RD_KAFKA_CONFIG_SOURCE_UNKNOWN_CONFIG
Definition rdkafka.h:7776
@ RD_KAFKA_CONFIG_SOURCE__CNT
Definition rdkafka.h:7794
@ RD_KAFKA_CONFIG_SOURCE_DYNAMIC_BROKER_CONFIG
Definition rdkafka.h:7780
@ RD_KAFKA_CONFIG_SOURCE_DYNAMIC_TOPIC_CONFIG
Definition rdkafka.h:7778
@ RD_KAFKA_CONFIG_SOURCE_DEFAULT_CONFIG
Definition rdkafka.h:7789
@ RD_KAFKA_CONFIG_SOURCE_STATIC_BROKER_CONFIG
Definition rdkafka.h:7786
@ RD_KAFKA_CONFIG_SOURCE_DYNAMIC_DEFAULT_BROKER_CONFIG
Definition rdkafka.h:7783
@ RD_KAFKA_CONFIG_SOURCE_GROUP_CONFIG
Definition rdkafka.h:7791
RD_EXPORT int32_t rd_kafka_msg_partitioner_consistent(const rd_kafka_topic_t *rkt, const void *key, size_t keylen, int32_t partition_cnt, void *rkt_opaque, void *msg_opaque)
Consistent partitioner.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_message_headers(const rd_kafka_message_t *rkmessage, rd_kafka_headers_t **hdrsp)
Get the message header list.
RD_EXPORT void * rd_kafka_opaque(const rd_kafka_t *rk)
Retrieves the opaque pointer previously set with rd_kafka_conf_set_opaque()
RD_EXPORT void rd_kafka_DeleteRecords(rd_kafka_t *rk, rd_kafka_DeleteRecords_t **del_records, size_t del_record_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Delete records (messages) in topic partitions older than the offsets provided.
RD_EXPORT rd_kafka_AclPermissionType_t rd_kafka_AclBinding_permission_type(const rd_kafka_AclBinding_t *acl)
RD_EXPORT const rd_kafka_ConfigEntry_t ** rd_kafka_ConfigEntry_synonyms(const rd_kafka_ConfigEntry_t *entry, size_t *cntp)
RD_EXPORT const char * rd_kafka_AclBinding_name(const rd_kafka_AclBinding_t *acl)
RD_EXPORT const rd_kafka_message_t * rd_kafka_event_message_next(rd_kafka_event_t *rkev)
rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_destroy_t(rd_kafka_t *rk, void *ic_opaque)
on_destroy() is called from rd_kafka_destroy() or (rd_kafka_new() if rd_kafka_new() fails during init...
Definition rdkafka.h:6424
RD_EXPORT rd_kafka_DeleteTopic_t * rd_kafka_DeleteTopic_new(const char *topic)
Create a new DeleteTopic object. This object is later passed to rd_kafka_DeleteTopics().
RD_EXPORT const rd_kafka_TopicDescription_t ** rd_kafka_DescribeTopics_result_topics(const rd_kafka_DescribeTopics_result_t *result, size_t *cntp)
Get an array of topic results from a DescribeTopics result.
RD_EXPORT const char * rd_kafka_consumer_group_type_name(rd_kafka_consumer_group_type_t type)
Returns a name for a group type code.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_flush(rd_kafka_t *rk, int timeout_ms)
Wait until all outstanding produce requests, et.al, are completed. This should typically be done prio...
RD_EXPORT void rd_kafka_AlterUserScramCredentials(rd_kafka_t *rk, rd_kafka_UserScramCredentialAlteration_t **alterations, size_t alteration_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Alter SASL/SCRAM credentials. This operation is supported by brokers with version 2....
RD_EXPORT const char * rd_kafka_UserScramCredentialsDescription_user(const rd_kafka_UserScramCredentialsDescription_t *description)
Returns the username of a UserScramCredentialsDescription.
RD_EXPORT int rd_kafka_error_txn_requires_abort(const rd_kafka_error_t *error)
RD_EXPORT const rd_kafka_topic_partition_t * rd_kafka_topic_partition_result_partition(const rd_kafka_topic_partition_result_t *partition_result)
Topic Partition Result provides per-topic+partition operation result Consists of TopicPartition objec...
RD_EXPORT void rd_kafka_group_list_destroy(const struct rd_kafka_group_list *grplist)
Release list memory.
rd_kafka_AclPermissionType_t
Apache Kafka ACL permission types.
Definition rdkafka.h:9766
@ RD_KAFKA_ACL_PERMISSION_TYPE_DENY
Definition rdkafka.h:9770
@ RD_KAFKA_ACL_PERMISSION_TYPE_ALLOW
Definition rdkafka.h:9771
@ RD_KAFKA_ACL_PERMISSION_TYPE_UNKNOWN
Definition rdkafka.h:9767
@ RD_KAFKA_ACL_PERMISSION_TYPE_ANY
Definition rdkafka.h:9768
rd_kafka_thread_type_t
librdkafka internal thread type.
Definition rdkafka.h:5502
@ RD_KAFKA_THREAD_MAIN
Definition rdkafka.h:5503
@ RD_KAFKA_THREAD_BACKGROUND
Definition rdkafka.h:5504
@ RD_KAFKA_THREAD_BROKER
Definition rdkafka.h:5505
RD_EXPORT int rd_kafka_ConsumerGroupListing_is_simple_consumer_group(const rd_kafka_ConsumerGroupListing_t *grplist)
Is the grplist group a simple consumer group.
RD_EXPORT void rd_kafka_destroy_flags(rd_kafka_t *rk, int flags)
Destroy Kafka handle according to specified destroy flags.
RD_EXPORT rd_kafka_AclOperation_t rd_kafka_AclBinding_operation(const rd_kafka_AclBinding_t *acl)
RD_EXPORT const char * rd_kafka_Node_rack(const rd_kafka_Node_t *node)
Get the rack of node.
RD_EXPORT void rd_kafka_ListConsumerGroupOffsets(rd_kafka_t *rk, rd_kafka_ListConsumerGroupOffsets_t **list_grpoffsets, size_t list_grpoffsets_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
List committed offsets for a set of partitions in a consumer group.
RD_EXPORT void rd_kafka_conf_set_closesocket_cb(rd_kafka_conf_t *conf, int(*closesocket_cb)(int sockfd, void *opaque))
Set close socket callback.
RD_EXPORT int32_t rd_kafka_msg_partitioner_murmur2_random(const rd_kafka_topic_t *rkt, const void *key, size_t keylen, int32_t partition_cnt, void *rkt_opaque, void *msg_opaque)
Consistent-Random Murmur2 partitioner (Java compatible).
RD_EXPORT rd_kafka_resp_err_t rd_kafka_subscription(rd_kafka_t *rk, rd_kafka_topic_partition_list_t **topics)
Returns the current topic subscription.
RD_EXPORT rd_kafka_type_t rd_kafka_type(const rd_kafka_t *rk)
Returns Kafka handle type.
RD_EXPORT void rd_kafka_topic_partition_set_leader_epoch(rd_kafka_topic_partition_t *rktpar, int32_t leader_epoch)
Sets the offset leader epoch (use -1 to clear).
rd_kafka_cert_enc_t
SSL certificate encoding.
Definition rdkafka.h:2504
@ RD_KAFKA_CERT_ENC_PEM
Definition rdkafka.h:2507
@ RD_KAFKA_CERT_ENC_PKCS12
Definition rdkafka.h:2505
@ RD_KAFKA_CERT_ENC_DER
Definition rdkafka.h:2506
RD_EXPORT int rd_kafka_brokers_add(rd_kafka_t *rk, const char *brokerlist)
Adds one or more brokers to the kafka handle's list of initial bootstrap brokers.
RD_EXPORT int rd_kafka_unittest(void)
Run librdkafka's built-in unit-tests.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_commit(rd_kafka_t *rk, const rd_kafka_topic_partition_list_t *offsets, int async)
Commit offsets on broker for the provided list of partitions.
RD_EXPORT const char * rd_kafka_MemberDescription_consumer_id(const rd_kafka_MemberDescription_t *member)
Gets consumer id of member.
RD_EXPORT int64_t rd_kafka_message_latency(const rd_kafka_message_t *rkmessage)
Returns the latency for a produced message measured from the produce() call.
RD_EXPORT const char * rd_kafka_ConfigSource_name(rd_kafka_ConfigSource_t confsource)
RD_EXPORT rd_kafka_resp_err_t rd_kafka_interceptor_add_on_acknowledgement(rd_kafka_t *rk, const char *ic_name, rd_kafka_interceptor_f_on_acknowledgement_t *on_acknowledgement, void *ic_opaque)
Append an on_acknowledgement() interceptor.
rd_kafka_event_t rd_kafka_DeleteGroups_result_t
Definition rdkafka.h:5878
rd_kafka_event_t rd_kafka_ElectLeaders_result_t
Definition rdkafka.h:5896
RD_EXPORT rd_kafka_conf_res_t rd_kafka_conf_set(rd_kafka_conf_t *conf, const char *name, const char *value, char *errstr, size_t errstr_size)
Sets a configuration property.
rd_kafka_event_t rd_kafka_DescribeConfigs_result_t
Definition rdkafka.h:5870
RD_EXPORT uint16_t rd_kafka_Node_port(const rd_kafka_Node_t *node)
Get the port of node.
RD_EXPORT rd_kafka_message_t * rd_kafka_consumer_poll(rd_kafka_t *rk, int timeout_ms)
Poll the consumer for messages or events.
RD_EXPORT int32_t rd_kafka_controllerid(rd_kafka_t *rk, int timeout_ms)
Returns the current ControllerId as reported in broker metadata.
RD_EXPORT const rd_kafka_MemberAssignment_t * rd_kafka_MemberDescription_target_assignment(const rd_kafka_MemberDescription_t *member)
Gets target assignment of member.
RD_EXPORT int rd_kafka_ConfigEntry_is_default(const rd_kafka_ConfigEntry_t *entry)
RD_EXPORT rd_kafka_ScramMechanism_t rd_kafka_ScramCredentialInfo_mechanism(const rd_kafka_ScramCredentialInfo_t *scram_credential_info)
Returns the mechanism of a given ScramCredentialInfo.
RD_EXPORT void rd_kafka_conf_set_events(rd_kafka_conf_t *conf, int events)
Enable event sourcing. events is a bitmask of RD_KAFKA_EVENT_* of events to enable for consumption by...
RD_EXPORT rd_kafka_conf_res_t rd_kafka_conf_get(const rd_kafka_conf_t *conf, const char *name, char *dest, size_t *dest_size)
Retrieve configuration value for property name.
RD_EXPORT const rd_kafka_ListOffsetsResultInfo_t ** rd_kafka_ListOffsets_result_infos(const rd_kafka_ListOffsets_result_t *result, size_t *cntp)
Returns the array of ListOffsetsResultInfo in result and populates the size of the array in cntp.
RD_EXPORT RD_DEPRECATED rd_kafka_resp_err_t rd_kafka_errno2err(int errnox)
Converts the system errno value errnox to a rd_kafka_resp_err_t error code upon failure from the foll...
RD_EXPORT int32_t rd_kafka_msg_partitioner_fnv1a(const rd_kafka_topic_t *rkt, const void *key, size_t keylen, int32_t partition_cnt, void *rkt_opaque, void *msg_opaque)
FNV-1a partitioner.
rd_kafka_event_t rd_kafka_DeleteConsumerGroupOffsets_result_t
Definition rdkafka.h:5880
RD_EXPORT void rd_kafka_topic_conf_set_partitioner_cb(rd_kafka_topic_conf_t *topic_conf, int32_t(*partitioner)(const rd_kafka_topic_t *rkt, const void *keydata, size_t keylen, int32_t partition_cnt, void *rkt_opaque, void *msg_opaque))
Producer: Set partitioner callback in provided topic conf object.
RD_EXPORT const char * rd_kafka_err2str(rd_kafka_resp_err_t err)
Returns a human readable representation of a kafka error.
RD_EXPORT const char * rd_kafka_event_stats(rd_kafka_event_t *rkev)
Extract stats from the event.
RD_EXPORT void rd_kafka_destroy(rd_kafka_t *rk)
Destroy Kafka handle.
RD_EXPORT int rd_kafka_Node_id(const rd_kafka_Node_t *node)
Get the id of node.
RD_EXPORT rd_kafka_error_t * rd_kafka_sasl_background_callbacks_enable(rd_kafka_t *rk)
Enable SASL OAUTHBEARER refresh callbacks on the librdkafka background thread.
RD_EXPORT void rd_kafka_conf_set_dr_msg_cb(rd_kafka_conf_t *conf, void(*dr_msg_cb)(rd_kafka_t *rk, const rd_kafka_message_t *rkmessage, void *opaque))
Producer: Set delivery report callback in provided conf object.
RD_EXPORT rd_kafka_error_t * rd_kafka_offset_store_message(rd_kafka_message_t *rkmessage)
Store offset +1 for the consumed message.
RD_EXPORT rd_kafka_consumer_group_type_t rd_kafka_consumer_group_type_code(const char *name)
Returns a code for a group type name.
RD_EXPORT const rd_kafka_ListConsumerGroups_result_t * rd_kafka_event_ListConsumerGroups_result(rd_kafka_event_t *rkev)
Get ListConsumerGroups result.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_producev(rd_kafka_t *rk,...)
Produce and send a single message to broker.
RD_EXPORT void rd_kafka_topic_conf_destroy(rd_kafka_topic_conf_t *topic_conf)
Destroys a topic conf object.
RD_EXPORT const char * rd_kafka_consumer_group_metadata_group_instance_id(const rd_kafka_consumer_group_metadata_t *group_metadata)
Get group instance id of a group metadata.
RD_EXPORT rd_kafka_NewTopic_t * rd_kafka_NewTopic_new(const char *topic, int num_partitions, int replication_factor, char *errstr, size_t errstr_size)
Create a new NewTopic object. This object is later passed to rd_kafka_CreateTopics().
RD_EXPORT void rd_kafka_topic_partition_destroy(rd_kafka_topic_partition_t *rktpar)
Destroy a rd_kafka_topic_partition_t.
RD_EXPORT const rd_kafka_Node_t ** rd_kafka_TopicPartitionInfo_isr(const rd_kafka_TopicPartitionInfo_t *partition, size_t *cntp)
Gets the partition in-sync replicas for partition.
RD_EXPORT const rd_kafka_group_result_t ** rd_kafka_DeleteGroups_result_groups(const rd_kafka_DeleteGroups_result_t *result, size_t *cntp)
Get an array of group results from a DeleteGroups result.
RD_EXPORT int32_t rd_kafka_msg_partitioner_fnv1a_random(const rd_kafka_topic_t *rkt, const void *key, size_t keylen, int32_t partition_cnt, void *rkt_opaque, void *msg_opaque)
Consistent-Random FNV-1a partitioner.
RD_EXPORT void rd_kafka_DescribeConfigs(rd_kafka_t *rk, rd_kafka_ConfigResource_t **configs, size_t config_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Get configuration for the specified resources in configs.
rd_kafka_type_t
rd_kafka_t handle type.
Definition rdkafka.h:209
@ RD_KAFKA_CONSUMER
Definition rdkafka.h:211
@ RD_KAFKA_PRODUCER
Definition rdkafka.h:210
RD_EXPORT rd_kafka_conf_res_t rd_kafka_topic_conf_set(rd_kafka_topic_conf_t *conf, const char *name, const char *value, char *errstr, size_t errstr_size)
Sets a single rd_kafka_topic_conf_t value by property name.
RD_EXPORT void rd_kafka_ElectLeaders(rd_kafka_t *rk, rd_kafka_ElectLeaders_t *elect_leaders, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Elect Leaders for the provided Topic Partitions according to the specified election type.
RD_EXPORT void rd_kafka_DescribeUserScramCredentials(rd_kafka_t *rk, const char **users, size_t user_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Describe SASL/SCRAM credentials. This operation is supported by brokers with version 2....
RD_EXPORT const char * rd_kafka_version_str(void)
Returns the librdkafka version as string.
RD_EXPORT void rd_kafka_set_log_level(rd_kafka_t *rk, int level)
Specifies the maximum logging level emitted by internal kafka logging and debugging.
RD_EXPORT const rd_kafka_AlterConsumerGroupOffsets_result_t * rd_kafka_event_AlterConsumerGroupOffsets_result(rd_kafka_event_t *rkev)
Get AlterConsumerGroupOffsets result.
RD_EXPORT void rd_kafka_queue_io_event_enable(rd_kafka_queue_t *rkqu, int fd, const void *payload, size_t size)
Enable IO event triggering for queue.
rd_kafka_ResourcePatternType_t
Apache Kafka pattern types.
Definition rdkafka.h:7893
@ RD_KAFKA_RESOURCE_PATTERN_LITERAL
Definition rdkafka.h:7901
@ RD_KAFKA_RESOURCE_PATTERN_ANY
Definition rdkafka.h:7897
@ RD_KAFKA_RESOURCE_PATTERN_MATCH
Definition rdkafka.h:7899
@ RD_KAFKA_RESOURCE_PATTERN_PREFIXED
Definition rdkafka.h:7903
@ RD_KAFKA_RESOURCE_PATTERN_UNKNOWN
Definition rdkafka.h:7895
RD_EXPORT rd_kafka_event_t * rd_kafka_queue_poll(rd_kafka_queue_t *rkqu, int timeout_ms)
Poll a queue for an event for max timeout_ms.
RD_EXPORT rd_kafka_consumer_group_state_t rd_kafka_ConsumerGroupDescription_state(const rd_kafka_ConsumerGroupDescription_t *grpdesc)
Gets state for the grpdesc group.
RD_EXPORT int rd_kafka_error_is_retriable(const rd_kafka_error_t *error)
rd_kafka_ResourceType_t
Apache Kafka resource types.
Definition rdkafka.h:7879
@ RD_KAFKA_RESOURCE_BROKER
Definition rdkafka.h:7884
@ RD_KAFKA_RESOURCE__CNT
Definition rdkafka.h:7886
@ RD_KAFKA_RESOURCE_TOPIC
Definition rdkafka.h:7882
@ RD_KAFKA_RESOURCE_GROUP
Definition rdkafka.h:7883
@ RD_KAFKA_RESOURCE_TRANSACTIONAL_ID
Definition rdkafka.h:7885
@ RD_KAFKA_RESOURCE_UNKNOWN
Definition rdkafka.h:7880
@ RD_KAFKA_RESOURCE_ANY
Definition rdkafka.h:7881
struct rd_kafka_ConsumerGroupDescription_s rd_kafka_ConsumerGroupDescription_t
DescribeConsumerGroups result type.
Definition rdkafka.h:8751
RD_EXPORT void rd_kafka_conf_set_error_cb(rd_kafka_conf_t *conf, void(*error_cb)(rd_kafka_t *rk, int err, const char *reason, void *opaque))
Set error callback in provided conf object.
RD_EXPORT int rd_kafka_consume_stop(rd_kafka_topic_t *rkt, int32_t partition)
Stop consuming messages for topic rkt and partition, purging all messages currently in the local queu...
RD_EXPORT const rd_kafka_error_t * rd_kafka_DeleteAcls_result_response_error(const rd_kafka_DeleteAcls_result_response_t *result_response)
RD_EXPORT void rd_kafka_conf_enable_sasl_queue(rd_kafka_conf_t *conf, int enable)
Enable/disable creation of a queue specific to SASL events and callbacks.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_resume_partitions(rd_kafka_t *rk, rd_kafka_topic_partition_list_t *partitions)
Resume producing consumption for the provided list of partitions.
RD_EXPORT void rd_kafka_ConfigResource_destroy(rd_kafka_ConfigResource_t *config)
Destroy and free a ConfigResource object previously created with rd_kafka_ConfigResource_new()
RD_EXPORT const char * rd_kafka_event_name(const rd_kafka_event_t *rkev)
RD_EXPORT void rd_kafka_DeleteTopic_destroy_array(rd_kafka_DeleteTopic_t **del_topics, size_t del_topic_cnt)
Helper function to destroy all DeleteTopic objects in the del_topics array (of del_topic_cnt elements...
RD_EXPORT rd_kafka_NewPartitions_t * rd_kafka_NewPartitions_new(const char *topic, size_t new_total_cnt, char *errstr, size_t errstr_size)
Create a new NewPartitions. This object is later passed to rd_kafka_CreatePartitions() to increase th...
RD_EXPORT rd_kafka_resp_err_t rd_kafka_get_watermark_offsets(rd_kafka_t *rk, const char *topic, int32_t partition, int64_t *low, int64_t *high)
Get last known low (oldest/beginning) and high (newest/end) offsets for partition.
RD_EXPORT rd_kafka_AlterConsumerGroupOffsets_t * rd_kafka_AlterConsumerGroupOffsets_new(const char *group_id, const rd_kafka_topic_partition_list_t *partitions)
Create a new AlterConsumerGroupOffsets object. This object is later passed to rd_kafka_AlterConsumerG...
RD_EXPORT int rd_kafka_topic_partition_available(const rd_kafka_topic_t *rkt, int32_t partition)
Check if partition is available (has a leader broker).
RD_EXPORT const rd_kafka_ConsumerGroupDescription_t ** rd_kafka_DescribeConsumerGroups_result_groups(const rd_kafka_DescribeConsumerGroups_result_t *result, size_t *cntp)
Get an array of group results from a DescribeConsumerGroups result.
RD_EXPORT void rd_kafka_ListConsumerGroups(rd_kafka_t *rk, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
List the consumer groups available in the cluster.
RD_EXPORT rd_kafka_conf_t * rd_kafka_conf_new(void)
Create configuration object.
RD_EXPORT const char * rd_kafka_ConsumerGroupListing_group_id(const rd_kafka_ConsumerGroupListing_t *grplist)
Gets the group id for the grplist group.
rd_kafka_msg_status_t
Message persistence status can be used by the application to find out if a produced message was persi...
Definition rdkafka.h:1630
@ RD_KAFKA_MSG_STATUS_POSSIBLY_PERSISTED
Definition rdkafka.h:1639
@ RD_KAFKA_MSG_STATUS_PERSISTED
Definition rdkafka.h:1644
@ RD_KAFKA_MSG_STATUS_NOT_PERSISTED
Definition rdkafka.h:1634
RD_EXPORT const rd_kafka_DeleteAcls_result_response_t ** rd_kafka_DeleteAcls_result_responses(const rd_kafka_DeleteAcls_result_t *result, size_t *cntp)
Get an array of DeleteAcls result responses from a DeleteAcls result.
RD_EXPORT int rd_kafka_outq_len(rd_kafka_t *rk)
Returns the current out queue length.
RD_EXPORT int rd_kafka_poll(rd_kafka_t *rk, int timeout_ms)
Polls the provided kafka handle for events.
RD_EXPORT rd_kafka_consumer_group_metadata_t * rd_kafka_consumer_group_metadata(rd_kafka_t *rk)
rd_kafka_event_t rd_kafka_AlterConfigs_result_t
Definition rdkafka.h:5866
rd_kafka_conf_res_t
Configuration result type.
Definition rdkafka.h:1753
@ RD_KAFKA_CONF_OK
Definition rdkafka.h:1758
@ RD_KAFKA_CONF_INVALID
Definition rdkafka.h:1755
@ RD_KAFKA_CONF_UNKNOWN
Definition rdkafka.h:1754
RD_EXPORT const rd_kafka_error_t ** rd_kafka_ListConsumerGroups_result_errors(const rd_kafka_ListConsumerGroups_result_t *result, size_t *cntp)
Get an array of errors from a ListConsumerGroups call result.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_header_add(rd_kafka_headers_t *hdrs, const char *name, ssize_t name_size, const void *value, ssize_t value_size)
Add header with name name and value val (copied) of size size (not including null-terminator).
RD_EXPORT int32_t rd_kafka_topic_partition_get_leader_epoch(const rd_kafka_topic_partition_t *rktpar)
RD_EXPORT rd_kafka_consumer_group_type_t rd_kafka_ConsumerGroupListing_type(const rd_kafka_ConsumerGroupListing_t *grplist)
Gets type for the grplist group.
RD_EXPORT const rd_kafka_topic_partition_t * rd_kafka_ListOffsetsResultInfo_topic_partition(const rd_kafka_ListOffsetsResultInfo_t *result_info)
Returns the topic partition of the passed result_info.
RD_EXPORT int32_t rd_kafka_message_leader_epoch(const rd_kafka_message_t *rkmessage)
RD_EXPORT const rd_kafka_IncrementalAlterConfigs_result_t * rd_kafka_event_IncrementalAlterConfigs_result(rd_kafka_event_t *rkev)
Get IncrementalAlterConfigs result.
RD_EXPORT rd_kafka_topic_partition_t * rd_kafka_event_topic_partition(rd_kafka_event_t *rkev)
RD_EXPORT rd_kafka_consumer_group_metadata_t * rd_kafka_consumer_group_metadata_new(const char *group_id)
Create a new consumer group metadata object. This is typically only used for writing tests.
RD_EXPORT const char * rd_kafka_message_errstr(const rd_kafka_message_t *rkmessage)
Returns the error string for an errored rd_kafka_message_t or NULL if there was no error.
RD_EXPORT void rd_kafka_NewPartitions_destroy(rd_kafka_NewPartitions_t *new_parts)
Destroy and free a NewPartitions object previously created with rd_kafka_NewPartitions_new()
RD_EXPORT rd_kafka_resp_err_t rd_kafka_NewTopic_set_replica_assignment(rd_kafka_NewTopic_t *new_topic, int32_t partition, int32_t *broker_ids, size_t broker_id_cnt, char *errstr, size_t errstr_size)
Set the replica (broker) assignment for partition to the replica set in broker_ids (of broker_id_cnt ...
RD_EXPORT void rd_kafka_AlterConfigs(rd_kafka_t *rk, rd_kafka_ConfigResource_t **configs, size_t config_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Update the configuration for the specified resources. Updates are not transactional so they may succe...
RD_EXPORT void rd_kafka_AdminOptions_destroy(rd_kafka_AdminOptions_t *options)
Destroy a AdminOptions object.
RD_EXPORT rd_kafka_conf_res_t rd_kafka_conf_set_ssl_cert_verify_cb(rd_kafka_conf_t *conf, int(*ssl_cert_verify_cb)(rd_kafka_t *rk, const char *broker_name, int32_t broker_id, int *x509_error, int depth, const char *buf, size_t size, char *errstr, size_t errstr_size, void *opaque))
Sets the verification callback of the broker certificate.
RD_EXPORT rd_kafka_error_t * rd_kafka_produceva(rd_kafka_t *rk, const rd_kafka_vu_t *vus, size_t cnt)
Produce and send a single message to broker.
RD_EXPORT int rd_kafka_consume_start(rd_kafka_topic_t *rkt, int32_t partition, int64_t offset)
Start consuming messages for topic rkt and partition at offset offset which may either be an absolute...
RD_EXPORT int rd_kafka_produce(rd_kafka_topic_t *rkt, int32_t partition, int msgflags, void *payload, size_t len, const void *key, size_t keylen, void *msg_opaque)
Produce and send a single message to broker.
RD_EXPORT void * rd_kafka_topic_opaque(const rd_kafka_topic_t *rkt)
Get the rkt_opaque pointer that was set in the topic configuration with rd_kafka_topic_conf_set_opaqu...
RD_EXPORT rd_kafka_resp_err_t rd_kafka_offsets_for_times(rd_kafka_t *rk, rd_kafka_topic_partition_list_t *offsets, int timeout_ms)
Look up the offsets for the given partitions by timestamp.
rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_consume_t(rd_kafka_t *rk, rd_kafka_message_t *rkmessage, void *ic_opaque)
on_consume() is called just prior to passing the message to the application in rd_kafka_consumer_poll...
Definition rdkafka.h:6497
RD_EXPORT rd_kafka_resp_err_t rd_kafka_AdminOptions_set_broker(rd_kafka_AdminOptions_t *options, int32_t broker_id, char *errstr, size_t errstr_size)
Override what broker the Admin request will be sent to.
RD_EXPORT const rd_kafka_Node_t * rd_kafka_DescribeCluster_result_controller(const rd_kafka_DescribeCluster_result_t *result)
Gets the current controller for the result cluster.
RD_EXPORT void rd_kafka_DeleteTopics(rd_kafka_t *rk, rd_kafka_DeleteTopic_t **del_topics, size_t del_topic_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Delete topics from cluster as specified by the topics array of size topic_cnt elements.
RD_EXPORT const rd_kafka_AclOperation_t * rd_kafka_TopicDescription_authorized_operations(const rd_kafka_TopicDescription_t *topicdesc, size_t *cntp)
Gets the topic authorized ACL operations for the topicdesc topic.
RD_EXPORT const rd_kafka_DeleteGroups_result_t * rd_kafka_event_DeleteGroups_result(rd_kafka_event_t *rkev)
Get DeleteGroups result.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_last_error(void)
Returns the last error code generated by a legacy API call in the current thread.
RD_EXPORT const rd_kafka_ListConsumerGroupOffsets_result_t * rd_kafka_event_ListConsumerGroupOffsets_result(rd_kafka_event_t *rkev)
Get ListConsumerGroupOffsets result.
struct rd_kafka_AclBinding_s rd_kafka_AclBinding_t
ACL Binding is used to create access control lists.
Definition rdkafka.h:9741
rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_thread_exit_t(rd_kafka_t *rk, rd_kafka_thread_type_t thread_type, const char *thread_name, void *ic_opaque)
on_thread_exit() is called just prior to a librdkafka-managed thread exiting from the exiting thread ...
Definition rdkafka.h:6644
RD_EXPORT const rd_kafka_DeleteAcls_result_t * rd_kafka_event_DeleteAcls_result(rd_kafka_event_t *rkev)
RD_EXPORT rd_kafka_resp_err_t rd_kafka_AdminOptions_set_validate_only(rd_kafka_AdminOptions_t *options, int true_or_false, char *errstr, size_t errstr_size)
Tell broker to only validate the request, without performing the requested operation (create topics,...
RD_EXPORT const char * rd_kafka_ConfigResource_error_string(const rd_kafka_ConfigResource_t *config)
rd_kafka_ElectionType_t
Apache Kafka Election Types.
Definition rdkafka.h:10066
@ RD_KAFKA_ELECTION_TYPE_PREFERRED
Definition rdkafka.h:10067
@ RD_KAFKA_ELECTION_TYPE_UNCLEAN
Definition rdkafka.h:10068
rd_kafka_event_t rd_kafka_IncrementalAlterConfigs_result_t
Definition rdkafka.h:5868
RD_EXPORT rd_kafka_queue_t * rd_kafka_queue_get_background(rd_kafka_t *rk)
RD_EXPORT void rd_kafka_DescribeCluster(rd_kafka_t *rk, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Describes the cluster.
RD_EXPORT int rd_kafka_TopicDescription_is_internal(const rd_kafka_TopicDescription_t *topicdesc)
Gets if the topicdesc topic is internal.
RD_EXPORT rd_kafka_error_t * rd_kafka_init_transactions(rd_kafka_t *rk, int timeout_ms)
Initialize transactions for the producer instance.
RD_EXPORT const rd_kafka_group_result_t ** rd_kafka_ListConsumerGroupOffsets_result_groups(const rd_kafka_ListConsumerGroupOffsets_result_t *result, size_t *cntp)
Get an array of results from a ListConsumerGroupOffsets result.
RD_EXPORT int64_t rd_kafka_ListOffsetsResultInfo_timestamp(const rd_kafka_ListOffsetsResultInfo_t *result_info)
Returns the timestamp corresponding to the offset in result_info.
RD_EXPORT int32_t rd_kafka_msg_partitioner_consistent_random(const rd_kafka_topic_t *rkt, const void *key, size_t keylen, int32_t partition_cnt, void *rkt_opaque, void *msg_opaque)
Consistent-Random partitioner.
RD_EXPORT rd_kafka_error_t * rd_kafka_send_offsets_to_transaction(rd_kafka_t *rk, const rd_kafka_topic_partition_list_t *offsets, const rd_kafka_consumer_group_metadata_t *cgmetadata, int timeout_ms)
Sends a list of topic partition offsets to the consumer group coordinator for cgmetadata,...
RD_EXPORT rd_kafka_Uuid_t * rd_kafka_Uuid_new(int64_t most_significant_bits, int64_t least_significant_bits)
Creates a new UUID.
RD_EXPORT const char * rd_kafka_group_result_name(const rd_kafka_group_result_t *groupres)
RD_EXPORT rd_kafka_queue_t * rd_kafka_queue_get_main(rd_kafka_t *rk)
RD_EXPORT const char * rd_kafka_MemberDescription_group_instance_id(const rd_kafka_MemberDescription_t *member)
Gets group instance id of member.
RD_EXPORT void rd_kafka_consumer_group_metadata_destroy(rd_kafka_consumer_group_metadata_t *)
Frees the consumer group metadata object as returned by rd_kafka_consumer_group_metadata().
struct rd_kafka_ListConsumerGroupOffsets_s rd_kafka_ListConsumerGroupOffsets_t
Definition rdkafka.h:9132
RD_EXPORT void rd_kafka_event_destroy(rd_kafka_event_t *rkev)
Destroy an event.
struct rd_kafka_MemberDescription_s rd_kafka_MemberDescription_t
Member description included in ConsumerGroupDescription.
Definition rdkafka.h:8758
RD_EXPORT rd_kafka_msg_status_t rd_kafka_message_status(const rd_kafka_message_t *rkmessage)
Returns the message's persistence status in the topic log.
RD_EXPORT const char * rd_kafka_error_name(const rd_kafka_error_t *error)
rd_kafka_event_t rd_kafka_DescribeUserScramCredentials_result_t
Definition rdkafka.h:5890
RD_EXPORT void rd_kafka_AclBinding_destroy_array(rd_kafka_AclBinding_t **acl_bindings, size_t acl_bindings_cnt)
Helper function to destroy all AclBinding objects in the acl_bindings array (of acl_bindings_cnt elem...
RD_EXPORT void rd_kafka_conf_destroy(rd_kafka_conf_t *conf)
Destroys a conf object.
RD_EXPORT const rd_kafka_MemberAssignment_t * rd_kafka_MemberDescription_assignment(const rd_kafka_MemberDescription_t *member)
Gets assignment of member.
RD_EXPORT void * rd_kafka_mem_malloc(rd_kafka_t *rk, size_t size)
Allocate memory using the same allocator librdkafka uses.
RD_EXPORT void rd_kafka_queue_yield(rd_kafka_queue_t *rkqu)
Cancels the current rd_kafka_queue_poll() on rkqu.
RD_EXPORT void rd_kafka_UserScramCredentialAlteration_destroy(rd_kafka_UserScramCredentialAlteration_t *alteration)
Destroys a UserScramCredentialAlteration given its pointer.
RD_EXPORT void rd_kafka_topic_partition_list_sort(rd_kafka_topic_partition_list_t *rktparlist, int(*cmp)(const void *a, const void *b, void *cmp_opaque), void *cmp_opaque)
Sort list using comparator cmp.
RD_EXPORT rd_kafka_ResourceType_t rd_kafka_AclBinding_restype(const rd_kafka_AclBinding_t *acl)
RD_EXPORT rd_kafka_resp_err_t rd_kafka_interceptor_add_on_send(rd_kafka_t *rk, const char *ic_name, rd_kafka_interceptor_f_on_send_t *on_send, void *ic_opaque)
Append an on_send() interceptor.
RD_EXPORT const int rd_kafka_TopicPartitionInfo_partition(const rd_kafka_TopicPartitionInfo_t *partition)
Gets the partition id for partition.
RD_EXPORT rd_kafka_resp_err_t rd_kafka_commit_queue(rd_kafka_t *rk, const rd_kafka_topic_partition_list_t *offsets, rd_kafka_queue_t *rkqu, void(*cb)(rd_kafka_t *rk, rd_kafka_resp_err_t err, rd_kafka_topic_partition_list_t *offsets, void *commit_opaque), void *commit_opaque)
Commit offsets on broker for the provided list of partitions.
RD_EXPORT const rd_kafka_AclBinding_t ** rd_kafka_DescribeAcls_result_acls(const rd_kafka_DescribeAcls_result_t *result, size_t *cntp)
Get an array of resource results from a DescribeAcls result.
rd_kafka_timestamp_type_t
Definition rdkafka.h:220
@ RD_KAFKA_TIMESTAMP_NOT_AVAILABLE
Definition rdkafka.h:221
@ RD_KAFKA_TIMESTAMP_CREATE_TIME
Definition rdkafka.h:222
@ RD_KAFKA_TIMESTAMP_LOG_APPEND_TIME
Definition rdkafka.h:223
RD_EXPORT rd_kafka_resp_err_t rd_kafka_message_detach_headers(rd_kafka_message_t *rkmessage, rd_kafka_headers_t **hdrsp)
Get the message header list and detach the list from the message making the application the owner of ...
RD_EXPORT size_t rd_kafka_header_cnt(const rd_kafka_headers_t *hdrs)
Returns the number of header key/value pairs.
RD_EXPORT const char * rd_kafka_AclOperation_name(rd_kafka_AclOperation_t acl_operation)
RD_EXPORT void rd_kafka_DeleteGroups(rd_kafka_t *rk, rd_kafka_DeleteGroup_t **del_groups, size_t del_group_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Delete groups from cluster as specified by the del_groups array of size del_group_cnt elements.
rd_kafka_AclBinding_t rd_kafka_AclBindingFilter_t
ACL Binding filter is used to filter access control lists.
Definition rdkafka.h:9747
RD_EXPORT const char * rd_kafka_message_produce_errstr(const rd_kafka_message_t *rkmessage)
Returns the error string for an errored produced rd_kafka_message_t or NULL if there was no error.
RD_EXPORT void rd_kafka_DeleteConsumerGroupOffsets(rd_kafka_t *rk, rd_kafka_DeleteConsumerGroupOffsets_t **del_grpoffsets, size_t del_grpoffsets_cnt, const rd_kafka_AdminOptions_t *options, rd_kafka_queue_t *rkqu)
Delete committed offsets for a set of partitions in a consumer group. This will succeed at the partit...
RD_EXPORT rd_kafka_ResourceType_t rd_kafka_ConfigResource_type(const rd_kafka_ConfigResource_t *config)
RD_EXPORT rd_kafka_resp_err_t rd_kafka_ConfigResource_set_config(rd_kafka_ConfigResource_t *config, const char *name, const char *value)
Set configuration name value pair.
RD_EXPORT rd_kafka_topic_conf_t * rd_kafka_default_topic_conf_dup(rd_kafka_t *rk)
Creates a copy/duplicate of rk 's default topic configuration object.
rd_kafka_ScramMechanism_t
Apache Kafka ScramMechanism values.
Definition rdkafka.h:9502
RD_EXPORT const char * rd_kafka_name(const rd_kafka_t *rk)
Returns Kafka handle name.
rd_kafka_resp_err_t() rd_kafka_interceptor_f_on_conf_destroy_t(void *ic_opaque)
on_conf_destroy() is called from rd_kafka_*_conf_destroy() in the order the interceptors were added.
Definition rdkafka.h:6387
RD_EXPORT const rd_kafka_Node_t ** rd_kafka_DescribeCluster_result_nodes(const rd_kafka_DescribeCluster_result_t *result, size_t *cntp)
Gets the broker nodes for the result cluster.
RD_EXPORT int64_t rd_kafka_Uuid_least_significant_bits(const rd_kafka_Uuid_t *uuid)
Gets least significant 64 bits for the given UUID.
struct rd_kafka_MemberAssignment_s rd_kafka_MemberAssignment_t
Member assignment included in MemberDescription.
Definition rdkafka.h:8764
RD_EXPORT rd_kafka_ConfigSource_t rd_kafka_ConfigEntry_source(const rd_kafka_ConfigEntry_t *entry)
RD_EXPORT const rd_kafka_DescribeCluster_result_t * rd_kafka_event_DescribeCluster_result(rd_kafka_event_t *rkev)
Get DescribeCluster result.
Error code value, name and description. Typically for use with language bindings to automatically exp...
Definition rdkafka.h:671
const char * name
Definition rdkafka.h:673
rd_kafka_resp_err_t code
Definition rdkafka.h:672
const char * desc
Definition rdkafka.h:674
Group information.
Definition rdkafka.h:5221
struct rd_kafka_metadata_broker broker
Definition rdkafka.h:5222
int member_cnt
Definition rdkafka.h:5229
char * state
Definition rdkafka.h:5225
char * group
Definition rdkafka.h:5223
struct rd_kafka_group_member_info * members
Definition rdkafka.h:5228
char * protocol
Definition rdkafka.h:5227
char * protocol_type
Definition rdkafka.h:5226
rd_kafka_resp_err_t err
Definition rdkafka.h:5224
List of groups.
Definition rdkafka.h:5237
int group_cnt
Definition rdkafka.h:5239
struct rd_kafka_group_info * groups
Definition rdkafka.h:5238
Group member information.
Definition rdkafka.h:5179
char * member_id
Definition rdkafka.h:5180
int member_assignment_size
Definition rdkafka.h:5188
int member_metadata_size
Definition rdkafka.h:5185
void * member_metadata
Definition rdkafka.h:5183
char * client_host
Definition rdkafka.h:5182
void * member_assignment
Definition rdkafka.h:5186
char * client_id
Definition rdkafka.h:5181
A Kafka message as returned by the rd_kafka_consume*() family of functions as well as provided to the...
Definition rdkafka.h:1466
size_t key_len
Definition rdkafka.h:1479
size_t len
Definition rdkafka.h:1474
void * _private
Definition rdkafka.h:1489
void * key
Definition rdkafka.h:1477
int64_t offset
Definition rdkafka.h:1481
void * payload
Definition rdkafka.h:1470
int32_t partition
Definition rdkafka.h:1469
rd_kafka_topic_t * rkt
Definition rdkafka.h:1468
rd_kafka_resp_err_t err
Definition rdkafka.h:1467
Broker information.
Definition rdkafka.h:5034
int32_t id
Definition rdkafka.h:5035
int port
Definition rdkafka.h:5037
char * host
Definition rdkafka.h:5036
Partition information.
Definition rdkafka.h:5043
int32_t leader
Definition rdkafka.h:5046
int32_t * isrs
Definition rdkafka.h:5050
int replica_cnt
Definition rdkafka.h:5047
rd_kafka_resp_err_t err
Definition rdkafka.h:5045
int isr_cnt
Definition rdkafka.h:5049
int32_t id
Definition rdkafka.h:5044
int32_t * replicas
Definition rdkafka.h:5048
Metadata container.
Definition rdkafka.h:5067
int32_t orig_broker_id
Definition rdkafka.h:5074
char * orig_broker_name
Definition rdkafka.h:5075
int broker_cnt
Definition rdkafka.h:5068
struct rd_kafka_metadata_topic * topics
Definition rdkafka.h:5072
int topic_cnt
Definition rdkafka.h:5071
struct rd_kafka_metadata_broker * brokers
Definition rdkafka.h:5069
Topic information.
Definition rdkafka.h:5056
char * topic
Definition rdkafka.h:5057
struct rd_kafka_metadata_partition * partitions
Definition rdkafka.h:5059
rd_kafka_resp_err_t err
Definition rdkafka.h:5060
int partition_cnt
Definition rdkafka.h:5058
A growable list of Topic+Partitions.
Definition rdkafka.h:977
int cnt
Definition rdkafka.h:978
int size
Definition rdkafka.h:979
rd_kafka_topic_partition_t * elems
Definition rdkafka.h:980
Topic+Partition place holder.
Definition rdkafka.h:926
int64_t offset
Definition rdkafka.h:929
size_t metadata_size
Definition rdkafka.h:931
void * opaque
Definition rdkafka.h:932
char * topic
Definition rdkafka.h:927
void * metadata
Definition rdkafka.h:930
int32_t partition
Definition rdkafka.h:928
rd_kafka_resp_err_t err
Definition rdkafka.h:933
void * _private
Definition rdkafka.h:934
VTYPE + argument container for use with rd_kafka_produce_va()
Definition rdkafka.h:1170
rd_kafka_vtype_t vtype
Definition rdkafka.h:1171