You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by bq...@apache.org on 2017/11/16 21:16:03 UTC
[33/42] nifi-minifi-cpp git commit: MINIFICPP-274: PutKafka Processor
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9f66960e/thirdparty/librdkafka-0.11.1/src-cpp/rdkafkacpp.h
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.1/src-cpp/rdkafkacpp.h b/thirdparty/librdkafka-0.11.1/src-cpp/rdkafkacpp.h
new file mode 100644
index 0000000..43e2af0
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.1/src-cpp/rdkafkacpp.h
@@ -0,0 +1,2190 @@
+/*
+ * librdkafka - Apache Kafka C/C++ library
+ *
+ * Copyright (c) 2014 Magnus Edenhill
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright notice,
+ * this list of conditions and the following disclaimer in the documentation
+ * and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#pragma once
+
+/**
+ * @file rdkafkacpp.h
+ * @brief Apache Kafka C/C++ consumer and producer client library.
+ *
+ * rdkafkacpp.h contains the public C++ API for librdkafka.
+ * The API is documented in this file as comments prefixing the class,
+ * function, type, enum, define, etc.
+ * For more information, see the C interface in rdkafka.h and read the
+ * manual in INTRODUCTION.md.
+ * The C++ interface is STD C++ '03 compliant and adheres to the
+ * Google C++ Style Guide.
+
+ * @sa For the C interface see rdkafka.h
+ *
+ * @tableofcontents
+ */
+
+/**@cond NO_DOC*/
+#include <string>
+#include <list>
+#include <vector>
+#include <stdint.h>
+
+
+#ifdef _MSC_VER
+#undef RD_EXPORT
+#ifdef LIBRDKAFKA_STATICLIB
+#define RD_EXPORT
+#else
+#ifdef LIBRDKAFKACPP_EXPORTS
+#define RD_EXPORT __declspec(dllexport)
+#else
+#define RD_EXPORT __declspec(dllimport)
+#endif
+#endif
+#else
+#define RD_EXPORT
+#endif
+
+/**@endcond*/
+
+namespace RdKafka {
+
+
+/**
+ * @name Miscellaneous APIs
+ * @{
+ */
+
+/**
+ * @brief librdkafka version
+ *
+ * Interpreted as hex \c MM.mm.rr.xx:
+ * - MM = Major
+ * - mm = minor
+ * - rr = revision
+ * - xx = pre-release id (0xff is the final release)
+ *
+ * E.g.: \c 0x000801ff = 0.8.1
+ *
+ * @remark This value should only be used during compile time,
+ * for runtime checks of version use RdKafka::version()
+ */
+#define RD_KAFKA_VERSION 0x000b01ff
+
+/**
+ * @brief Returns the librdkafka version as integer.
+ *
+ * @sa See RD_KAFKA_VERSION for how to parse the integer format.
+ */
+RD_EXPORT
+int version ();
+
+/**
+ * @brief Returns the librdkafka version as string.
+ */
+RD_EXPORT
+std::string version_str();
+
+/**
+ * @brief Returns a CSV list of the supported debug contexts
+ * for use with Conf::Set("debug", ..).
+ */
+RD_EXPORT
+std::string get_debug_contexts();
+
+/**
+ * @brief Wait for all rd_kafka_t objects to be destroyed.
+ *
+ * @returns 0 if all kafka objects are now destroyed, or -1 if the
+ * timeout was reached.
+ * Since RdKafka handle deletion is an asynch operation the
+ * \p wait_destroyed() function can be used for applications where
+ * a clean shutdown is required.
+ */
+RD_EXPORT
+int wait_destroyed(int timeout_ms);
+
+
+/**@}*/
+
+
+
+/**
+ * @name Constants, errors, types
+ * @{
+ *
+ *
+ */
+
+/**
+ * @brief Error codes.
+ *
+ * The negative error codes delimited by two underscores
+ * (\c _ERR__..) denotes errors internal to librdkafka and are
+ * displayed as \c \"Local: \<error string..\>\", while the error codes
+ * delimited by a single underscore (\c ERR_..) denote broker
+ * errors and are displayed as \c \"Broker: \<error string..\>\".
+ *
+ * @sa Use RdKafka::err2str() to translate an error code a human readable string
+ */
+enum ErrorCode {
+ /* Internal errors to rdkafka: */
+ /** Begin internal error codes */
+ ERR__BEGIN = -200,
+ /** Received message is incorrect */
+ ERR__BAD_MSG = -199,
+ /** Bad/unknown compression */
+ ERR__BAD_COMPRESSION = -198,
+ /** Broker is going away */
+ ERR__DESTROY = -197,
+ /** Generic failure */
+ ERR__FAIL = -196,
+ /** Broker transport failure */
+ ERR__TRANSPORT = -195,
+ /** Critical system resource */
+ ERR__CRIT_SYS_RESOURCE = -194,
+ /** Failed to resolve broker */
+ ERR__RESOLVE = -193,
+ /** Produced message timed out*/
+ ERR__MSG_TIMED_OUT = -192,
+ /** Reached the end of the topic+partition queue on
+ * the broker. Not really an error. */
+ ERR__PARTITION_EOF = -191,
+ /** Permanent: Partition does not exist in cluster. */
+ ERR__UNKNOWN_PARTITION = -190,
+ /** File or filesystem error */
+ ERR__FS = -189,
+ /** Permanent: Topic does not exist in cluster. */
+ ERR__UNKNOWN_TOPIC = -188,
+ /** All broker connections are down. */
+ ERR__ALL_BROKERS_DOWN = -187,
+ /** Invalid argument, or invalid configuration */
+ ERR__INVALID_ARG = -186,
+ /** Operation timed out */
+ ERR__TIMED_OUT = -185,
+ /** Queue is full */
+ ERR__QUEUE_FULL = -184,
+ /** ISR count < required.acks */
+ ERR__ISR_INSUFF = -183,
+ /** Broker node update */
+ ERR__NODE_UPDATE = -182,
+ /** SSL error */
+ ERR__SSL = -181,
+ /** Waiting for coordinator to become available. */
+ ERR__WAIT_COORD = -180,
+ /** Unknown client group */
+ ERR__UNKNOWN_GROUP = -179,
+ /** Operation in progress */
+ ERR__IN_PROGRESS = -178,
+ /** Previous operation in progress, wait for it to finish. */
+ ERR__PREV_IN_PROGRESS = -177,
+ /** This operation would interfere with an existing subscription */
+ ERR__EXISTING_SUBSCRIPTION = -176,
+ /** Assigned partitions (rebalance_cb) */
+ ERR__ASSIGN_PARTITIONS = -175,
+ /** Revoked partitions (rebalance_cb) */
+ ERR__REVOKE_PARTITIONS = -174,
+ /** Conflicting use */
+ ERR__CONFLICT = -173,
+ /** Wrong state */
+ ERR__STATE = -172,
+ /** Unknown protocol */
+ ERR__UNKNOWN_PROTOCOL = -171,
+ /** Not implemented */
+ ERR__NOT_IMPLEMENTED = -170,
+ /** Authentication failure*/
+ ERR__AUTHENTICATION = -169,
+ /** No stored offset */
+ ERR__NO_OFFSET = -168,
+ /** Outdated */
+ ERR__OUTDATED = -167,
+ /** Timed out in queue */
+ ERR__TIMED_OUT_QUEUE = -166,
+ /** Feature not supported by broker */
+ ERR__UNSUPPORTED_FEATURE = -165,
+ /** Awaiting cache update */
+ ERR__WAIT_CACHE = -164,
+ /** Operation interrupted */
+ ERR__INTR = -163,
+ /** Key serialization error */
+ ERR__KEY_SERIALIZATION = -162,
+ /** Value serialization error */
+ ERR__VALUE_SERIALIZATION = -161,
+ /** Key deserialization error */
+ ERR__KEY_DESERIALIZATION = -160,
+ /** Value deserialization error */
+ ERR__VALUE_DESERIALIZATION = -159,
+ /** End internal error codes */
+ ERR__END = -100,
+
+ /* Kafka broker errors: */
+ /** Unknown broker error */
+ ERR_UNKNOWN = -1,
+ /** Success */
+ ERR_NO_ERROR = 0,
+ /** Offset out of range */
+ ERR_OFFSET_OUT_OF_RANGE = 1,
+ /** Invalid message */
+ ERR_INVALID_MSG = 2,
+ /** Unknown topic or partition */
+ ERR_UNKNOWN_TOPIC_OR_PART = 3,
+ /** Invalid message size */
+ ERR_INVALID_MSG_SIZE = 4,
+ /** Leader not available */
+ ERR_LEADER_NOT_AVAILABLE = 5,
+ /** Not leader for partition */
+ ERR_NOT_LEADER_FOR_PARTITION = 6,
+ /** Request timed out */
+ ERR_REQUEST_TIMED_OUT = 7,
+ /** Broker not available */
+ ERR_BROKER_NOT_AVAILABLE = 8,
+ /** Replica not available */
+ ERR_REPLICA_NOT_AVAILABLE = 9,
+ /** Message size too large */
+ ERR_MSG_SIZE_TOO_LARGE = 10,
+ /** StaleControllerEpochCode */
+ ERR_STALE_CTRL_EPOCH = 11,
+ /** Offset metadata string too large */
+ ERR_OFFSET_METADATA_TOO_LARGE = 12,
+ /** Broker disconnected before response received */
+ ERR_NETWORK_EXCEPTION = 13,
+ /** Group coordinator load in progress */
+ ERR_GROUP_LOAD_IN_PROGRESS = 14,
+ /** Group coordinator not available */
+ ERR_GROUP_COORDINATOR_NOT_AVAILABLE = 15,
+ /** Not coordinator for group */
+ ERR_NOT_COORDINATOR_FOR_GROUP = 16,
+ /** Invalid topic */
+ ERR_TOPIC_EXCEPTION = 17,
+ /** Message batch larger than configured server segment size */
+ ERR_RECORD_LIST_TOO_LARGE = 18,
+ /** Not enough in-sync replicas */
+ ERR_NOT_ENOUGH_REPLICAS = 19,
+ /** Message(s) written to insufficient number of in-sync replicas */
+ ERR_NOT_ENOUGH_REPLICAS_AFTER_APPEND = 20,
+ /** Invalid required acks value */
+ ERR_INVALID_REQUIRED_ACKS = 21,
+ /** Specified group generation id is not valid */
+ ERR_ILLEGAL_GENERATION = 22,
+ /** Inconsistent group protocol */
+ ERR_INCONSISTENT_GROUP_PROTOCOL = 23,
+ /** Invalid group.id */
+ ERR_INVALID_GROUP_ID = 24,
+ /** Unknown member */
+ ERR_UNKNOWN_MEMBER_ID = 25,
+ /** Invalid session timeout */
+ ERR_INVALID_SESSION_TIMEOUT = 26,
+ /** Group rebalance in progress */
+ ERR_REBALANCE_IN_PROGRESS = 27,
+ /** Commit offset data size is not valid */
+ ERR_INVALID_COMMIT_OFFSET_SIZE = 28,
+ /** Topic authorization failed */
+ ERR_TOPIC_AUTHORIZATION_FAILED = 29,
+ /** Group authorization failed */
+ ERR_GROUP_AUTHORIZATION_FAILED = 30,
+ /** Cluster authorization failed */
+ ERR_CLUSTER_AUTHORIZATION_FAILED = 31,
+ /** Invalid timestamp */
+ ERR_INVALID_TIMESTAMP = 32,
+ /** Unsupported SASL mechanism */
+ ERR_UNSUPPORTED_SASL_MECHANISM = 33,
+ /** Illegal SASL state */
+ ERR_ILLEGAL_SASL_STATE = 34,
+ /** Unuspported version */
+ ERR_UNSUPPORTED_VERSION = 35,
+ /** Topic already exists */
+ ERR_TOPIC_ALREADY_EXISTS = 36,
+ /** Invalid number of partitions */
+ ERR_INVALID_PARTITIONS = 37,
+ /** Invalid replication factor */
+ ERR_INVALID_REPLICATION_FACTOR = 38,
+ /** Invalid replica assignment */
+ ERR_INVALID_REPLICA_ASSIGNMENT = 39,
+ /** Invalid config */
+ ERR_INVALID_CONFIG = 40,
+ /** Not controller for cluster */
+ ERR_NOT_CONTROLLER = 41,
+ /** Invalid request */
+ ERR_INVALID_REQUEST = 42,
+ /** Message format on broker does not support request */
+ ERR_UNSUPPORTED_FOR_MESSAGE_FORMAT = 43,
+ /** Isolation policy volation */
+ ERR_POLICY_VIOLATION = 44,
+ /** Broker received an out of order sequence number */
+ ERR_OUT_OF_ORDER_SEQUENCE_NUMBER = 45,
+ /** Broker received a duplicate sequence number */
+ ERR_DUPLICATE_SEQUENCE_NUMBER = 46,
+ /** Producer attempted an operation with an old epoch */
+ ERR_INVALID_PRODUCER_EPOCH = 47,
+ /** Producer attempted a transactional operation in an invalid state */
+ ERR_INVALID_TXN_STATE = 48,
+ /** Producer attempted to use a producer id which is not
+ * currently assigned to its transactional id */
+ ERR_INVALID_PRODUCER_ID_MAPPING = 49,
+ /** Transaction timeout is larger than the maximum
+ * value allowed by the broker's max.transaction.timeout.ms */
+ ERR_INVALID_TRANSACTION_TIMEOUT = 50,
+ /** Producer attempted to update a transaction while another
+ * concurrent operation on the same transaction was ongoing */
+ ERR_CONCURRENT_TRANSACTIONS = 51,
+ /** Indicates that the transaction coordinator sending a
+ * WriteTxnMarker is no longer the current coordinator for a
+ * given producer */
+ ERR_TRANSACTION_COORDINATOR_FENCED = 52,
+ /** Transactional Id authorization failed */
+ ERR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED = 53,
+ /** Security features are disabled */
+ ERR_SECURITY_DISABLED = 54,
+ /** Operation not attempted */
+ ERR_OPERATION_NOT_ATTEMPTED = 55
+};
+
+
+/**
+ * @brief Returns a human readable representation of a kafka error.
+ */
+RD_EXPORT
+std::string err2str(RdKafka::ErrorCode err);
+
+
+/**@} */
+
+
+
+/**@cond NO_DOC*/
+/* Forward declarations */
+class Producer;
+class Message;
+class Queue;
+class Event;
+class Topic;
+class TopicPartition;
+class Metadata;
+class KafkaConsumer;
+/**@endcond*/
+
+
+/**
+ * @name Callback classes
+ * @{
+ *
+ *
+ * librdkafka uses (optional) callbacks to propagate information and
+ * delegate decisions to the application logic.
+ *
+ * An application must call RdKafka::poll() at regular intervals to
+ * serve queued callbacks.
+ */
+
+
+/**
+ * @brief Delivery Report callback class
+ *
+ * The delivery report callback will be called once for each message
+ * accepted by RdKafka::Producer::produce() (et.al) with
+ * RdKafka::Message::err() set to indicate the result of the produce request.
+ *
+ * The callback is called when a message is succesfully produced or
+ * if librdkafka encountered a permanent failure, or the retry counter for
+ * temporary errors has been exhausted.
+ *
+ * An application must call RdKafka::poll() at regular intervals to
+ * serve queued delivery report callbacks.
+
+ */
+class RD_EXPORT DeliveryReportCb {
+ public:
+ /**
+ * @brief Delivery report callback.
+ */
+ virtual void dr_cb (Message &message) = 0;
+
+ virtual ~DeliveryReportCb() { }
+};
+
+
+/**
+ * @brief Partitioner callback class
+ *
+ * Generic partitioner callback class for implementing custom partitioners.
+ *
+ * @sa RdKafka::Conf::set() \c "partitioner_cb"
+ */
+class RD_EXPORT PartitionerCb {
+ public:
+ /**
+ * @brief Partitioner callback
+ *
+ * Return the partition to use for \p key in \p topic.
+ *
+ * The \p msg_opaque is the same \p msg_opaque provided in the
+ * RdKafka::Producer::produce() call.
+ *
+ * @remark \p key may be NULL or the empty.
+ *
+ * @returns Must return a value between 0 and \p partition_cnt (non-inclusive).
+ * May return RD_KAFKA_PARTITION_UA (-1) if partitioning failed.
+ *
+ * @sa The callback may use RdKafka::Topic::partition_available() to check
+ * if a partition has an active leader broker.
+ */
+ virtual int32_t partitioner_cb (const Topic *topic,
+ const std::string *key,
+ int32_t partition_cnt,
+ void *msg_opaque) = 0;
+
+ virtual ~PartitionerCb() { }
+};
+
+/**
+ * @brief Variant partitioner with key pointer
+ *
+ */
+class PartitionerKeyPointerCb {
+ public:
+ /**
+ * @brief Variant partitioner callback that gets \p key as pointer and length
+ * instead of as a const std::string *.
+ *
+ * @remark \p key may be NULL or have \p key_len 0.
+ *
+ * @sa See RdKafka::PartitionerCb::partitioner_cb() for exact semantics
+ */
+ virtual int32_t partitioner_cb (const Topic *topic,
+ const void *key,
+ size_t key_len,
+ int32_t partition_cnt,
+ void *msg_opaque) = 0;
+
+ virtual ~PartitionerKeyPointerCb() { }
+};
+
+
+
+/**
+ * @brief Event callback class
+ *
+ * Events are a generic interface for propagating errors, statistics, logs, etc
+ * from librdkafka to the application.
+ *
+ * @sa RdKafka::Event
+ */
+class RD_EXPORT EventCb {
+ public:
+ /**
+ * @brief Event callback
+ *
+ * @sa RdKafka::Event
+ */
+ virtual void event_cb (Event &event) = 0;
+
+ virtual ~EventCb() { }
+};
+
+
+/**
+ * @brief Event object class as passed to the EventCb callback.
+ */
+class RD_EXPORT Event {
+ public:
+ /** @brief Event type */
+ enum Type {
+ EVENT_ERROR, /**< Event is an error condition */
+ EVENT_STATS, /**< Event is a statistics JSON document */
+ EVENT_LOG, /**< Event is a log message */
+ EVENT_THROTTLE /**< Event is a throttle level signaling from the broker */
+ };
+
+ /** @brief EVENT_LOG severities (conforms to syslog(3) severities) */
+ enum Severity {
+ EVENT_SEVERITY_EMERG = 0,
+ EVENT_SEVERITY_ALERT = 1,
+ EVENT_SEVERITY_CRITICAL = 2,
+ EVENT_SEVERITY_ERROR = 3,
+ EVENT_SEVERITY_WARNING = 4,
+ EVENT_SEVERITY_NOTICE = 5,
+ EVENT_SEVERITY_INFO = 6,
+ EVENT_SEVERITY_DEBUG = 7
+ };
+
+ virtual ~Event () { }
+
+ /*
+ * Event Accessor methods
+ */
+
+ /**
+ * @returns The event type
+ * @remark Applies to all event types
+ */
+ virtual Type type () const = 0;
+
+ /**
+ * @returns Event error, if any.
+ * @remark Applies to all event types except THROTTLE
+ */
+ virtual ErrorCode err () const = 0;
+
+ /**
+ * @returns Log severity level.
+ * @remark Applies to LOG event type.
+ */
+ virtual Severity severity () const = 0;
+
+ /**
+ * @returns Log facility string.
+ * @remark Applies to LOG event type.
+ */
+ virtual std::string fac () const = 0;
+
+ /**
+ * @returns Log message string.
+ *
+ * \c EVENT_LOG: Log message string.
+ * \c EVENT_STATS: JSON object (as string).
+ *
+ * @remark Applies to LOG event type.
+ */
+ virtual std::string str () const = 0;
+
+ /**
+ * @returns Throttle time in milliseconds.
+ * @remark Applies to THROTTLE event type.
+ */
+ virtual int throttle_time () const = 0;
+
+ /**
+ * @returns Throttling broker's name.
+ * @remark Applies to THROTTLE event type.
+ */
+ virtual std::string broker_name () const = 0;
+
+ /**
+ * @returns Throttling broker's id.
+ * @remark Applies to THROTTLE event type.
+ */
+ virtual int broker_id () const = 0;
+};
+
+
+
+/**
+ * @brief Consume callback class
+ */
+class RD_EXPORT ConsumeCb {
+ public:
+ /**
+ * @brief The consume callback is used with
+ * RdKafka::Consumer::consume_callback()
+ * methods and will be called for each consumed \p message.
+ *
+ * The callback interface is optional but provides increased performance.
+ */
+ virtual void consume_cb (Message &message, void *opaque) = 0;
+
+ virtual ~ConsumeCb() { }
+};
+
+
+/**
+ * @brief \b KafkaConsunmer: Rebalance callback class
+ */
+class RD_EXPORT RebalanceCb {
+public:
+ /**
+ * @brief Group rebalance callback for use with RdKafka::KafkaConsunmer
+ *
+ * Registering a \p rebalance_cb turns off librdkafka's automatic
+ * partition assignment/revocation and instead delegates that responsibility
+ * to the application's \p rebalance_cb.
+ *
+ * The rebalance callback is responsible for updating librdkafka's
+ * assignment set based on the two events: RdKafka::ERR__ASSIGN_PARTITIONS
+ * and RdKafka::ERR__REVOKE_PARTITIONS but should also be able to handle
+ * arbitrary rebalancing failures where \p err is neither of those.
+ * @remark In this latter case (arbitrary error), the application must
+ * call unassign() to synchronize state.
+
+ *
+ * Without a rebalance callback this is done automatically by librdkafka
+ * but registering a rebalance callback gives the application flexibility
+ * in performing other operations along with the assinging/revocation,
+ * such as fetching offsets from an alternate location (on assign)
+ * or manually committing offsets (on revoke).
+ *
+ * The following example show's the application's responsibilities:
+ * @code
+ * class MyRebalanceCb : public RdKafka::RebalanceCb {
+ * public:
+ * void rebalance_cb (RdKafka::KafkaConsumer *consumer,
+ * RdKafka::ErrorCode err,
+ * std::vector<RdKafka::TopicPartition*> &partitions) {
+ * if (err == RdKafka::ERR__ASSIGN_PARTITIONS) {
+ * // application may load offets from arbitrary external
+ * // storage here and update \p partitions
+ *
+ * consumer->assign(partitions);
+ *
+ * } else if (err == RdKafka::ERR__REVOKE_PARTITIONS) {
+ * // Application may commit offsets manually here
+ * // if auto.commit.enable=false
+ *
+ * consumer->unassign();
+ *
+ * } else {
+ * std::cerr << "Rebalancing error: <<
+ * RdKafka::err2str(err) << std::endl;
+ * consumer->unassign();
+ * }
+ * }
+ * }
+ * @endcode
+ */
+ virtual void rebalance_cb (RdKafka::KafkaConsumer *consumer,
+ RdKafka::ErrorCode err,
+ std::vector<TopicPartition*>&partitions) = 0;
+
+ virtual ~RebalanceCb() { }
+};
+
+
+/**
+ * @brief Offset Commit callback class
+ */
+class RD_EXPORT OffsetCommitCb {
+public:
+ /**
+ * @brief Set offset commit callback for use with consumer groups
+ *
+ * The results of automatic or manual offset commits will be scheduled
+ * for this callback and is served by RdKafka::KafkaConsumer::consume().
+ *
+ * If no partitions had valid offsets to commit this callback will be called
+ * with \p err == ERR__NO_OFFSET which is not to be considered an error.
+ *
+ * The \p offsets list contains per-partition information:
+ * - \c topic The topic committed
+ * - \c partition The partition committed
+ * - \c offset: Committed offset (attempted)
+ * - \c err: Commit error
+ */
+ virtual void offset_commit_cb(RdKafka::ErrorCode err,
+ std::vector<TopicPartition*>&offsets) = 0;
+
+ virtual ~OffsetCommitCb() { }
+};
+
+
+
+/**
+ * @brief \b Portability: SocketCb callback class
+ *
+ */
+class RD_EXPORT SocketCb {
+ public:
+ /**
+ * @brief Socket callback
+ *
+ * The socket callback is responsible for opening a socket
+ * according to the supplied \p domain, \p type and \p protocol.
+ * The socket shall be created with \c CLOEXEC set in a racefree fashion, if
+ * possible.
+ *
+ * It is typically not required to register an alternative socket
+ * implementation
+ *
+ * @returns The socket file descriptor or -1 on error (\c errno must be set)
+ */
+ virtual int socket_cb (int domain, int type, int protocol) = 0;
+
+ virtual ~SocketCb() { }
+};
+
+
+/**
+ * @brief \b Portability: OpenCb callback class
+ *
+ */
+class RD_EXPORT OpenCb {
+ public:
+ /**
+ * @brief Open callback
+ * The open callback is responsible for opening the file specified by
+ * \p pathname, using \p flags and \p mode.
+ * The file shall be opened with \c CLOEXEC set in a racefree fashion, if
+ * possible.
+ *
+ * It is typically not required to register an alternative open implementation
+ *
+ * @remark Not currently available on native Win32
+ */
+ virtual int open_cb (const std::string &path, int flags, int mode) = 0;
+
+ virtual ~OpenCb() { }
+};
+
+
+/**@}*/
+
+
+
+
+/**
+ * @name Configuration interface
+ * @{
+ *
+ */
+
+/**
+ * @brief Configuration interface
+ *
+ * Holds either global or topic configuration that are passed to
+ * RdKafka::Consumer::create(), RdKafka::Producer::create(),
+ * RdKafka::KafkaConsumer::create(), etc.
+ *
+ * @sa CONFIGURATION.md for the full list of supported properties.
+ */
+class RD_EXPORT Conf {
+ public:
+ /**
+ * @brief Configuration object type
+ */
+ enum ConfType {
+ CONF_GLOBAL, /**< Global configuration */
+ CONF_TOPIC /**< Topic specific configuration */
+ };
+
+ /**
+ * @brief RdKafka::Conf::Set() result code
+ */
+ enum ConfResult {
+ CONF_UNKNOWN = -2, /**< Unknown configuration property */
+ CONF_INVALID = -1, /**< Invalid configuration value */
+ CONF_OK = 0 /**< Configuration property was succesfully set */
+ };
+
+
+ /**
+ * @brief Create configuration object
+ */
+ static Conf *create (ConfType type);
+
+ virtual ~Conf () { }
+
+ /**
+ * @brief Set configuration property \p name to value \p value.
+ *
+ * Fallthrough:
+ * Topic-level configuration properties may be set using this interface
+ * in which case they are applied on the \c default_topic_conf.
+ * If no \c default_topic_conf has been set one will be created.
+ * Any sub-sequent set("default_topic_conf", ..) calls will
+ * replace the current default topic configuration.
+
+ * @returns CONF_OK on success, else writes a human readable error
+ * description to \p errstr on error.
+ */
+ virtual Conf::ConfResult set (const std::string &name,
+ const std::string &value,
+ std::string &errstr) = 0;
+
+ /** @brief Use with \p name = \c \"dr_cb\" */
+ virtual Conf::ConfResult set (const std::string &name,
+ DeliveryReportCb *dr_cb,
+ std::string &errstr) = 0;
+
+ /** @brief Use with \p name = \c \"event_cb\" */
+ virtual Conf::ConfResult set (const std::string &name,
+ EventCb *event_cb,
+ std::string &errstr) = 0;
+
+ /** @brief Use with \p name = \c \"default_topic_conf\"
+ *
+ * Sets the default topic configuration to use for for automatically
+ * subscribed topics.
+ *
+ * @sa RdKafka::KafkaConsumer::subscribe()
+ */
+ virtual Conf::ConfResult set (const std::string &name,
+ const Conf *topic_conf,
+ std::string &errstr) = 0;
+
+ /** @brief Use with \p name = \c \"partitioner_cb\" */
+ virtual Conf::ConfResult set (const std::string &name,
+ PartitionerCb *partitioner_cb,
+ std::string &errstr) = 0;
+
+ /** @brief Use with \p name = \c \"partitioner_key_pointer_cb\" */
+ virtual Conf::ConfResult set (const std::string &name,
+ PartitionerKeyPointerCb *partitioner_kp_cb,
+ std::string &errstr) = 0;
+
+ /** @brief Use with \p name = \c \"socket_cb\" */
+ virtual Conf::ConfResult set (const std::string &name, SocketCb *socket_cb,
+ std::string &errstr) = 0;
+
+ /** @brief Use with \p name = \c \"open_cb\" */
+ virtual Conf::ConfResult set (const std::string &name, OpenCb *open_cb,
+ std::string &errstr) = 0;
+
+ /** @brief Use with \p name = \c \"rebalance_cb\" */
+ virtual Conf::ConfResult set (const std::string &name,
+ RebalanceCb *rebalance_cb,
+ std::string &errstr) = 0;
+
+ /** @brief Use with \p name = \c \"offset_commit_cb\" */
+ virtual Conf::ConfResult set (const std::string &name,
+ OffsetCommitCb *offset_commit_cb,
+ std::string &errstr) = 0;
+
+ /** @brief Query single configuration value
+ *
+ * Do not use this method to get callbacks registered by the configuration file.
+ * Instead use the specific get() methods with the specific callback parameter in the signature.
+ *
+ * Fallthrough:
+ * Topic-level configuration properties from the \c default_topic_conf
+ * may be retrieved using this interface.
+ *
+ * @returns CONF_OK if the property was set previously set and
+ * returns the value in \p value. */
+ virtual Conf::ConfResult get(const std::string &name,
+ std::string &value) const = 0;
+
+ /** @brief Query single configuration value
+ * @returns CONF_OK if the property was set previously set and
+ * returns the value in \p dr_cb. */
+ virtual Conf::ConfResult get(DeliveryReportCb *&dr_cb) const = 0;
+
+ /** @brief Query single configuration value
+ * @returns CONF_OK if the property was set previously set and
+ * returns the value in \p event_cb. */
+ virtual Conf::ConfResult get(EventCb *&event_cb) const = 0;
+
+ /** @brief Query single configuration value
+ * @returns CONF_OK if the property was set previously set and
+ * returns the value in \p partitioner_cb. */
+ virtual Conf::ConfResult get(PartitionerCb *&partitioner_cb) const = 0;
+
+ /** @brief Query single configuration value
+ * @returns CONF_OK if the property was set previously set and
+ * returns the value in \p partitioner_kp_cb. */
+ virtual Conf::ConfResult get(PartitionerKeyPointerCb *&partitioner_kp_cb) const = 0;
+
+ /** @brief Query single configuration value
+ * @returns CONF_OK if the property was set previously set and
+ * returns the value in \p socket_cb. */
+ virtual Conf::ConfResult get(SocketCb *&socket_cb) const = 0;
+
+ /** @brief Query single configuration value
+ * @returns CONF_OK if the property was set previously set and
+ * returns the value in \p open_cb. */
+ virtual Conf::ConfResult get(OpenCb *&open_cb) const = 0;
+
+ /** @brief Query single configuration value
+ * @returns CONF_OK if the property was set previously set and
+ * returns the value in \p rebalance_cb. */
+ virtual Conf::ConfResult get(RebalanceCb *&rebalance_cb) const = 0;
+
+ /** @brief Query single configuration value
+ * @returns CONF_OK if the property was set previously set and
+ * returns the value in \p offset_commit_cb. */
+ virtual Conf::ConfResult get(OffsetCommitCb *&offset_commit_cb) const = 0;
+
+ /** @brief Dump configuration names and values to list containing
+ * name,value tuples */
+ virtual std::list<std::string> *dump () = 0;
+
+ /** @brief Use with \p name = \c \"consume_cb\" */
+ virtual Conf::ConfResult set (const std::string &name, ConsumeCb *consume_cb,
+ std::string &errstr) = 0;
+};
+
+/**@}*/
+
+
+/**
+ * @name Kafka base client handle
+ * @{
+ *
+ */
+
+/**
+ * @brief Base handle, super class for specific clients.
+ */
+class RD_EXPORT Handle {
+ public:
+ virtual ~Handle() { }
+
+ /** @returns the name of the handle */
+ virtual const std::string name () const = 0;
+
+ /**
+ * @brief Returns the client's broker-assigned group member id
+ *
+ * @remark This currently requires the high-level KafkaConsumer
+ *
+ * @returns Last assigned member id, or empty string if not currently
+ * a group member.
+ */
+ virtual const std::string memberid () const = 0;
+
+
+ /**
+ * @brief Polls the provided kafka handle for events.
+ *
+ * Events will trigger application provided callbacks to be called.
+ *
+ * The \p timeout_ms argument specifies the maximum amount of time
+ * (in milliseconds) that the call will block waiting for events.
+ * For non-blocking calls, provide 0 as \p timeout_ms.
+ * To wait indefinately for events, provide -1.
+ *
+ * Events:
+ * - delivery report callbacks (if an RdKafka::DeliveryCb is configured) [producer]
+ * - event callbacks (if an RdKafka::EventCb is configured) [producer & consumer]
+ *
+ * @remark An application should make sure to call poll() at regular
+ * intervals to serve any queued callbacks waiting to be called.
+ *
+ * @warning This method MUST NOT be used with the RdKafka::KafkaConsumer,
+ * use its RdKafka::KafkaConsumer::consume() instead.
+ *
+ * @returns the number of events served.
+ */
+ virtual int poll (int timeout_ms) = 0;
+
+ /**
+ * @brief Returns the current out queue length
+ *
+ * The out queue contains messages and requests waiting to be sent to,
+ * or acknowledged by, the broker.
+ */
+ virtual int outq_len () = 0;
+
+ /**
+ * @brief Request Metadata from broker.
+ *
+ * Parameters:
+ * \p all_topics - if non-zero: request info about all topics in cluster,
+ * if zero: only request info about locally known topics.
+ * \p only_rkt - only request info about this topic
+ * \p metadatap - pointer to hold metadata result.
+ * The \p *metadatap pointer must be released with \c delete.
+ * \p timeout_ms - maximum response time before failing.
+ *
+ * @returns RdKafka::ERR_NO_ERROR on success (in which case \p *metadatap
+ * will be set), else RdKafka::ERR__TIMED_OUT on timeout or
+ * other error code on error.
+ */
+ virtual ErrorCode metadata (bool all_topics, const Topic *only_rkt,
+ Metadata **metadatap, int timeout_ms) = 0;
+
+
+ /**
+ * @brief Pause producing or consumption for the provided list of partitions.
+ *
+ * Success or error is returned per-partition in the \p partitions list.
+ *
+ * @returns ErrorCode::NO_ERROR
+ *
+ * @sa resume()
+ */
+ virtual ErrorCode pause (std::vector<TopicPartition*> &partitions) = 0;
+
+
+ /**
+ * @brief Resume producing or consumption for the provided list of partitions.
+ *
+ * Success or error is returned per-partition in the \p partitions list.
+ *
+ * @returns ErrorCode::NO_ERROR
+ *
+ * @sa pause()
+ */
+ virtual ErrorCode resume (std::vector<TopicPartition*> &partitions) = 0;
+
+
+ /**
+ * @brief Query broker for low (oldest/beginning)
+ * and high (newest/end) offsets for partition.
+ *
+ * Offsets are returned in \p *low and \p *high respectively.
+ *
+ * @returns RdKafka::ERR_NO_ERROR on success or an error code on failure.
+ */
+ virtual ErrorCode query_watermark_offsets (const std::string &topic,
+ int32_t partition,
+ int64_t *low, int64_t *high,
+ int timeout_ms) = 0;
+
+ /**
+ * @brief Get last known low (oldest/beginning)
+ * and high (newest/end) offsets for partition.
+ *
+ * The low offset is updated periodically (if statistics.interval.ms is set)
+ * while the high offset is updated on each fetched message set from the
+ * broker.
+ *
+ * If there is no cached offset (either low or high, or both) then
+ * OFFSET_INVALID will be returned for the respective offset.
+ *
+ * Offsets are returned in \p *low and \p *high respectively.
+ *
+ * @returns RdKafka::ERR_NO_ERROR on success or an error code on failure.
+ *
+ * @remark Shall only be used with an active consumer instance.
+ */
+ virtual ErrorCode get_watermark_offsets (const std::string &topic,
+ int32_t partition,
+ int64_t *low, int64_t *high) = 0;
+
+
+ /**
+ * @brief Look up the offsets for the given partitions by timestamp.
+ *
+ * The returned offset for each partition is the earliest offset whose
+ * timestamp is greater than or equal to the given timestamp in the
+ * corresponding partition.
+ *
+ * The timestamps to query are represented as \c offset in \p offsets
+ * on input, and \c offset() will return the closest earlier offset
+ * for the timestamp on output.
+ *
+ * The function will block for at most \p timeout_ms milliseconds.
+ *
+ * @remark Duplicate Topic+Partitions are not supported.
+ * @remark Errors are also returned per TopicPartition, see \c err()
+ *
+ * @returns an error code for general errors, else RdKafka::ERR_NO_ERROR
+ * in which case per-partition errors might be set.
+ */
+ virtual ErrorCode offsetsForTimes (std::vector<TopicPartition*> &offsets,
+ int timeout_ms) = 0;
+
+
+ /**
+ * @brief Retrieve queue for a given partition.
+ *
+ * @returns The fetch queue for the given partition if successful. Else,
+ * NULL is returned.
+ *
+ * @remark This function only works on consumers.
+ */
+ virtual Queue *get_partition_queue (const TopicPartition *partition) = 0;
+
+ /**
+ * @brief Forward librdkafka logs (and debug) to the specified queue
+ * for serving with one of the ..poll() calls.
+ *
+ * This allows an application to serve log callbacks (\c log_cb)
+ * in its thread of choice.
+ *
+ * @param queue Queue to forward logs to. If the value is NULL the logs
+ * are forwarded to the main queue.
+ *
+ * @remark The configuration property \c log.queue MUST also be set to true.
+ *
+ * @remark librdkafka maintains its own reference to the provided queue.
+ *
+ * @returns ERR_NO_ERROR on success or an error code on error.
+ */
+ virtual ErrorCode set_log_queue (Queue *queue) = 0;
+
+ /**
+ * @brief Cancels the current callback dispatcher (Producer::poll(),
+ * Consumer::poll(), KafkaConsumer::consume(), etc).
+ *
+ * A callback may use this to force an immediate return to the calling
+ * code (caller of e.g. ..::poll()) without processing any further
+ * events.
+ *
+ * @remark This function MUST ONLY be called from within a
+ * librdkafka callback.
+ */
+ virtual void yield () = 0;
+
+ /**
+ * @brief Returns the ClusterId as reported in broker metadata.
+ *
+ * @param timeout_ms If there is no cached value from metadata retrieval
+ * then this specifies the maximum amount of time
+ * (in milliseconds) the call will block waiting
+ * for metadata to be retrieved.
+ * Use 0 for non-blocking calls.
+ *
+ * @remark Requires broker version >=0.10.0 and api.version.request=true.
+ *
+ * @returns Last cached ClusterId, or empty string if no ClusterId could be
+ * retrieved in the allotted timespan.
+ */
+ virtual const std::string clusterid (int timeout_ms) = 0;
+};
+
+
+/**@}*/
+
+
+/**
+ * @name Topic and partition objects
+ * @{
+ *
+ */
+
+/**
+ * @brief Topic+Partition
+ *
+ * This is a generic type to hold a single partition and various
+ * information about it.
+ *
+ * Is typically used with std::vector<RdKafka::TopicPartition*> to provide
+ * a list of partitions for different operations.
+ */
+class RD_EXPORT TopicPartition {
+public:
+ /**
+ * Create topic+partition object for \p topic and \p partition
+ * and optionally \p offset.
+ *
+ * Use \c delete to deconstruct.
+ */
+ static TopicPartition *create (const std::string &topic, int partition);
+ static TopicPartition *create (const std::string &topic, int partition,
+ int64_t offset);
+
+ virtual ~TopicPartition() = 0;
+
+ /**
+ * @brief Destroy/delete the TopicPartitions in \p partitions
+ * and clear the vector.
+ */
+ static void destroy (std::vector<TopicPartition*> &partitions);
+
+ /** @returns topic name */
+ virtual const std::string &topic () const = 0;
+
+ /** @returns partition id */
+ virtual int partition () const = 0;
+
+ /** @returns offset (if applicable) */
+ virtual int64_t offset () const = 0;
+
+ /** @brief Set offset */
+ virtual void set_offset (int64_t offset) = 0;
+
+ /** @returns error code (if applicable) */
+ virtual ErrorCode err () const = 0;
+};
+
+
+
+/**
+ * @brief Topic handle
+ *
+ */
+class RD_EXPORT Topic {
+ public:
+ /**
+ * @brief Unassigned partition.
+ *
+ * The unassigned partition is used by the producer API for messages
+ * that should be partitioned using the configured or default partitioner.
+ */
+ static const int32_t PARTITION_UA;
+
+ /** @brief Special offsets */
+ static const int64_t OFFSET_BEGINNING; /**< Consume from beginning */
+ static const int64_t OFFSET_END; /**< Consume from end */
+ static const int64_t OFFSET_STORED; /**< Use offset storage */
+ static const int64_t OFFSET_INVALID; /**< Invalid offset */
+
+
+ /**
+ * @brief Creates a new topic handle for topic named \p topic_str
+ *
+ * \p conf is an optional configuration for the topic that will be used
+ * instead of the default topic configuration.
+ * The \p conf object is reusable after this call.
+ *
+ * @returns the new topic handle or NULL on error (see \p errstr).
+ */
+ static Topic *create (Handle *base, const std::string &topic_str,
+ Conf *conf, std::string &errstr);
+
+ virtual ~Topic () = 0;
+
+
+ /** @returns the topic name */
+ virtual const std::string name () const = 0;
+
+ /**
+ * @returns true if \p partition is available for the topic (has leader).
+ * @warning \b MUST \b ONLY be called from within a
+ * RdKafka::PartitionerCb callback.
+ */
+ virtual bool partition_available (int32_t partition) const = 0;
+
+ /**
+ * @brief Store offset \p offset for topic partition \p partition.
+ * The offset will be committed (written) to the offset store according
+ * to \p auto.commit.interval.ms.
+ *
+ * @remark \c enable.auto.offset.store must be set to \c false when using this API.
+ *
+ * @returns RdKafka::ERR_NO_ERROR on success or an error code if none of the
+ * offsets could be stored.
+ */
+ virtual ErrorCode offset_store (int32_t partition, int64_t offset) = 0;
+};
+
+
+/**@}*/
+
+
+/**
+ * @name Message object
+ * @{
+ *
+ */
+
+
+/**
+ * @brief Message timestamp object
+ *
+ * Represents the number of milliseconds since the epoch (UTC).
+ *
+ * The MessageTimestampType dictates the timestamp type or origin.
+ *
+ * @remark Requires Apache Kafka broker version >= 0.10.0
+ *
+ */
+
+class RD_EXPORT MessageTimestamp {
+public:
+ enum MessageTimestampType {
+ MSG_TIMESTAMP_NOT_AVAILABLE, /**< Timestamp not available */
+ MSG_TIMESTAMP_CREATE_TIME, /**< Message creation time (source) */
+ MSG_TIMESTAMP_LOG_APPEND_TIME /**< Message log append time (broker) */
+ };
+
+ MessageTimestampType type; /**< Timestamp type */
+ int64_t timestamp; /**< Milliseconds since epoch (UTC). */
+};
+
+
+
+/**
+ * @brief Message object
+ *
+ * This object represents either a single consumed or produced message,
+ * or an event (\p err() is set).
+ *
+ * An application must check RdKafka::Message::err() to see if the
+ * object is a proper message (error is RdKafka::ERR_NO_ERROR) or a
+ * an error event.
+ *
+ */
+class RD_EXPORT Message {
+ public:
+ /**
+ * @brief Accessor functions*
+ * @remark Not all fields are present in all types of callbacks.
+ */
+
+ /** @returns The error string if object represent an error event,
+ * else an empty string. */
+ virtual std::string errstr() const = 0;
+
+ /** @returns The error code if object represents an error event, else 0. */
+ virtual ErrorCode err () const = 0;
+
+ /** @returns the RdKafka::Topic object for a message (if applicable),
+ * or NULL if a corresponding RdKafka::Topic object has not been
+ * explicitly created with RdKafka::Topic::create().
+ * In this case use topic_name() instead. */
+ virtual Topic *topic () const = 0;
+
+ /** @returns Topic name (if applicable, else empty string) */
+ virtual std::string topic_name () const = 0;
+
+ /** @returns Partition (if applicable) */
+ virtual int32_t partition () const = 0;
+
+ /** @returns Message payload (if applicable) */
+ virtual void *payload () const = 0 ;
+
+ /** @returns Message payload length (if applicable) */
+ virtual size_t len () const = 0;
+
+ /** @returns Message key as string (if applicable) */
+ virtual const std::string *key () const = 0;
+
+ /** @returns Message key as void pointer (if applicable) */
+ virtual const void *key_pointer () const = 0 ;
+
+ /** @returns Message key's binary length (if applicable) */
+ virtual size_t key_len () const = 0;
+
+ /** @returns Message or error offset (if applicable) */
+ virtual int64_t offset () const = 0;
+
+ /** @returns Message timestamp (if applicable) */
+ virtual MessageTimestamp timestamp () const = 0;
+
+ /** @returns The \p msg_opaque as provided to RdKafka::Producer::produce() */
+ virtual void *msg_opaque () const = 0;
+
+ virtual ~Message () = 0;
+
+ /** @returns the latency in microseconds for a produced message measured
+ * from the produce() call, or -1 if latency is not available. */
+ virtual int64_t latency () const = 0;
+};
+
+/**@}*/
+
+
+/**
+ * @name Queue interface
+ * @{
+ *
+ */
+
+
+/**
+ * @brief Queue interface
+ *
+ * Create a new message queue. Message queues allows the application
+ * to re-route consumed messages from multiple topic+partitions into
+ * one single queue point. This queue point, containing messages from
+ * a number of topic+partitions, may then be served by a single
+ * consume() method, rather than one per topic+partition combination.
+ *
+ * See the RdKafka::Consumer::start(), RdKafka::Consumer::consume(), and
+ * RdKafka::Consumer::consume_callback() methods that take a queue as the first
+ * parameter for more information.
+ */
+class RD_EXPORT Queue {
+ public:
+ /**
+ * @brief Create Queue object
+ */
+ static Queue *create (Handle *handle);
+
+ /**
+ * @brief Forward/re-route queue to \p dst.
+ * If \p dst is \c NULL, the forwarding is removed.
+ *
+ * The internal refcounts for both queues are increased.
+ *
+ * @remark Regardless of whether \p dst is NULL or not, after calling this
+ * function, \p src will not forward it's fetch queue to the consumer
+ * queue.
+ */
+ virtual ErrorCode forward (Queue *dst) = 0;
+
+
+ /**
+ * @brief Consume message or get error event from the queue.
+ *
+ * @remark Use \c delete to free the message.
+ *
+ * @returns One of:
+ * - proper message (RdKafka::Message::err() is ERR_NO_ERROR)
+ * - error event (RdKafka::Message::err() is != ERR_NO_ERROR)
+ * - timeout due to no message or event in \p timeout_ms
+ * (RdKafka::Message::err() is ERR__TIMED_OUT)
+ */
+ virtual Message *consume (int timeout_ms) = 0;
+
+ /**
+ * @brief Poll queue, serving any enqueued callbacks.
+ *
+ * @remark Must NOT be used for queues containing messages.
+ *
+ * @returns the number of events served or 0 on timeout.
+ */
+ virtual int poll (int timeout_ms) = 0;
+
+ virtual ~Queue () = 0;
+};
+
+/**@}*/
+
+
+/**
+ * @name KafkaConsumer
+ * @{
+ *
+ */
+
+
+/**
+ * @brief High-level KafkaConsumer (for brokers 0.9 and later)
+ *
+ * @remark Requires Apache Kafka >= 0.9.0 brokers
+ *
+ * Currently supports the \c range and \c roundrobin partition assignment
+ * strategies (see \c partition.assignment.strategy)
+ */
+class RD_EXPORT KafkaConsumer : public virtual Handle {
+public:
+ /**
+ * @brief Creates a KafkaConsumer.
+ *
+ * The \p conf object must have \c group.id set to the consumer group to join.
+ *
+ * Use RdKafka::KafkaConsumer::close() to shut down the consumer.
+ *
+ * @sa RdKafka::RebalanceCb
+ * @sa CONFIGURATION.md for \c group.id, \c session.timeout.ms,
+ * \c partition.assignment.strategy, etc.
+ */
+ static KafkaConsumer *create (Conf *conf, std::string &errstr);
+
+ virtual ~KafkaConsumer () = 0;
+
+
+ /** @brief Returns the current partition assignment as set by
+ * RdKafka::KafkaConsumer::assign() */
+ virtual ErrorCode assignment (std::vector<RdKafka::TopicPartition*> &partitions) = 0;
+
+ /** @brief Returns the current subscription as set by
+ * RdKafka::KafkaConsumer::subscribe() */
+ virtual ErrorCode subscription (std::vector<std::string> &topics) = 0;
+
+ /**
+ * @brief Update the subscription set to \p topics.
+ *
+ * Any previous subscription will be unassigned and unsubscribed first.
+ *
+ * The subscription set denotes the desired topics to consume and this
+ * set is provided to the partition assignor (one of the elected group
+ * members) for all clients which then uses the configured
+ * \c partition.assignment.strategy to assign the subscription sets's
+ * topics's partitions to the consumers, depending on their subscription.
+ *
+ * The result of such an assignment is a rebalancing which is either
+ * handled automatically in librdkafka or can be overriden by the application
+ * by providing a RdKafka::RebalanceCb.
+ *
+ * The rebalancing passes the assigned partition set to
+ * RdKafka::KafkaConsumer::assign() to update what partitions are actually
+ * being fetched by the KafkaConsumer.
+ *
+ * Regex pattern matching automatically performed for topics prefixed
+ * with \c \"^\" (e.g. \c \"^myPfx[0-9]_.*\"
+ *
+ * @returns an error if the provided list of topics is invalid.
+ */
+ virtual ErrorCode subscribe (const std::vector<std::string> &topics) = 0;
+
+ /** @brief Unsubscribe from the current subscription set. */
+ virtual ErrorCode unsubscribe () = 0;
+
+ /**
+ * @brief Update the assignment set to \p partitions.
+ *
+ * The assignment set is the set of partitions actually being consumed
+ * by the KafkaConsumer.
+ */
+ virtual ErrorCode assign (const std::vector<TopicPartition*> &partitions) = 0;
+
+ /**
+ * @brief Stop consumption and remove the current assignment.
+ */
+ virtual ErrorCode unassign () = 0;
+
+ /**
+ * @brief Consume message or get error event, triggers callbacks.
+ *
+ * Will automatically call registered callbacks for any such queued events,
+ * including RdKafka::RebalanceCb, RdKafka::EventCb, RdKafka::OffsetCommitCb,
+ * etc.
+ *
+ * @remark Use \c delete to free the message.
+ *
+ * @remark An application should make sure to call consume() at regular
+ * intervals, even if no messages are expected, to serve any
+ * queued callbacks waiting to be called. This is especially
+ * important when a RebalanceCb has been registered as it needs
+ * to be called and handled properly to synchronize internal
+ * consumer state.
+ *
+ * @remark Application MUST NOT call \p poll() on KafkaConsumer objects.
+ *
+ * @returns One of:
+ * - proper message (RdKafka::Message::err() is ERR_NO_ERROR)
+ * - error event (RdKafka::Message::err() is != ERR_NO_ERROR)
+ * - timeout due to no message or event in \p timeout_ms
+ * (RdKafka::Message::err() is ERR__TIMED_OUT)
+ */
+ virtual Message *consume (int timeout_ms) = 0;
+
+ /**
+ * @brief Commit offsets for the current assignment.
+ *
+ * @remark This is the synchronous variant that blocks until offsets
+ * are committed or the commit fails (see return value).
+ *
+ * @remark If a RdKafka::OffsetCommitCb callback is registered it will
+ * be called with commit details on a future call to
+ * RdKafka::KafkaConsumer::consume()
+
+ *
+ * @returns ERR_NO_ERROR or error code.
+ */
+ virtual ErrorCode commitSync () = 0;
+
+ /**
+ * @brief Asynchronous version of RdKafka::KafkaConsumer::CommitSync()
+ *
+ * @sa RdKafka::KafkaConsummer::commitSync()
+ */
+ virtual ErrorCode commitAsync () = 0;
+
+ /**
+ * @brief Commit offset for a single topic+partition based on \p message
+ *
+ * @remark This is the synchronous variant.
+ *
+ * @sa RdKafka::KafkaConsummer::commitSync()
+ */
+ virtual ErrorCode commitSync (Message *message) = 0;
+
+ /**
+ * @brief Commit offset for a single topic+partition based on \p message
+ *
+ * @remark This is the asynchronous variant.
+ *
+ * @sa RdKafka::KafkaConsummer::commitSync()
+ */
+ virtual ErrorCode commitAsync (Message *message) = 0;
+
+ /**
+ * @brief Commit offsets for the provided list of partitions.
+ *
+ * @remark This is the synchronous variant.
+ */
+ virtual ErrorCode commitSync (std::vector<TopicPartition*> &offsets) = 0;
+
+ /**
+ * @brief Commit offset for the provided list of partitions.
+ *
+ * @remark This is the asynchronous variant.
+ */
+ virtual ErrorCode commitAsync (const std::vector<TopicPartition*> &offsets) = 0;
+
+ /**
+ * @brief Commit offsets for the current assignment.
+ *
+ * @remark This is the synchronous variant that blocks until offsets
+ * are committed or the commit fails (see return value).
+ *
+ * @remark The provided callback will be called from this function.
+ *
+ * @returns ERR_NO_ERROR or error code.
+ */
+ virtual ErrorCode commitSync (OffsetCommitCb *offset_commit_cb) = 0;
+
+ /**
+ * @brief Commit offsets for the provided list of partitions.
+ *
+ * @remark This is the synchronous variant that blocks until offsets
+ * are committed or the commit fails (see return value).
+ *
+ * @remark The provided callback will be called from this function.
+ *
+ * @returns ERR_NO_ERROR or error code.
+ */
+ virtual ErrorCode commitSync (std::vector<TopicPartition*> &offsets,
+ OffsetCommitCb *offset_commit_cb) = 0;
+
+
+
+
+ /**
+ * @brief Retrieve committed offsets for topics+partitions.
+ *
+ * @returns RD_KAFKA_RESP_ERR_NO_ERROR on success in which case the
+ * \p offset or \p err field of each \p partitions' element is filled
+ * in with the stored offset, or a partition specific error.
+ * Else returns an error code.
+ */
+ virtual ErrorCode committed (std::vector<TopicPartition*> &partitions,
+ int timeout_ms) = 0;
+
+ /**
+ * @brief Retrieve current positions (offsets) for topics+partitions.
+ *
+ * @returns RD_KAFKA_RESP_ERR_NO_ERROR on success in which case the
+ * \p offset or \p err field of each \p partitions' element is filled
+ * in with the stored offset, or a partition specific error.
+ * Else returns an error code.
+ */
+ virtual ErrorCode position (std::vector<TopicPartition*> &partitions) = 0;
+
+
+ /**
+ * For pausing and resuming consumption, see
+ * @sa RdKafka::Handle::pause() and RdKafka::Handle::resume()
+ */
+
+
+ /**
+ * @brief Close and shut down the proper.
+ *
+ * This call will block until the following operations are finished:
+ * - Trigger a local rebalance to void the current assignment
+ * - Stop consumption for current assignment
+ * - Commit offsets
+ * - Leave group
+ *
+ * The maximum blocking time is roughly limited to session.timeout.ms.
+ *
+ * @remark Callbacks, such as RdKafka::RebalanceCb and
+ * RdKafka::OffsetCommitCb, etc, may be called.
+ *
+ * @remark The consumer object must later be freed with \c delete
+ */
+ virtual ErrorCode close () = 0;
+
+
+ /**
+ * @brief Seek consumer for topic+partition to offset which is either an
+ * absolute or logical offset.
+ *
+ * If \p timeout_ms is not 0 the call will wait this long for the
+ * seek to be performed. If the timeout is reached the internal state
+ * will be unknown and this function returns `ERR__TIMED_OUT`.
+ * If \p timeout_ms is 0 it will initiate the seek but return
+ * immediately without any error reporting (e.g., async).
+ *
+ * This call triggers a fetch queue barrier flush.
+ *
+ * @remark Consumtion for the given partition must have started for the
+ * seek to work. Use assign() to set the starting offset.
+ *
+ * @returns an ErrorCode to indicate success or failure.
+ */
+ virtual ErrorCode seek (const TopicPartition &partition, int timeout_ms) = 0;
+
+
+ /**
+ * @brief Store offset \p offset for topic partition \p partition.
+ * The offset will be committed (written) to the offset store according
+ * to \p auto.commit.interval.ms or the next manual offset-less commit*()
+ *
+ * Per-partition success/error status propagated through TopicPartition.err()
+ *
+ * @remark \c enable.auto.offset.store must be set to \c false when using this API.
+ *
+ * @returns RdKafka::ERR_NO_ERROR on success or an error code on error.
+ */
+ virtual ErrorCode offsets_store (std::vector<TopicPartition*> &offsets) = 0;
+};
+
+
+/**@}*/
+
+
+/**
+ * @name Simple Consumer (legacy)
+ * @{
+ *
+ */
+
+/**
+ * @brief Simple Consumer (legacy)
+ *
+ * A simple non-balanced, non-group-aware, consumer.
+ */
+class RD_EXPORT Consumer : public virtual Handle {
+ public:
+ /**
+ * @brief Creates a new Kafka consumer handle.
+ *
+ * \p conf is an optional object that will be used instead of the default
+ * configuration.
+ * The \p conf object is reusable after this call.
+ *
+ * @returns the new handle on success or NULL on error in which case
+ * \p errstr is set to a human readable error message.
+ */
+ static Consumer *create (Conf *conf, std::string &errstr);
+
+ virtual ~Consumer () = 0;
+
+
+ /**
+ * @brief Start consuming messages for topic and \p partition
+ * at offset \p offset which may either be a proper offset (0..N)
+ * or one of the the special offsets: \p OFFSET_BEGINNING or \p OFFSET_END.
+ *
+ * rdkafka will attempt to keep \p queued.min.messages (config property)
+ * messages in the local queue by repeatedly fetching batches of messages
+ * from the broker until the threshold is reached.
+ *
+ * The application shall use one of the \p ..->consume*() functions
+ * to consume messages from the local queue, each kafka message being
+ * represented as a `RdKafka::Message *` object.
+ *
+ * \p ..->start() must not be called multiple times for the same
+ * topic and partition without stopping consumption first with
+ * \p ..->stop().
+ *
+ * @returns an ErrorCode to indicate success or failure.
+ */
+ virtual ErrorCode start (Topic *topic, int32_t partition, int64_t offset) = 0;
+
+ /**
+ * @brief Start consuming messages for topic and \p partition on
+ * queue \p queue.
+ *
+ * @sa RdKafka::Consumer::start()
+ */
+ virtual ErrorCode start (Topic *topic, int32_t partition, int64_t offset,
+ Queue *queue) = 0;
+
+ /**
+ * @brief Stop consuming messages for topic and \p partition, purging
+ * all messages currently in the local queue.
+ *
+ * The application needs to be stop all consumers before destroying
+ * the Consumer handle.
+ *
+ * @returns an ErrorCode to indicate success or failure.
+ */
+ virtual ErrorCode stop (Topic *topic, int32_t partition) = 0;
+
+ /**
+ * @brief Seek consumer for topic+partition to \p offset which is either an
+ * absolute or logical offset.
+ *
+ * If \p timeout_ms is not 0 the call will wait this long for the
+ * seek to be performed. If the timeout is reached the internal state
+ * will be unknown and this function returns `ERR__TIMED_OUT`.
+ * If \p timeout_ms is 0 it will initiate the seek but return
+ * immediately without any error reporting (e.g., async).
+ *
+ * This call triggers a fetch queue barrier flush.
+ *
+ * @returns an ErrorCode to indicate success or failure.
+ */
+ virtual ErrorCode seek (Topic *topic, int32_t partition, int64_t offset,
+ int timeout_ms) = 0;
+
+ /**
+ * @brief Consume a single message from \p topic and \p partition.
+ *
+ * \p timeout_ms is maximum amount of time to wait for a message to be
+ * received.
+ * Consumer must have been previously started with \p ..->start().
+ *
+ * @returns a Message object, the application needs to check if message
+ * is an error or a proper message RdKafka::Message::err() and checking for
+ * \p ERR_NO_ERROR.
+ *
+ * The message object must be destroyed when the application is done with it.
+ *
+ * Errors (in RdKafka::Message::err()):
+ * - ERR__TIMED_OUT - \p timeout_ms was reached with no new messages fetched.
+ * - ERR__PARTITION_EOF - End of partition reached, not an error.
+ */
+ virtual Message *consume (Topic *topic, int32_t partition,
+ int timeout_ms) = 0;
+
+ /**
+ * @brief Consume a single message from the specified queue.
+ *
+ * \p timeout_ms is maximum amount of time to wait for a message to be
+ * received.
+ * Consumer must have been previously started on the queue with
+ * \p ..->start().
+ *
+ * @returns a Message object, the application needs to check if message
+ * is an error or a proper message \p Message->err() and checking for
+ * \p ERR_NO_ERROR.
+ *
+ * The message object must be destroyed when the application is done with it.
+ *
+ * Errors (in RdKafka::Message::err()):
+ * - ERR__TIMED_OUT - \p timeout_ms was reached with no new messages fetched
+ *
+ * Note that Message->topic() may be nullptr after certain kinds of
+ * errors, so applications should check that it isn't null before
+ * dereferencing it.
+ */
+ virtual Message *consume (Queue *queue, int timeout_ms) = 0;
+
+ /**
+ * @brief Consumes messages from \p topic and \p partition, calling
+ * the provided callback for each consumed messsage.
+ *
+ * \p consume_callback() provides higher throughput performance
+ * than \p consume().
+ *
+ * \p timeout_ms is the maximum amount of time to wait for one or
+ * more messages to arrive.
+ *
+ * The provided \p consume_cb instance has its \p consume_cb function
+ * called for every message received.
+ *
+ * The \p opaque argument is passed to the \p consume_cb as \p opaque.
+ *
+ * @returns the number of messages processed or -1 on error.
+ *
+ * @sa RdKafka::Consumer::consume()
+ */
+ virtual int consume_callback (Topic *topic, int32_t partition,
+ int timeout_ms,
+ ConsumeCb *consume_cb,
+ void *opaque) = 0;
+
+ /**
+ * @brief Consumes messages from \p queue, calling the provided callback for
+ * each consumed messsage.
+ *
+ * @sa RdKafka::Consumer::consume_callback()
+ */
+ virtual int consume_callback (Queue *queue, int timeout_ms,
+ RdKafka::ConsumeCb *consume_cb,
+ void *opaque) = 0;
+
+ /**
+ * @brief Converts an offset into the logical offset from the tail of a topic.
+ *
+ * \p offset is the (positive) number of items from the end.
+ *
+ * @returns the logical offset for message \p offset from the tail, this value
+ * may be passed to Consumer::start, et.al.
+ * @remark The returned logical offset is specific to librdkafka.
+ */
+ static int64_t OffsetTail(int64_t offset);
+};
+
+/**@}*/
+
+
+/**
+ * @name Producer
+ * @{
+ *
+ */
+
+
+/**
+ * @brief Producer
+ */
+class RD_EXPORT Producer : public virtual Handle {
+ public:
+ /**
+ * @brief Creates a new Kafka producer handle.
+ *
+ * \p conf is an optional object that will be used instead of the default
+ * configuration.
+ * The \p conf object is reusable after this call.
+ *
+ * @returns the new handle on success or NULL on error in which case
+ * \p errstr is set to a human readable error message.
+ */
+ static Producer *create (Conf *conf, std::string &errstr);
+
+
+ virtual ~Producer () = 0;
+
+ /**
+ * @brief RdKafka::Producer::produce() \p msgflags
+ *
+ * These flags are optional and mutually exclusive.
+ */
+ enum {
+ RK_MSG_FREE = 0x1, /**< rdkafka will free(3) \p payload
+ * when it is done with it. */
+ RK_MSG_COPY = 0x2, /**< the \p payload data will be copied
+ * and the \p payload pointer will not
+ * be used by rdkafka after the
+ * call returns. */
+ RK_MSG_BLOCK = 0x4 /**< Block produce*() on message queue
+ * full.
+ * WARNING:
+ * If a delivery report callback
+ * is used the application MUST
+ * call rd_kafka_poll() (or equiv.)
+ * to make sure delivered messages
+ * are drained from the internal
+ * delivery report queue.
+ * Failure to do so will result
+ * in indefinately blocking on
+ * the produce() call when the
+ * message queue is full.
+ */
+
+
+ /**@cond NO_DOC*/
+ /* For backwards compatibility: */
+#ifndef MSG_COPY /* defined in sys/msg.h */
+ , /** this comma must exist betwen
+ * RK_MSG_BLOCK and MSG_FREE
+ */
+ MSG_FREE = RK_MSG_FREE,
+ MSG_COPY = RK_MSG_COPY
+#endif
+ /**@endcond*/
+ };
+
+ /**
+ * @brief Produce and send a single message to broker.
+ *
+ * This is an asynch non-blocking API.
+ *
+ * \p partition is the target partition, either:
+ * - RdKafka::Topic::PARTITION_UA (unassigned) for
+ * automatic partitioning using the topic's partitioner function, or
+ * - a fixed partition (0..N)
+ *
+ * \p msgflags is zero or more of the following flags OR:ed together:
+ * RK_MSG_BLOCK - block \p produce*() call if
+ * \p queue.buffering.max.messages or
+ * \p queue.buffering.max.kbytes are exceeded.
+ * Messages are considered in-queue from the point they
+ * are accepted by produce() until their corresponding
+ * delivery report callback/event returns.
+ * It is thus a requirement to call
+ * poll() (or equiv.) from a separate
+ * thread when RK_MSG_BLOCK is used.
+ * See WARNING on \c RK_MSG_BLOCK above.
+ * RK_MSG_FREE - rdkafka will free(3) \p payload when it is done with it.
+ * RK_MSG_COPY - the \p payload data will be copied and the \p payload
+ * pointer will not be used by rdkafka after the
+ * call returns.
+ *
+ * NOTE: RK_MSG_FREE and RK_MSG_COPY are mutually exclusive.
+ *
+ * If the function returns -1 and RK_MSG_FREE was specified, then
+ * the memory associated with the payload is still the caller's
+ * responsibility.
+ *
+ * \p payload is the message payload of size \p len bytes.
+ *
+ * \p key is an optional message key, if non-NULL it
+ * will be passed to the topic partitioner as well as be sent with the
+ * message to the broker and passed on to the consumer.
+ *
+ * \p msg_opaque is an optional application-provided per-message opaque
+ * pointer that will provided in the delivery report callback (\p dr_cb) for
+ * referencing this message.
+ *
+ * @returns an ErrorCode to indicate success or failure:
+ * - ERR__QUEUE_FULL - maximum number of outstanding messages has been
+ * reached: \c queue.buffering.max.message
+ *
+ * - ERR_MSG_SIZE_TOO_LARGE - message is larger than configured max size:
+ * \c messages.max.bytes
+ *
+ * - ERR__UNKNOWN_PARTITION - requested \p partition is unknown in the
+ * Kafka cluster.
+ *
+ * - ERR__UNKNOWN_TOPIC - topic is unknown in the Kafka cluster.
+ */
+ virtual ErrorCode produce (Topic *topic, int32_t partition,
+ int msgflags,
+ void *payload, size_t len,
+ const std::string *key,
+ void *msg_opaque) = 0;
+
+ /**
+ * @brief Variant produce() that passes the key as a pointer and length
+ * instead of as a const std::string *.
+ */
+ 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;
+
+ /**
+ * @brief produce() variant that takes topic as a string (no need for
+ * creating a Topic object), and also allows providing the
+ * message timestamp (microseconds since beginning of epoch, UTC).
+ * Otherwise identical to produce() above.
+ */
+ 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;
+
+
+ /**
+ * @brief Variant produce() that accepts vectors for key and payload.
+ * The vector data will be copied.
+ */
+ virtual ErrorCode produce (Topic *topic, int32_t partition,
+ const std::vector<char> *payload,
+ const std::vector<char> *key,
+ void *msg_opaque) = 0;
+
+
+ /**
+ * @brief Wait until all outstanding produce requests, et.al, are completed.
+ * This should typically be done prior to destroying a producer instance
+ * to make sure all queued and in-flight produce requests are completed
+ * before terminating.
+ *
+ * @remark This function will call poll() and thus trigger callbacks.
+ *
+ * @returns ERR__TIMED_OUT if \p timeout_ms was reached before all
+ * outstanding requests were completed, else ERR_NO_ERROR
+ */
+ virtual ErrorCode flush (int timeout_ms) = 0;
+};
+
+/**@}*/
+
+
+/**
+ * @name Metadata interface
+ * @{
+ *
+ */
+
+
+/**
+ * @brief Metadata: Broker information
+ */
+class BrokerMetadata {
+ public:
+ /** @returns Broker id */
+ virtual int32_t id() const = 0;
+
+ /** @returns Broker hostname */
+ virtual const std::string host() const = 0;
+
+ /** @returns Broker listening port */
+ virtual int port() const = 0;
+
+ virtual ~BrokerMetadata() = 0;
+};
+
+
+
+/**
+ * @brief Metadata: Partition information
+ */
+class PartitionMetadata {
+ public:
+ /** @brief Replicas */
+ typedef std::vector<int32_t> ReplicasVector;
+ /** @brief ISRs (In-Sync-Replicas) */
+ typedef std::vector<int32_t> ISRSVector;
+
+ /** @brief Replicas iterator */
+ typedef ReplicasVector::const_iterator ReplicasIterator;
+ /** @brief ISRs iterator */
+ typedef ISRSVector::const_iterator ISRSIterator;
+
+
+ /** @returns Partition id */
+ virtual int32_t id() const = 0;
+
+ /** @returns Partition error reported by broker */
+ virtual ErrorCode err() const = 0;
+
+ /** @returns Leader broker (id) for partition */
+ virtual int32_t leader() const = 0;
+
+ /** @returns Replica brokers */
+ virtual const std::vector<int32_t> *replicas() const = 0;
+
+ /** @returns In-Sync-Replica brokers
+ * @warning The broker may return a cached/outdated list of ISRs.
+ */
+ virtual const std::vector<int32_t> *isrs() const = 0;
+
+ virtual ~PartitionMetadata() = 0;
+};
+
+
+
+/**
+ * @brief Metadata: Topic information
+ */
+class TopicMetadata {
+ public:
+ /** @brief Partitions */
+ typedef std::vector<const PartitionMetadata*> PartitionMetadataVector;
+ /** @brief Partitions iterator */
+ typedef PartitionMetadataVector::const_iterator PartitionMetadataIterator;
+
+ /** @returns Topic name */
+ virtual const std::string topic() const = 0;
+
+ /** @returns Partition list */
+ virtual const PartitionMetadataVector *partitions() const = 0;
+
+ /** @returns Topic error reported by broker */
+ virtual ErrorCode err() const = 0;
+
+ virtual ~TopicMetadata() = 0;
+};
+
+
+/**
+ * @brief Metadata container
+ */
+class Metadata {
+ public:
+ /** @brief Brokers */
+ typedef std::vector<const BrokerMetadata*> BrokerMetadataVector;
+ /** @brief Topics */
+ typedef std::vector<const TopicMetadata*> TopicMetadataVector;
+
+ /** @brief Brokers iterator */
+ typedef BrokerMetadataVector::const_iterator BrokerMetadataIterator;
+ /** @brief Topics iterator */
+ typedef TopicMetadataVector::const_iterator TopicMetadataIterator;
+
+
+ /** @brief Broker list */
+ virtual const BrokerMetadataVector *brokers() const = 0;
+
+ /** @brief Topic list */
+ virtual const TopicMetadataVector *topics() const = 0;
+
+ /** @brief Broker (id) originating this metadata */
+ virtual int32_t orig_broker_id() const = 0;
+
+ /** @brief Broker (name) originating this metadata */
+ virtual const std::string orig_broker_name() const = 0;
+
+ virtual ~Metadata() = 0;
+};
+
+/**@}*/
+
+}
+