29 #ifndef _RDKAFKACPP_H_
30 #define _RDKAFKACPP_H_
56 #include <sys/types.h>
63 #ifndef _SSIZE_T_DEFINED
64 #define _SSIZE_T_DEFINED
65 typedef SSIZE_T ssize_t;
69 #ifdef LIBRDKAFKA_STATICLIB
72 #ifdef LIBRDKAFKACPP_EXPORTS
73 #define RD_EXPORT __declspec(dllexport)
75 #define RD_EXPORT __declspec(dllimport)
87 struct rd_kafka_topic_s;
88 struct rd_kafka_message_s;
89 struct rd_kafka_conf_s;
90 struct rd_kafka_topic_conf_s;
114 #define RD_KAFKA_VERSION 0x010900ff
365 #define ERR_GROUP_LOAD_IN_PROGRESS ERR_COORDINATOR_LOAD_IN_PROGRESS
369 #define ERR_GROUP_COORDINATOR_NOT_AVAILABLE ERR_COORDINATOR_NOT_AVAILABLE
373 #define ERR_NOT_COORDINATOR_FOR_GROUP ERR_NOT_COORDINATOR
591 class TopicPartition;
635 virtual std::string
name()
const = 0;
640 virtual std::string
str()
const = 0;
748 const std::string &oauthbearer_config) = 0;
782 const std::string *key,
783 int32_t partition_cnt,
784 void *msg_opaque) = 0;
807 int32_t partition_cnt,
808 void *msg_opaque) = 0;
853 EVENT_SEVERITY_EMERG = 0,
854 EVENT_SEVERITY_ALERT = 1,
855 EVENT_SEVERITY_CRITICAL = 2,
856 EVENT_SEVERITY_ERROR = 3,
857 EVENT_SEVERITY_WARNING = 4,
858 EVENT_SEVERITY_NOTICE = 5,
859 EVENT_SEVERITY_INFO = 6,
860 EVENT_SEVERITY_DEBUG = 7
892 virtual std::string
fac()
const = 0;
902 virtual std::string
str()
const = 0;
1027 std::vector<TopicPartition *> &partitions) = 0;
1055 std::vector<TopicPartition *> &offsets) = 0;
1112 std::string &errstr) = 0;
1138 virtual int socket_cb(
int domain,
int type,
int protocol) = 0;
1162 virtual int open_cb(
const std::string &path,
int flags,
int mode) = 0;
1230 const std::string &value,
1231 std::string &errstr) = 0;
1236 std::string &errstr) = 0;
1240 const std::string &name,
1242 std::string &errstr) = 0;
1247 std::string &errstr) = 0;
1257 const Conf *topic_conf,
1258 std::string &errstr) = 0;
1263 std::string &errstr) = 0;
1268 std::string &errstr) = 0;
1273 std::string &errstr) = 0;
1278 std::string &errstr) = 0;
1283 std::string &errstr) = 0;
1288 std::string &errstr) = 0;
1296 std::string &errstr) = 0;
1334 std::string &errstr) = 0;
1349 std::string &value)
const = 0;
1404 virtual std::list<std::string> *
dump() = 0;
1409 std::string &errstr) = 0;
1461 std::string &errstr) = 0;
1487 std::string &errstr) = 0;
1508 virtual const std::string
name()
const = 0;
1545 virtual int poll(
int timeout_ms) = 0;
1571 const Topic *only_rkt,
1573 int timeout_ms) = 0;
1612 int timeout_ms) = 0;
1659 int timeout_ms) = 0;
1818 const std::string &token_value,
1819 int64_t md_lifetime_ms,
1820 const std::string &md_principal_name,
1821 const std::list<std::string> &extensions,
1822 std::string &errstr) = 0;
1842 const std::string &errstr) = 0;
1940 static void destroy(std::vector<TopicPartition *> &partitions);
1943 virtual const std::string &
topic()
const = 0;
1991 const std::string &topic_str,
1993 std::string &errstr);
1995 virtual ~
Topic() = 0;
1999 virtual const std::string
name()
const = 0;
2037 virtual struct rd_kafka_topic_s *
c_ptr() = 0;
2068 MSG_TIMESTAMP_LOG_APPEND_TIME
2109 Header(
const std::string &key,
const void *value,
size_t value_size) :
2110 key_(key), err_(
ERR_NO_ERROR), value_size_(value_size) {
2111 value_ = copy_value(value, value_size);
2131 key_(key), err_(err), value_(NULL), value_size_(value_size) {
2133 value_ = copy_value(value, value_size);
2142 key_(other.key_), err_(other.err_), value_size_(other.value_size_) {
2143 value_ = copy_value(other.value_, value_size_);
2152 if (&other ==
this) {
2158 value_size_ = other.value_size_;
2163 value_ = copy_value(other.value_, value_size_);
2186 return static_cast<const char *
>(value_);
2200 char *copy_value(
const void *value,
size_t value_size) {
2204 char *dest = (
char *)
mem_malloc(value_size + 1);
2205 memcpy(dest, (
const char *)value, value_size);
2206 dest[value_size] =
'\0';
2215 void *
operator new(size_t);
2246 size_t value_size) = 0;
2258 virtual ErrorCode add(
const std::string &key,
const std::string &value) = 0;
2289 virtual std::vector<Header>
get(
const std::string &key)
const = 0;
2336 MSG_STATUS_NOT_PERSISTED = 0,
2341 MSG_STATUS_POSSIBLY_PERSISTED = 1,
2346 MSG_STATUS_PERSISTED = 2,
2377 virtual size_t len()
const = 0;
2380 virtual const std::string *
key()
const = 0;
2419 virtual struct rd_kafka_message_s *
c_ptr() = 0;
2508 virtual int poll(
int timeout_ms) = 0;
2510 virtual ~
Queue() = 0;
2586 std::vector<RdKafka::TopicPartition *> &partitions) = 0;
2735 const std::vector<TopicPartition *> &offsets) = 0;
2773 int timeout_ms) = 0;
2914 const std::vector<TopicPartition *> &partitions) = 0;
2933 const std::vector<TopicPartition *> &partitions) = 0;
3054 int timeout_ms) = 0;
3212 MSG_FREE = RK_MSG_FREE,
3213 MSG_COPY = RK_MSG_COPY
3279 const std::string *key,
3280 void *msg_opaque) = 0;
3293 void *msg_opaque) = 0;
3309 void *msg_opaque) = 0;
3327 void *msg_opaque) = 0;
3336 const std::vector<char> *payload,
3337 const std::vector<char> *key,
3338 void *msg_opaque) = 0;
3394 PURGE_INFLIGHT = 0x2,
3401 PURGE_NON_BLOCKING = 0x4
3495 const std::vector<TopicPartition *> &offsets,
3497 int timeout_ms) = 0;
3580 virtual int32_t
id()
const = 0;
3583 virtual const std::string
host()
const = 0;
3610 virtual int32_t
id()
const = 0;
3619 virtual const std::vector<int32_t> *
replicas()
const = 0;
3624 virtual const std::vector<int32_t> *
isrs()
const = 0;
3642 virtual const std::string
topic()
const = 0;
Configuration interface.
Definition: rdkafkacpp.h:1188
virtual Conf::ConfResult get(OpenCb *&open_cb) const =0
Query single configuration value.
ConfType
Configuration object type.
Definition: rdkafkacpp.h:1193
@ CONF_GLOBAL
Definition: rdkafkacpp.h:1194
virtual Conf::ConfResult set(const std::string &name, EventCb *event_cb, std::string &errstr)=0
Use with name = "event_cb".
virtual struct rd_kafka_topic_conf_s * c_ptr_topic()=0
Returns the underlying librdkafka C rd_kafka_topic_conf_t handle.
virtual Conf::ConfResult set(const std::string &name, OffsetCommitCb *offset_commit_cb, std::string &errstr)=0
Use with name = "offset_commit_cb".
ConfResult
RdKafka::Conf::Set() result code.
Definition: rdkafkacpp.h:1201
virtual Conf::ConfResult set(const std::string &name, PartitionerKeyPointerCb *partitioner_kp_cb, std::string &errstr)=0
Use with name = "partitioner_key_pointer_cb".
virtual Conf::ConfResult get(EventCb *&event_cb) const =0
Query single configuration value.
virtual Conf::ConfResult set_engine_callback_data(void *value, std::string &errstr)=0
Set callback_data for ssl engine.
static Conf * create(ConfType type)
Create configuration object.
virtual Conf::ConfResult set(const std::string &name, DeliveryReportCb *dr_cb, std::string &errstr)=0
Use with name = "dr_cb".
virtual Conf::ConfResult get(SocketCb *&socket_cb) const =0
Query single configuration value.
virtual Conf::ConfResult set(const std::string &name, const Conf *topic_conf, std::string &errstr)=0
Use with name = "default_topic_conf".
virtual Conf::ConfResult get(OAuthBearerTokenRefreshCb *&oauthbearer_token_refresh_cb) const =0
Query single configuration value.
virtual Conf::ConfResult set(const std::string &name, RebalanceCb *rebalance_cb, std::string &errstr)=0
Use with name = "rebalance_cb".
virtual Conf::ConfResult set(const std::string &name, ConsumeCb *consume_cb, std::string &errstr)=0
Use with name = "consume_cb".
virtual Conf::ConfResult get(RebalanceCb *&rebalance_cb) const =0
Query single configuration value.
virtual Conf::ConfResult get(PartitionerCb *&partitioner_cb) const =0
Query single configuration value.
virtual Conf::ConfResult get(OffsetCommitCb *&offset_commit_cb) const =0
Query single configuration value.
virtual Conf::ConfResult get(PartitionerKeyPointerCb *&partitioner_kp_cb) const =0
Query single configuration value.
virtual std::list< std::string > * dump()=0
Dump configuration names and values to list containing name,value tuples.
virtual Conf::ConfResult set_ssl_cert(RdKafka::CertificateType cert_type, RdKafka::CertificateEncoding cert_enc, const void *buffer, size_t size, std::string &errstr)=0
Set certificate/key cert_type from the cert_enc encoded memory at buffer of size bytes.
virtual Conf::ConfResult set(const std::string &name, SocketCb *socket_cb, std::string &errstr)=0
Use with name = "socket_cb".
virtual Conf::ConfResult set(const std::string &name, OpenCb *open_cb, std::string &errstr)=0
Use with name = "open_cb".
virtual Conf::ConfResult enable_sasl_queue(bool enable, std::string &errstr)=0
Enable/disable creation of a queue specific to SASL events and callbacks.
virtual Conf::ConfResult set(const std::string &name, SslCertificateVerifyCb *ssl_cert_verify_cb, std::string &errstr)=0
Use with name = "ssl_cert_verify_cb".
virtual Conf::ConfResult set(const std::string &name, OAuthBearerTokenRefreshCb *oauthbearer_token_refresh_cb, std::string &errstr)=0
Use with name = "oauthbearer_token_refresh_cb".
virtual Conf::ConfResult set(const std::string &name, const std::string &value, std::string &errstr)=0
Set configuration property name to value value.
virtual Conf::ConfResult get(SslCertificateVerifyCb *&ssl_cert_verify_cb) const =0
Use with name = "ssl_cert_verify_cb".
virtual struct rd_kafka_conf_s * c_ptr_global()=0
Returns the underlying librdkafka C rd_kafka_conf_t handle.
virtual Conf::ConfResult set(const std::string &name, PartitionerCb *partitioner_cb, std::string &errstr)=0
Use with name = "partitioner_cb".
virtual Conf::ConfResult get(DeliveryReportCb *&dr_cb) const =0
Query single configuration value.
virtual Conf::ConfResult get(const std::string &name, std::string &value) const =0
Query single configuration value.
Consume callback class.
Definition: rdkafkacpp.h:936
virtual void consume_cb(Message &message, void *opaque)=0
The consume callback is used with RdKafka::Consumer::consume_callback() methods and will be called fo...
Simple Consumer (legacy)
Definition: rdkafkacpp.h:2977
virtual ErrorCode start(Topic *topic, int32_t partition, int64_t offset, Queue *queue)=0
Start consuming messages for topic and partition on queue queue.
virtual Message * consume(Topic *topic, int32_t partition, int timeout_ms)=0
Consume a single message from topic and partition.
virtual ErrorCode stop(Topic *topic, int32_t partition)=0
Stop consuming messages for topic and partition, purging all messages currently in the local queue.
virtual int consume_callback(Topic *topic, int32_t partition, int timeout_ms, ConsumeCb *consume_cb, void *opaque)=0
Consumes messages from topic and partition, calling the provided callback for each consumed messsage.
virtual Message * consume(Queue *queue, int timeout_ms)=0
Consume a single message from the specified queue.
static int64_t OffsetTail(int64_t offset)
Converts an offset into the logical offset from the tail of a topic.
virtual int consume_callback(Queue *queue, int timeout_ms, RdKafka::ConsumeCb *consume_cb, void *opaque)=0
Consumes messages from queue, calling the provided callback for each consumed messsage.
static Consumer * create(const Conf *conf, std::string &errstr)
Creates a new Kafka consumer handle.
virtual ErrorCode seek(Topic *topic, int32_t partition, int64_t offset, int timeout_ms)=0
Seek consumer for topic+partition to offset which is either an absolute or logical offset.
virtual ErrorCode start(Topic *topic, int32_t partition, int64_t offset)=0
Start consuming messages for topic and partition at offset offset which may either be a proper offset...
Delivery Report callback class.
Definition: rdkafkacpp.h:698
virtual void dr_cb(Message &message)=0
Delivery report callback.
The Error class is used as a return value from APIs to propagate an error. The error consists of an e...
Definition: rdkafkacpp.h:613
virtual ErrorCode code() const =0
virtual std::string name() const =0
virtual std::string str() const =0
virtual bool is_fatal() const =0
static Error * create(ErrorCode code, const std::string *errstr)
Create error object.
virtual bool is_retriable() const =0
virtual bool txn_requires_abort() const =0
Event callback class.
Definition: rdkafkacpp.h:824
virtual void event_cb(Event &event)=0
Event callback.
Event object class as passed to the EventCb callback.
Definition: rdkafkacpp.h:841
virtual int throttle_time() const =0
virtual std::string broker_name() const =0
virtual bool fatal() const =0
virtual std::string fac() const =0
virtual ErrorCode err() const =0
virtual Type type() const =0
virtual std::string str() const =0
Severity
EVENT_LOG severities (conforms to syslog(3) severities)
Definition: rdkafkacpp.h:852
Type
Event type.
Definition: rdkafkacpp.h:844
@ EVENT_ERROR
Definition: rdkafkacpp.h:845
@ EVENT_STATS
Definition: rdkafkacpp.h:846
@ EVENT_LOG
Definition: rdkafkacpp.h:847
virtual Severity severity() const =0
virtual int broker_id() const =0
Base handle, super class for specific clients.
Definition: rdkafkacpp.h:1502
virtual Queue * get_partition_queue(const TopicPartition *partition)=0
Retrieve queue for a given partition.
virtual ErrorCode offsetsForTimes(std::vector< TopicPartition * > &offsets, int timeout_ms)=0
Look up the offsets for the given partitions by timestamp.
virtual ErrorCode oauthbearer_set_token(const std::string &token_value, int64_t md_lifetime_ms, const std::string &md_principal_name, const std::list< std::string > &extensions, std::string &errstr)=0
Set SASL/OAUTHBEARER token and metadata.
virtual ErrorCode resume(std::vector< TopicPartition * > &partitions)=0
Resume producing or consumption for the provided list of partitions.
virtual Error * sasl_background_callbacks_enable()=0
Enable SASL OAUTHBEARER refresh callbacks on the librdkafka background thread.
virtual void mem_free(void *ptr)=0
Free pointer returned by librdkafka.
virtual const std::string clusterid(int timeout_ms)=0
Returns the ClusterId as reported in broker metadata.
virtual int outq_len()=0
Returns the current out queue length.
virtual void * mem_malloc(size_t size)=0
Allocate memory using the same allocator librdkafka uses.
virtual void yield()=0
Cancels the current callback dispatcher (Handle::poll(), KafkaConsumer::consume(),...
virtual ErrorCode pause(std::vector< TopicPartition * > &partitions)=0
Pause producing or consumption for the provided list of partitions.
virtual int32_t controllerid(int timeout_ms)=0
Returns the current ControllerId (controller broker id) as reported in broker metadata.
virtual struct rd_kafka_s * c_ptr()=0
Returns the underlying librdkafka C rd_kafka_t handle.
virtual Queue * get_sasl_queue()=0
virtual int poll(int timeout_ms)=0
Polls the provided kafka handle for events.
virtual ErrorCode oauthbearer_set_token_failure(const std::string &errstr)=0
SASL/OAUTHBEARER token refresh failure indicator.
virtual ErrorCode query_watermark_offsets(const std::string &topic, int32_t partition, int64_t *low, int64_t *high, int timeout_ms)=0
Query broker for low (oldest/beginning) and high (newest/end) offsets for partition.
virtual ErrorCode get_watermark_offsets(const std::string &topic, int32_t partition, int64_t *low, int64_t *high)=0
Get last known low (oldest/beginning) and high (newest/end) offsets for partition.
virtual const std::string name() const =0
virtual ErrorCode fatal_error(std::string &errstr) const =0
Returns the first fatal error set on this client instance, or ERR_NO_ERROR if no fatal error has occu...
virtual ErrorCode metadata(bool all_topics, const Topic *only_rkt, Metadata **metadatap, int timeout_ms)=0
Request Metadata from broker.
virtual ErrorCode set_log_queue(Queue *queue)=0
Forward librdkafka logs (and debug) to the specified queue for serving with one of the ....
virtual const std::string memberid() const =0
Returns the client's broker-assigned group member id.
virtual Queue * get_background_queue()=0
High-level KafkaConsumer (for brokers 0.9 and later)
Definition: rdkafkacpp.h:2565
virtual ErrorCode commitSync()=0
Commit offsets for the current assignment.
static KafkaConsumer * create(const Conf *conf, std::string &errstr)
Creates a KafkaConsumer.
virtual ErrorCode unassign()=0
Stop consumption and remove the current assignment.
virtual ErrorCode commitAsync(Message *message)=0
Commit offset for a single topic+partition based on message.
virtual Error * incremental_assign(const std::vector< TopicPartition * > &partitions)=0
Incrementally add partitions to the current assignment.
virtual ErrorCode subscription(std::vector< std::string > &topics)=0
Returns the current subscription as set by RdKafka::KafkaConsumer::subscribe()
virtual ErrorCode position(std::vector< TopicPartition * > &partitions)=0
Retrieve current positions (offsets) for topics+partitions.
virtual std::string rebalance_protocol()=0
The rebalance protocol currently in use. This will be "NONE" if the consumer has not (yet) joined a g...
virtual ErrorCode close()=0
Close and shut down the consumer.
virtual ErrorCode commitAsync(const std::vector< TopicPartition * > &offsets)=0
Commit offset for the provided list of partitions.
virtual ErrorCode commitSync(OffsetCommitCb *offset_commit_cb)=0
Commit offsets for the current assignment.
virtual bool assignment_lost()=0
Check whether the consumer considers the current assignment to have been lost involuntarily....
virtual ErrorCode commitAsync()=0
Asynchronous version of RdKafka::KafkaConsumer::CommitSync()
virtual ErrorCode subscribe(const std::vector< std::string > &topics)=0
Update the subscription set to topics.
virtual ConsumerGroupMetadata * groupMetadata()=0
virtual Error * close(Queue *queue)=0
Close and shut down the consumer.
virtual Message * consume(int timeout_ms)=0
Consume message or get error event, triggers callbacks.
virtual ErrorCode commitSync(std::vector< TopicPartition * > &offsets)=0
Commit offsets for the provided list of partitions.
virtual ErrorCode unsubscribe()=0
Unsubscribe from the current subscription set.
virtual ErrorCode commitSync(std::vector< TopicPartition * > &offsets, OffsetCommitCb *offset_commit_cb)=0
Commit offsets for the provided list of partitions.
virtual ErrorCode committed(std::vector< TopicPartition * > &partitions, int timeout_ms)=0
Retrieve committed offsets for topics+partitions.
virtual ErrorCode seek(const TopicPartition &partition, int timeout_ms)=0
Seek consumer for topic+partition to offset which is either an absolute or logical offset.
virtual ErrorCode offsets_store(std::vector< TopicPartition * > &offsets)=0
Store offset offset for topic partition partition. The offset will be committed (written) to the offs...
virtual ErrorCode commitSync(Message *message)=0
Commit offset for a single topic+partition based on message.
virtual Error * incremental_unassign(const std::vector< TopicPartition * > &partitions)=0
Incrementally remove partitions from the current assignment.
virtual ErrorCode assign(const std::vector< TopicPartition * > &partitions)=0
Update the assignment set to partitions.
virtual ErrorCode assignment(std::vector< RdKafka::TopicPartition * > &partitions)=0
Returns the current partition assignment as set by RdKafka::KafkaConsumer::assign()
Message timestamp object.
Definition: rdkafkacpp.h:2062
int64_t timestamp
Definition: rdkafkacpp.h:2072
MessageTimestampType
Definition: rdkafkacpp.h:2065
@ MSG_TIMESTAMP_CREATE_TIME
Definition: rdkafkacpp.h:2067
@ MSG_TIMESTAMP_NOT_AVAILABLE
Definition: rdkafkacpp.h:2066
MessageTimestampType type
Definition: rdkafkacpp.h:2071
Message object.
Definition: rdkafkacpp.h:2328
virtual std::string topic_name() const =0
virtual Status status() const =0
Returns the message's persistence status in the topic log.
virtual const std::string * key() const =0
virtual std::string errstr() const =0
Accessor functions*.
virtual int32_t broker_id() const =0
virtual const void * key_pointer() const =0
virtual RdKafka::Headers * headers()=0
virtual int64_t offset() const =0
virtual size_t len() const =0
virtual struct rd_kafka_message_s * c_ptr()=0
Returns the underlying librdkafka C rd_kafka_message_t handle.
virtual Topic * topic() const =0
virtual MessageTimestamp timestamp() const =0
virtual int64_t latency() const =0
virtual ErrorCode err() const =0
Status
Message persistence status can be used by the application to find out if a produced message was persi...
Definition: rdkafkacpp.h:2332
virtual size_t key_len() const =0
virtual RdKafka::Headers * headers(RdKafka::ErrorCode *err)=0
virtual void * msg_opaque() const =0
virtual void * payload() const =0
virtual int32_t partition() const =0
SASL/OAUTHBEARER token refresh callback class.
Definition: rdkafkacpp.h:737
virtual void oauthbearer_token_refresh_cb(RdKafka::Handle *handle, const std::string &oauthbearer_config)=0
SASL/OAUTHBEARER token refresh callback class.
Offset Commit callback class.
Definition: rdkafkacpp.h:1037
virtual void offset_commit_cb(RdKafka::ErrorCode err, std::vector< TopicPartition * > &offsets)=0
Set offset commit callback for use with consumer groups.
Portability: OpenCb callback class
Definition: rdkafkacpp.h:1149
virtual int open_cb(const std::string &path, int flags, int mode)=0
Open callback The open callback is responsible for opening the file specified by pathname,...
Partitioner callback class.
Definition: rdkafkacpp.h:762
virtual int32_t partitioner_cb(const Topic *topic, const std::string *key, int32_t partition_cnt, void *msg_opaque)=0
Partitioner callback.
Variant partitioner with key pointer.
Definition: rdkafkacpp.h:794
virtual int32_t partitioner_cb(const Topic *topic, const void *key, size_t key_len, int32_t partition_cnt, void *msg_opaque)=0
Variant partitioner callback that gets key as pointer and length instead of as a const std::string *.
Producer.
Definition: rdkafkacpp.h:3159
virtual Error * abort_transaction(int timeout_ms)=0
Aborts the ongoing transaction.
virtual Error * init_transactions(int timeout_ms)=0
Initialize transactions for the producer instance.
virtual ErrorCode produce(const std::string topic_name, int32_t partition, int msgflags, void *payload, size_t len, const void *key, size_t key_len, int64_t timestamp, void *msg_opaque)=0
produce() variant that takes topic as a string (no need for creating a Topic object),...
virtual Error * commit_transaction(int timeout_ms)=0
Commit the current transaction as started with begin_transaction().
virtual ErrorCode produce(Topic *topic, int32_t partition, int msgflags, void *payload, size_t len, const void *key, size_t key_len, void *msg_opaque)=0
Variant produce() that passes the key as a pointer and length instead of as a const std::string *.
virtual Error * send_offsets_to_transaction(const std::vector< TopicPartition * > &offsets, const ConsumerGroupMetadata *group_metadata, int timeout_ms)=0
Sends a list of topic partition offsets to the consumer group coordinator for group_metadata,...
virtual ErrorCode purge(int purge_flags)=0
Purge messages currently handled by the producer instance.
virtual ErrorCode produce(const std::string topic_name, int32_t partition, int msgflags, void *payload, size_t len, const void *key, size_t key_len, int64_t timestamp, RdKafka::Headers *headers, void *msg_opaque)=0
produce() variant that that allows for Header support on produce Otherwise identical to produce() abo...
virtual ErrorCode produce(Topic *topic, int32_t partition, int msgflags, void *payload, size_t len, const std::string *key, void *msg_opaque)=0
Produce and send a single message to broker.
virtual ErrorCode produce(Topic *topic, int32_t partition, const std::vector< char > *payload, const std::vector< char > *key, void *msg_opaque)=0
Variant produce() that accepts vectors for key and payload. The vector data will be copied.
virtual Error * begin_transaction()=0
init_transactions() must have been called successfully (once) before this function is called.
static Producer * create(const Conf *conf, std::string &errstr)
Creates a new Kafka producer handle.
virtual ErrorCode flush(int timeout_ms)=0
Wait until all outstanding produce requests, et.al, are completed. This should typically be done prio...
Queue interface.
Definition: rdkafkacpp.h:2468
static Queue * create(Handle *handle)
Create Queue object.
virtual ErrorCode forward(Queue *dst)=0
Forward/re-route queue to dst. If dst is NULL, the forwarding is removed.
virtual int poll(int timeout_ms)=0
Poll queue, serving any enqueued callbacks.
virtual void io_event_enable(int fd, const void *payload, size_t size)=0
Enable IO event triggering for queue.
virtual Message * consume(int timeout_ms)=0
Consume message or get error event from the queue.
KafkaConsumer: Rebalance callback class
Definition: rdkafkacpp.h:955
virtual void rebalance_cb(RdKafka::KafkaConsumer *consumer, RdKafka::ErrorCode err, std::vector< TopicPartition * > &partitions)=0
Group rebalance callback for use with RdKafka::KafkaConsumer.
Portability: SocketCb callback class
Definition: rdkafkacpp.h:1123
virtual int socket_cb(int domain, int type, int protocol)=0
Socket callback.
SSL broker certificate verification class.
Definition: rdkafkacpp.h:1068
virtual bool ssl_cert_verify_cb(const std::string &broker_name, int32_t broker_id, int *x509_error, int depth, const char *buf, size_t size, std::string &errstr)=0
SSL broker certificate verification callback.
Topic+Partition.
Definition: rdkafkacpp.h:1915
static void destroy(std::vector< TopicPartition * > &partitions)
Destroy/delete the TopicPartitions in partitions and clear the vector.
virtual const std::string & topic() const =0
static TopicPartition * create(const std::string &topic, int partition, int64_t offset)
Create topic+partition object for topic and partition with offset offset.
virtual void set_offset(int64_t offset)=0
Set offset.
virtual int partition() const =0
static TopicPartition * create(const std::string &topic, int partition)
Create topic+partition object for topic and partition.
virtual int64_t offset() const =0
virtual ErrorCode err() const =0
Topic handle.
Definition: rdkafkacpp.h:1964
static const int64_t OFFSET_INVALID
Definition: rdkafkacpp.h:1978
virtual ErrorCode offset_store(int32_t partition, int64_t offset)=0
Store offset offset + 1 for topic partition partition. The offset will be committed (written) to the ...
static const int64_t OFFSET_BEGINNING
Special offsets.
Definition: rdkafkacpp.h:1975
static Topic * create(Handle *base, const std::string &topic_str, const Conf *conf, std::string &errstr)
Creates a new topic handle for topic named topic_str.
virtual bool partition_available(int32_t partition) const =0
static const int64_t OFFSET_STORED
Definition: rdkafkacpp.h:1977
virtual struct rd_kafka_topic_s * c_ptr()=0
Returns the underlying librdkafka C rd_kafka_topic_t handle.
static const int32_t PARTITION_UA
Unassigned partition.
Definition: rdkafkacpp.h:1972
static const int64_t OFFSET_END
Definition: rdkafkacpp.h:1976
virtual const std::string name() const =0
RD_EXPORT std::string get_debug_contexts()
Returns a CSV list of the supported debug contexts for use with Conf::Set("debug",...
RD_EXPORT std::string version_str()
Returns the librdkafka version as string.
CertificateEncoding
SSL certificate encoding.
Definition: rdkafkacpp.h:571
@ CERT_ENC_PKCS12
Definition: rdkafkacpp.h:572
@ CERT_ENC_DER
Definition: rdkafkacpp.h:573
@ CERT_ENC_PEM
Definition: rdkafkacpp.h:574
RD_EXPORT int wait_destroyed(int timeout_ms)
Wait for all rd_kafka_t objects to be destroyed.
RD_EXPORT std::string err2str(RdKafka::ErrorCode err)
Returns a human readable representation of a kafka error.
ErrorCode
Error codes.
Definition: rdkafkacpp.h:200
@ ERR__APPLICATION
Definition: rdkafkacpp.h:320
@ ERR_PRINCIPAL_DESERIALIZATION_FAILURE
Definition: rdkafkacpp.h:544
@ ERR_NOT_LEADER_FOR_PARTITION
Definition: rdkafkacpp.h:347
@ ERR_NON_EMPTY_GROUP
Definition: rdkafkacpp.h:482
@ ERR__DESTROY
Definition: rdkafkacpp.h:209
@ ERR_INVALID_REQUEST
Definition: rdkafkacpp.h:425
@ ERR_GROUP_ID_NOT_FOUND
Definition: rdkafkacpp.h:484
@ ERR_INVALID_SESSION_TIMEOUT
Definition: rdkafkacpp.h:393
@ ERR__UNDERFLOW
Definition: rdkafkacpp.h:296
@ ERR__ASSIGN_PARTITIONS
Definition: rdkafkacpp.h:256
@ ERR__UNKNOWN_BROKER
Definition: rdkafkacpp.h:314
@ ERR_CLUSTER_AUTHORIZATION_FAILED
Definition: rdkafkacpp.h:403
@ ERR_UNKNOWN
Definition: rdkafkacpp.h:333
@ ERR_ELIGIBLE_LEADERS_NOT_AVAILABLE
Definition: rdkafkacpp.h:513
@ ERR__REVOKE_PARTITIONS
Definition: rdkafkacpp.h:258
@ ERR__FATAL
Definition: rdkafkacpp.h:306
@ ERR_SASL_AUTHENTICATION_FAILED
Definition: rdkafkacpp.h:462
@ ERR_FEATURE_UPDATE_FAILED
Definition: rdkafkacpp.h:542
@ ERR__CRIT_SYS_RESOURCE
Definition: rdkafkacpp.h:215
@ ERR__ASSIGNMENT_LOST
Definition: rdkafkacpp.h:322
@ ERR_STALE_BROKER_EPOCH
Definition: rdkafkacpp.h:500
@ ERR_INVALID_TXN_STATE
Definition: rdkafkacpp.h:437
@ ERR__UNKNOWN_PARTITION
Definition: rdkafkacpp.h:226
@ ERR__WAIT_CACHE
Definition: rdkafkacpp.h:278
@ ERR_UNSTABLE_OFFSET_COMMIT
Definition: rdkafkacpp.h:524
@ ERR__NODE_UPDATE
Definition: rdkafkacpp.h:242
@ ERR_NO_ERROR
Definition: rdkafkacpp.h:335
@ ERR_INCONSISTENT_GROUP_PROTOCOL
Definition: rdkafkacpp.h:387
@ ERR__UNKNOWN_GROUP
Definition: rdkafkacpp.h:248
@ ERR_INVALID_PARTITIONS
Definition: rdkafkacpp.h:415
@ ERR_DELEGATION_TOKEN_NOT_FOUND
Definition: rdkafkacpp.h:470
@ ERR__NOOP
Definition: rdkafkacpp.h:324
@ ERR_NO_REASSIGNMENT_IN_PROGRESS
Definition: rdkafkacpp.h:517
@ ERR__NOT_IMPLEMENTED
Definition: rdkafkacpp.h:266
@ ERR__MSG_TIMED_OUT
Definition: rdkafkacpp.h:219
@ ERR_INVALID_REPLICATION_FACTOR
Definition: rdkafkacpp.h:417
@ ERR__READ_ONLY
Definition: rdkafkacpp.h:292
@ ERR_UNKNOWN_MEMBER_ID
Definition: rdkafkacpp.h:391
@ ERR_OUT_OF_ORDER_SEQUENCE_NUMBER
Definition: rdkafkacpp.h:431
@ ERR_INVALID_GROUP_ID
Definition: rdkafkacpp.h:389
@ ERR__RETRY
Definition: rdkafkacpp.h:300
@ ERR__TIMED_OUT_QUEUE
Definition: rdkafkacpp.h:274
@ ERR__NOT_CONFIGURED
Definition: rdkafkacpp.h:316
@ ERR_DUPLICATE_RESOURCE
Definition: rdkafkacpp.h:533
@ ERR_LEADER_NOT_AVAILABLE
Definition: rdkafkacpp.h:345
@ ERR_INVALID_FETCH_SESSION_EPOCH
Definition: rdkafkacpp.h:488
@ ERR_KAFKA_STORAGE_ERROR
Definition: rdkafkacpp.h:458
@ ERR_ILLEGAL_GENERATION
Definition: rdkafkacpp.h:385
@ ERR_TOPIC_ALREADY_EXISTS
Definition: rdkafkacpp.h:413
@ ERR__AUTHENTICATION
Definition: rdkafkacpp.h:268
@ ERR_TOPIC_EXCEPTION
Definition: rdkafkacpp.h:375
@ ERR__RESOLVE
Definition: rdkafkacpp.h:217
@ ERR_INVALID_REQUIRED_ACKS
Definition: rdkafkacpp.h:383
@ ERR_FETCH_SESSION_ID_NOT_FOUND
Definition: rdkafkacpp.h:486
@ ERR__SSL
Definition: rdkafkacpp.h:244
@ ERR_COORDINATOR_NOT_AVAILABLE
Definition: rdkafkacpp.h:367
@ ERR__CONFLICT
Definition: rdkafkacpp.h:260
@ ERR_GROUP_SUBSCRIBED_TO_TOPIC
Definition: rdkafkacpp.h:520
@ ERR__STATE
Definition: rdkafkacpp.h:262
@ ERR__KEY_SERIALIZATION
Definition: rdkafkacpp.h:282
@ ERR_INVALID_MSG_SIZE
Definition: rdkafkacpp.h:343
@ ERR_DELEGATION_TOKEN_OWNER_MISMATCH
Definition: rdkafkacpp.h:472
@ ERR_REASSIGNMENT_IN_PROGRESS
Definition: rdkafkacpp.h:466
@ ERR__OUTDATED
Definition: rdkafkacpp.h:272
@ ERR_OFFSET_NOT_AVAILABLE
Definition: rdkafkacpp.h:502
@ ERR_THROTTLING_QUOTA_EXCEEDED
Definition: rdkafkacpp.h:526
@ ERR_REQUEST_TIMED_OUT
Definition: rdkafkacpp.h:349
@ ERR_INVALID_TRANSACTION_TIMEOUT
Definition: rdkafkacpp.h:443
@ ERR_CONCURRENT_TRANSACTIONS
Definition: rdkafkacpp.h:446
@ ERR_STALE_CTRL_EPOCH
Definition: rdkafkacpp.h:357
@ ERR_UNSUPPORTED_VERSION
Definition: rdkafkacpp.h:411
@ ERR__WAIT_COORD
Definition: rdkafkacpp.h:246
@ ERR_ELECTION_NOT_NEEDED
Definition: rdkafkacpp.h:515
@ ERR_INVALID_COMMIT_OFFSET_SIZE
Definition: rdkafkacpp.h:397
@ ERR_DELEGATION_TOKEN_EXPIRED
Definition: rdkafkacpp.h:478
@ ERR_TOPIC_AUTHORIZATION_FAILED
Definition: rdkafkacpp.h:399
@ ERR_MSG_SIZE_TOO_LARGE
Definition: rdkafkacpp.h:355
@ ERR_NETWORK_EXCEPTION
Definition: rdkafkacpp.h:361
@ ERR_INVALID_PRINCIPAL_TYPE
Definition: rdkafkacpp.h:480
@ ERR__MAX_POLL_EXCEEDED
Definition: rdkafkacpp.h:312
@ ERR_FENCED_LEADER_EPOCH
Definition: rdkafkacpp.h:494
@ ERR__BEGIN
Definition: rdkafkacpp.h:203
@ ERR_DELEGATION_TOKEN_AUTHORIZATION_FAILED
Definition: rdkafkacpp.h:476
@ ERR_UNACCEPTABLE_CREDENTIAL
Definition: rdkafkacpp.h:535
@ ERR_RESOURCE_NOT_FOUND
Definition: rdkafkacpp.h:531
@ ERR_GROUP_AUTHORIZATION_FAILED
Definition: rdkafkacpp.h:401
@ ERR_LOG_DIR_NOT_FOUND
Definition: rdkafkacpp.h:460
@ ERR__AUTO_OFFSET_RESET
Definition: rdkafkacpp.h:326
@ ERR__NO_OFFSET
Definition: rdkafkacpp.h:270
@ ERR__PURGE_INFLIGHT
Definition: rdkafkacpp.h:304
@ ERR_INVALID_RECORD
Definition: rdkafkacpp.h:522
@ ERR_FENCED_INSTANCE_ID
Definition: rdkafkacpp.h:511
@ ERR_ILLEGAL_SASL_STATE
Definition: rdkafkacpp.h:409
@ ERR__FAIL
Definition: rdkafkacpp.h:211
@ ERR_MEMBER_ID_REQUIRED
Definition: rdkafkacpp.h:504
@ ERR__EXISTING_SUBSCRIPTION
Definition: rdkafkacpp.h:254
@ ERR__PREV_IN_PROGRESS
Definition: rdkafkacpp.h:252
@ ERR_PRODUCER_FENCED
Definition: rdkafkacpp.h:529
@ ERR__TIMED_OUT
Definition: rdkafkacpp.h:236
@ ERR_PREFERRED_LEADER_NOT_AVAILABLE
Definition: rdkafkacpp.h:506
@ ERR__PARTITION_EOF
Definition: rdkafkacpp.h:224
@ ERR_NOT_ENOUGH_REPLICAS_AFTER_APPEND
Definition: rdkafkacpp.h:381
@ ERR__QUEUE_FULL
Definition: rdkafkacpp.h:238
@ ERR_INVALID_MSG
Definition: rdkafkacpp.h:339
@ ERR_TOPIC_DELETION_DISABLED
Definition: rdkafkacpp.h:492
@ ERR__VALUE_DESERIALIZATION
Definition: rdkafkacpp.h:288
@ ERR_DELEGATION_TOKEN_REQUEST_NOT_ALLOWED
Definition: rdkafkacpp.h:474
@ ERR__GAPLESS_GUARANTEE
Definition: rdkafkacpp.h:310
@ ERR_NOT_COORDINATOR
Definition: rdkafkacpp.h:371
@ ERR__UNSUPPORTED_FEATURE
Definition: rdkafkacpp.h:276
@ ERR_INVALID_PRODUCER_EPOCH
Definition: rdkafkacpp.h:435
@ ERR_DUPLICATE_SEQUENCE_NUMBER
Definition: rdkafkacpp.h:433
@ ERR_OFFSET_METADATA_TOO_LARGE
Definition: rdkafkacpp.h:359
@ ERR_NOT_ENOUGH_REPLICAS
Definition: rdkafkacpp.h:379
@ ERR_UNSUPPORTED_SASL_MECHANISM
Definition: rdkafkacpp.h:407
@ ERR__INCONSISTENT
Definition: rdkafkacpp.h:308
@ ERR_UNKNOWN_PRODUCER_ID
Definition: rdkafkacpp.h:464
@ ERR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED
Definition: rdkafkacpp.h:452
@ ERR_UNSUPPORTED_COMPRESSION_TYPE
Definition: rdkafkacpp.h:498
@ ERR_UNKNOWN_LEADER_EPOCH
Definition: rdkafkacpp.h:496
@ ERR_LISTENER_NOT_FOUND
Definition: rdkafkacpp.h:490
@ ERR__PURGE_QUEUE
Definition: rdkafkacpp.h:302
@ ERR_SECURITY_DISABLED
Definition: rdkafkacpp.h:454
@ ERR__ISR_INSUFF
Definition: rdkafkacpp.h:240
@ ERR__FENCED
Definition: rdkafkacpp.h:318
@ ERR_INCONSISTENT_VOTER_SET
Definition: rdkafkacpp.h:538
@ ERR__BAD_COMPRESSION
Definition: rdkafkacpp.h:207
@ ERR__ALL_BROKERS_DOWN
Definition: rdkafkacpp.h:232
@ ERR__VALUE_SERIALIZATION
Definition: rdkafkacpp.h:284
@ ERR_OPERATION_NOT_ATTEMPTED
Definition: rdkafkacpp.h:456
@ ERR__END
Definition: rdkafkacpp.h:329
@ ERR__IN_PROGRESS
Definition: rdkafkacpp.h:250
@ ERR_INVALID_CONFIG
Definition: rdkafkacpp.h:421
@ ERR_REPLICA_NOT_AVAILABLE
Definition: rdkafkacpp.h:353
@ ERR_INVALID_REPLICA_ASSIGNMENT
Definition: rdkafkacpp.h:419
@ ERR_BROKER_NOT_AVAILABLE
Definition: rdkafkacpp.h:351
@ ERR__UNKNOWN_PROTOCOL
Definition: rdkafkacpp.h:264
@ ERR__BAD_MSG
Definition: rdkafkacpp.h:205
@ ERR__NOENT
Definition: rdkafkacpp.h:294
@ ERR_REBALANCE_IN_PROGRESS
Definition: rdkafkacpp.h:395
@ ERR_COORDINATOR_LOAD_IN_PROGRESS
Definition: rdkafkacpp.h:363
@ ERR_RECORD_LIST_TOO_LARGE
Definition: rdkafkacpp.h:377
@ ERR_INVALID_PRODUCER_ID_MAPPING
Definition: rdkafkacpp.h:440
@ ERR__PARTIAL
Definition: rdkafkacpp.h:290
@ ERR_TRANSACTION_COORDINATOR_FENCED
Definition: rdkafkacpp.h:450
@ ERR_UNKNOWN_TOPIC_OR_PART
Definition: rdkafkacpp.h:341
@ ERR__UNKNOWN_TOPIC
Definition: rdkafkacpp.h:230
@ ERR__INVALID_ARG
Definition: rdkafkacpp.h:234
@ ERR_UNSUPPORTED_FOR_MESSAGE_FORMAT
Definition: rdkafkacpp.h:427
@ ERR_INVALID_TIMESTAMP
Definition: rdkafkacpp.h:405
@ ERR__INVALID_TYPE
Definition: rdkafkacpp.h:298
@ ERR_OFFSET_OUT_OF_RANGE
Definition: rdkafkacpp.h:337
@ ERR_INVALID_UPDATE_VERSION
Definition: rdkafkacpp.h:540
@ ERR_DELEGATION_TOKEN_AUTH_DISABLED
Definition: rdkafkacpp.h:468
@ ERR_POLICY_VIOLATION
Definition: rdkafkacpp.h:429
@ ERR__KEY_DESERIALIZATION
Definition: rdkafkacpp.h:286
@ ERR__FS
Definition: rdkafkacpp.h:228
@ ERR_GROUP_MAX_SIZE_REACHED
Definition: rdkafkacpp.h:508
@ ERR_NOT_CONTROLLER
Definition: rdkafkacpp.h:423
@ ERR__INTR
Definition: rdkafkacpp.h:280
@ ERR__TRANSPORT
Definition: rdkafkacpp.h:213
RD_EXPORT int version()
Returns the librdkafka version as integer.
RD_EXPORT void * mem_malloc(size_t size)
Allocate memory using the same allocator librdkafka uses.
CertificateType
SSL certificate types.
Definition: rdkafkacpp.h:560
@ CERT_PRIVATE_KEY
Definition: rdkafkacpp.h:562
@ CERT_PUBLIC_KEY
Definition: rdkafkacpp.h:561
@ CERT_CA
Definition: rdkafkacpp.h:563
RD_EXPORT void mem_free(void *ptr)
Free pointer returned by librdkafka.