rdkafkacpp.h

Go to the documentation of this file.

/*
 * 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.
 */

#ifndef _RDKAFKACPP_H_
#define _RDKAFKACPP_H_

#include <string>
#include <list>
#include <vector>
#include <cstdlib>
#include <cstring>
#include <stdint.h>
#include <sys/types.h>

#ifdef _WIN32
#ifndef ssize_t
#ifndef _BASETSD_H_
#include <basetsd.h>
#endif
#ifndef _SSIZE_T_DEFINED
#define _SSIZE_T_DEFINED
typedef SSIZE_T ssize_t;
#endif
#endif
#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

extern "C" {
        /* Forward declarations */
        struct rd_kafka_s;
        struct rd_kafka_topic_s;
        struct rd_kafka_message_s;
        struct rd_kafka_conf_s;
        struct rd_kafka_topic_conf_s;
}

namespace RdKafka {

#define RD_KAFKA_VERSION  0x010502ff

RD_EXPORT
int          version ();

RD_EXPORT
std::string  version_str();

RD_EXPORT
std::string get_debug_contexts();

RD_EXPORT
int          wait_destroyed(int timeout_ms);


enum ErrorCode {
    /* Internal errors to rdkafka: */
    ERR__BEGIN = -200,
    ERR__BAD_MSG = -199,
    ERR__BAD_COMPRESSION = -198,
    ERR__DESTROY = -197,
    ERR__FAIL = -196,
    ERR__TRANSPORT = -195,
    ERR__CRIT_SYS_RESOURCE = -194,
    ERR__RESOLVE = -193,
    ERR__MSG_TIMED_OUT = -192,
    ERR__PARTITION_EOF = -191,
    ERR__UNKNOWN_PARTITION = -190,
    ERR__FS = -189,
    ERR__UNKNOWN_TOPIC = -188,
    ERR__ALL_BROKERS_DOWN = -187,
    ERR__INVALID_ARG = -186,
    ERR__TIMED_OUT = -185,
    ERR__QUEUE_FULL = -184,
        ERR__ISR_INSUFF = -183,
        ERR__NODE_UPDATE = -182,
    ERR__SSL = -181,
        ERR__WAIT_COORD = -180,
        ERR__UNKNOWN_GROUP = -179,
        ERR__IN_PROGRESS = -178,
        ERR__PREV_IN_PROGRESS = -177,
        ERR__EXISTING_SUBSCRIPTION = -176,
        ERR__ASSIGN_PARTITIONS = -175,
        ERR__REVOKE_PARTITIONS = -174,
        ERR__CONFLICT = -173,
        ERR__STATE = -172,
        ERR__UNKNOWN_PROTOCOL = -171,
        ERR__NOT_IMPLEMENTED = -170,
    ERR__AUTHENTICATION = -169,
    ERR__NO_OFFSET = -168,
    ERR__OUTDATED = -167,
    ERR__TIMED_OUT_QUEUE = -166,
        ERR__UNSUPPORTED_FEATURE = -165,
        ERR__WAIT_CACHE = -164,
        ERR__INTR = -163,
        ERR__KEY_SERIALIZATION = -162,
        ERR__VALUE_SERIALIZATION = -161,
        ERR__KEY_DESERIALIZATION = -160,
        ERR__VALUE_DESERIALIZATION = -159,
        ERR__PARTIAL = -158,
        ERR__READ_ONLY = -157,
        ERR__NOENT = -156,
        ERR__UNDERFLOW = -155,
        ERR__INVALID_TYPE = -154,
        ERR__RETRY = -153,
        ERR__PURGE_QUEUE = -152,
        ERR__PURGE_INFLIGHT = -151,
        ERR__FATAL = -150,
        ERR__INCONSISTENT = -149,
        ERR__GAPLESS_GUARANTEE = -148,
        ERR__MAX_POLL_EXCEEDED = -147,
        ERR__UNKNOWN_BROKER = -146,
        ERR__NOT_CONFIGURED = -145,
        ERR__FENCED = -144,
        ERR__APPLICATION = -143,

    ERR__END = -100,

    /* Kafka broker errors: */
    ERR_UNKNOWN = -1,
    ERR_NO_ERROR = 0,
    ERR_OFFSET_OUT_OF_RANGE = 1,
    ERR_INVALID_MSG = 2,
    ERR_UNKNOWN_TOPIC_OR_PART = 3,
    ERR_INVALID_MSG_SIZE = 4,
    ERR_LEADER_NOT_AVAILABLE = 5,
    ERR_NOT_LEADER_FOR_PARTITION = 6,
    ERR_REQUEST_TIMED_OUT = 7,
    ERR_BROKER_NOT_AVAILABLE = 8,
    ERR_REPLICA_NOT_AVAILABLE = 9,
    ERR_MSG_SIZE_TOO_LARGE = 10,
    ERR_STALE_CTRL_EPOCH = 11,
    ERR_OFFSET_METADATA_TOO_LARGE = 12,
    ERR_NETWORK_EXCEPTION = 13,
        ERR_COORDINATOR_LOAD_IN_PROGRESS = 14,
#define ERR_GROUP_LOAD_IN_PROGRESS           ERR_COORDINATOR_LOAD_IN_PROGRESS

        ERR_COORDINATOR_NOT_AVAILABLE = 15,
#define ERR_GROUP_COORDINATOR_NOT_AVAILABLE  ERR_COORDINATOR_NOT_AVAILABLE

        ERR_NOT_COORDINATOR = 16,
#define ERR_NOT_COORDINATOR_FOR_GROUP        ERR_NOT_COORDINATOR

        ERR_TOPIC_EXCEPTION = 17,
        ERR_RECORD_LIST_TOO_LARGE = 18,
        ERR_NOT_ENOUGH_REPLICAS = 19,
        ERR_NOT_ENOUGH_REPLICAS_AFTER_APPEND = 20,
        ERR_INVALID_REQUIRED_ACKS = 21,
        ERR_ILLEGAL_GENERATION = 22,
        ERR_INCONSISTENT_GROUP_PROTOCOL = 23,
    ERR_INVALID_GROUP_ID = 24,
        ERR_UNKNOWN_MEMBER_ID = 25,
        ERR_INVALID_SESSION_TIMEOUT = 26,
    ERR_REBALANCE_IN_PROGRESS = 27,
        ERR_INVALID_COMMIT_OFFSET_SIZE = 28,
        ERR_TOPIC_AUTHORIZATION_FAILED = 29,
    ERR_GROUP_AUTHORIZATION_FAILED = 30,
    ERR_CLUSTER_AUTHORIZATION_FAILED = 31,
        ERR_INVALID_TIMESTAMP = 32,
        ERR_UNSUPPORTED_SASL_MECHANISM = 33,
        ERR_ILLEGAL_SASL_STATE = 34,
        ERR_UNSUPPORTED_VERSION = 35,
        ERR_TOPIC_ALREADY_EXISTS = 36,
        ERR_INVALID_PARTITIONS = 37,
        ERR_INVALID_REPLICATION_FACTOR = 38,
        ERR_INVALID_REPLICA_ASSIGNMENT = 39,
        ERR_INVALID_CONFIG = 40,
        ERR_NOT_CONTROLLER = 41,
        ERR_INVALID_REQUEST = 42,
        ERR_UNSUPPORTED_FOR_MESSAGE_FORMAT = 43,
        ERR_POLICY_VIOLATION = 44,
        ERR_OUT_OF_ORDER_SEQUENCE_NUMBER = 45,
        ERR_DUPLICATE_SEQUENCE_NUMBER = 46,
        ERR_INVALID_PRODUCER_EPOCH = 47,
        ERR_INVALID_TXN_STATE = 48,
        ERR_INVALID_PRODUCER_ID_MAPPING = 49,
        ERR_INVALID_TRANSACTION_TIMEOUT = 50,
        ERR_CONCURRENT_TRANSACTIONS = 51,
        ERR_TRANSACTION_COORDINATOR_FENCED = 52,
        ERR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED = 53,
        ERR_SECURITY_DISABLED = 54,
        ERR_OPERATION_NOT_ATTEMPTED = 55,
        ERR_KAFKA_STORAGE_ERROR = 56,
        ERR_LOG_DIR_NOT_FOUND = 57,
        ERR_SASL_AUTHENTICATION_FAILED = 58,
        ERR_UNKNOWN_PRODUCER_ID = 59,
        ERR_REASSIGNMENT_IN_PROGRESS = 60,
        ERR_DELEGATION_TOKEN_AUTH_DISABLED = 61,
        ERR_DELEGATION_TOKEN_NOT_FOUND = 62,
        ERR_DELEGATION_TOKEN_OWNER_MISMATCH = 63,
        ERR_DELEGATION_TOKEN_REQUEST_NOT_ALLOWED = 64,
        ERR_DELEGATION_TOKEN_AUTHORIZATION_FAILED = 65,
        ERR_DELEGATION_TOKEN_EXPIRED = 66,
        ERR_INVALID_PRINCIPAL_TYPE = 67,
        ERR_NON_EMPTY_GROUP = 68,
        ERR_GROUP_ID_NOT_FOUND = 69,
        ERR_FETCH_SESSION_ID_NOT_FOUND = 70,
        ERR_INVALID_FETCH_SESSION_EPOCH = 71,
        ERR_LISTENER_NOT_FOUND = 72,
        ERR_TOPIC_DELETION_DISABLED = 73,
        ERR_FENCED_LEADER_EPOCH = 74,
        ERR_UNKNOWN_LEADER_EPOCH = 75,
        ERR_UNSUPPORTED_COMPRESSION_TYPE = 76,
        ERR_STALE_BROKER_EPOCH = 77,
        ERR_OFFSET_NOT_AVAILABLE = 78,
        ERR_MEMBER_ID_REQUIRED = 79,
        ERR_PREFERRED_LEADER_NOT_AVAILABLE = 80,
        ERR_GROUP_MAX_SIZE_REACHED = 81,
        ERR_FENCED_INSTANCE_ID = 82,
        ERR_ELIGIBLE_LEADERS_NOT_AVAILABLE = 83,
        ERR_ELECTION_NOT_NEEDED = 84,
        ERR_NO_REASSIGNMENT_IN_PROGRESS = 85,
        ERR_GROUP_SUBSCRIBED_TO_TOPIC = 86,
        ERR_INVALID_RECORD = 87,
        ERR_UNSTABLE_OFFSET_COMMIT = 88
};


RD_EXPORT
std::string  err2str(RdKafka::ErrorCode err);



enum CertificateType {
  CERT_PUBLIC_KEY,   
  CERT_PRIVATE_KEY,  
  CERT_CA,           
  CERT__CNT
};

enum CertificateEncoding {
  CERT_ENC_PKCS12,  
  CERT_ENC_DER,     
  CERT_ENC_PEM,     
  CERT_ENC__CNT
};

/* Forward declarations */
class Producer;
class Message;
class Headers;
class Queue;
class Event;
class Topic;
class TopicPartition;
class Metadata;
class KafkaConsumer;
class RD_EXPORT Error {
 public:

 static Error *create (ErrorCode code, const std::string *errstr);

 virtual ~Error () { }

 /*
  * Error accessor methods
  */

 virtual ErrorCode code () const = 0;

 virtual std::string name () const = 0;

 virtual std::string str () const = 0;

 virtual bool is_fatal () const = 0;

 virtual bool is_retriable () const = 0;

 virtual bool txn_requires_abort () const = 0;
};

class RD_EXPORT DeliveryReportCb {
 public:
  virtual void dr_cb (Message &message) = 0;

  virtual ~DeliveryReportCb() { }
};


class RD_EXPORT OAuthBearerTokenRefreshCb {
 public:
  virtual void oauthbearer_token_refresh_cb (const std::string &oauthbearer_config) = 0;

  virtual ~OAuthBearerTokenRefreshCb() { }
};


class RD_EXPORT PartitionerCb {
 public:
  virtual int32_t partitioner_cb (const Topic *topic,
                                  const std::string *key,
                                  int32_t partition_cnt,
                                  void *msg_opaque) = 0;

  virtual ~PartitionerCb() { }
};

class PartitionerKeyPointerCb {
 public:
  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() { }
};



class RD_EXPORT EventCb {
 public:
  virtual void event_cb (Event &event) = 0;

  virtual ~EventCb() { }
};


class RD_EXPORT Event {
 public:
  enum Type {
    EVENT_ERROR,     
    EVENT_STATS,     
    EVENT_LOG,       
    EVENT_THROTTLE   
  };

  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
   */

  virtual Type        type () const = 0;

  virtual ErrorCode   err () const = 0;

  virtual Severity    severity () const = 0;

  virtual std::string fac () const = 0;

  virtual std::string str () const = 0;

  virtual int         throttle_time () const = 0;

  virtual std::string broker_name () const = 0;

  virtual int         broker_id () const = 0;


  virtual bool        fatal () const = 0;
};



class RD_EXPORT ConsumeCb {
 public:
  virtual void consume_cb (Message &message, void *opaque) = 0;

  virtual ~ConsumeCb() { }
};


class RD_EXPORT RebalanceCb {
public:
 virtual void rebalance_cb (RdKafka::KafkaConsumer *consumer,
                RdKafka::ErrorCode err,
                            std::vector<TopicPartition*>&partitions) = 0;

 virtual ~RebalanceCb() { }
};


class RD_EXPORT OffsetCommitCb {
public:
  virtual void offset_commit_cb(RdKafka::ErrorCode err,
                                std::vector<TopicPartition*>&offsets) = 0;

  virtual ~OffsetCommitCb() { }
};



class RD_EXPORT SslCertificateVerifyCb {
public:
  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;

  virtual ~SslCertificateVerifyCb() {}
};


class RD_EXPORT SocketCb {
 public:
  virtual int socket_cb (int domain, int type, int protocol) = 0;

  virtual ~SocketCb() { }
};


class RD_EXPORT OpenCb {
 public:
  virtual int open_cb (const std::string &path, int flags, int mode) = 0;

  virtual ~OpenCb() { }
};


class RD_EXPORT Conf {
 public:
  enum ConfType {
    CONF_GLOBAL, 
    CONF_TOPIC   
  };

  enum ConfResult {
    CONF_UNKNOWN = -2,  
    CONF_INVALID = -1,  
    CONF_OK = 0         
  };


  static Conf *create (ConfType type);

  virtual ~Conf () { }

  virtual Conf::ConfResult set (const std::string &name,
                                const std::string &value,
                                std::string &errstr) = 0;

  virtual Conf::ConfResult set (const std::string &name,
                                DeliveryReportCb *dr_cb,
                                std::string &errstr) = 0;

  virtual Conf::ConfResult set (const std::string &name,
                        OAuthBearerTokenRefreshCb *oauthbearer_token_refresh_cb,
                        std::string &errstr) = 0;

  virtual Conf::ConfResult set (const std::string &name,
                                EventCb *event_cb,
                                std::string &errstr) = 0;

  virtual Conf::ConfResult set (const std::string &name,
                                const Conf *topic_conf,
                                std::string &errstr) = 0;

  virtual Conf::ConfResult set (const std::string &name,
                                PartitionerCb *partitioner_cb,
                                std::string &errstr) = 0;

  virtual Conf::ConfResult set (const std::string &name,
                                PartitionerKeyPointerCb *partitioner_kp_cb,
                                std::string &errstr) = 0;

  virtual Conf::ConfResult set (const std::string &name, SocketCb *socket_cb,
                                std::string &errstr) = 0;

  virtual Conf::ConfResult set (const std::string &name, OpenCb *open_cb,
                                std::string &errstr) = 0;

  virtual Conf::ConfResult set (const std::string &name,
                                RebalanceCb *rebalance_cb,
                                std::string &errstr) = 0;

  virtual Conf::ConfResult set (const std::string &name,
                                OffsetCommitCb *offset_commit_cb,
                                std::string &errstr) = 0;

  virtual Conf::ConfResult set(const std::string &name,
                               SslCertificateVerifyCb *ssl_cert_verify_cb,
                               std::string &errstr) = 0;

  virtual Conf::ConfResult set_ssl_cert (RdKafka::CertificateType cert_type,
                                         RdKafka::CertificateEncoding cert_enc,
                                         const void *buffer, size_t size,
                                         std::string &errstr) = 0;

  virtual Conf::ConfResult get(const std::string &name,
      std::string &value) const = 0;

  virtual Conf::ConfResult get(DeliveryReportCb *&dr_cb) const = 0;

  virtual Conf::ConfResult get(
          OAuthBearerTokenRefreshCb *&oauthbearer_token_refresh_cb) const = 0;

  virtual Conf::ConfResult get(EventCb *&event_cb) const = 0;

  virtual Conf::ConfResult get(PartitionerCb *&partitioner_cb) const = 0;

  virtual Conf::ConfResult get(PartitionerKeyPointerCb *&partitioner_kp_cb) const = 0;

  virtual Conf::ConfResult get(SocketCb *&socket_cb) const = 0;

  virtual Conf::ConfResult get(OpenCb *&open_cb) const = 0;

  virtual Conf::ConfResult get(RebalanceCb *&rebalance_cb) const = 0;

  virtual Conf::ConfResult get(OffsetCommitCb *&offset_commit_cb) const = 0;

  virtual Conf::ConfResult get(SslCertificateVerifyCb *&ssl_cert_verify_cb) const = 0;

  virtual std::list<std::string> *dump () = 0;

  virtual Conf::ConfResult set (const std::string &name, ConsumeCb *consume_cb,
                                std::string &errstr) = 0;

  virtual struct rd_kafka_conf_s *c_ptr_global () = 0;

  virtual struct rd_kafka_topic_conf_s *c_ptr_topic () = 0;
};

class RD_EXPORT Handle {
 public:
  virtual ~Handle() { }

  virtual const std::string name () const = 0;

  virtual const std::string memberid () const = 0;


  virtual int poll (int timeout_ms) = 0;

  virtual int outq_len () = 0;

  virtual ErrorCode metadata (bool all_topics, const Topic *only_rkt,
                              Metadata **metadatap, int timeout_ms) = 0;


  virtual ErrorCode pause (std::vector<TopicPartition*> &partitions) = 0;


  virtual ErrorCode resume (std::vector<TopicPartition*> &partitions) = 0;


  virtual ErrorCode query_watermark_offsets (const std::string &topic,
                         int32_t partition,
                         int64_t *low, int64_t *high,
                         int timeout_ms) = 0;

  virtual ErrorCode get_watermark_offsets (const std::string &topic,
                       int32_t partition,
                       int64_t *low, int64_t *high) = 0;


  virtual ErrorCode offsetsForTimes (std::vector<TopicPartition*> &offsets,
                                     int timeout_ms) = 0;


  virtual Queue *get_partition_queue (const TopicPartition *partition) = 0;

  virtual ErrorCode set_log_queue (Queue *queue) = 0;

  virtual void yield () = 0;

  virtual const std::string clusterid (int timeout_ms) = 0;

  virtual struct rd_kafka_s *c_ptr () = 0;

  virtual int32_t controllerid (int timeout_ms) = 0;


  virtual ErrorCode fatal_error (std::string &errstr) const = 0;

  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;

    virtual ErrorCode oauthbearer_set_token_failure (const std::string &errstr) = 0;
};


class RD_EXPORT TopicPartition {
public:
  static TopicPartition *create (const std::string &topic, int partition);

  static TopicPartition *create (const std::string &topic, int partition,
                                 int64_t offset);

  virtual ~TopicPartition() = 0;

  static void destroy (std::vector<TopicPartition*> &partitions);

  virtual const std::string &topic () const = 0;

  virtual int partition () const = 0;

  virtual int64_t offset () const = 0;

  virtual void set_offset (int64_t offset) = 0;

  virtual ErrorCode err () const = 0;
};



class RD_EXPORT Topic {
 public:
  static const int32_t PARTITION_UA;

  static const int64_t OFFSET_BEGINNING; 
  static const int64_t OFFSET_END; 
  static const int64_t OFFSET_STORED; 
  static const int64_t OFFSET_INVALID; 
  static Topic *create (Handle *base, const std::string &topic_str,
                        const Conf *conf, std::string &errstr);

  virtual ~Topic () = 0;


  virtual const std::string name () const = 0;

  virtual bool partition_available (int32_t partition) const = 0;

  virtual ErrorCode offset_store (int32_t partition, int64_t offset) = 0;

  virtual struct rd_kafka_topic_s *c_ptr () = 0;
};


class RD_EXPORT MessageTimestamp {
public:
  enum MessageTimestampType {
    MSG_TIMESTAMP_NOT_AVAILABLE,   
    MSG_TIMESTAMP_CREATE_TIME,     
    MSG_TIMESTAMP_LOG_APPEND_TIME  
  };

  MessageTimestampType type;       
  int64_t timestamp;               
};


class RD_EXPORT Headers {
public:
  virtual ~Headers() = 0;

  class Header {
   public:
    Header(const std::string &key,
           const void *value,
           size_t value_size):
    key_(key), err_(ERR_NO_ERROR), value_size_(value_size) {
      value_ = copy_value(value, value_size);
    }

    Header(const std::string &key,
           const void *value,
           size_t value_size,
           const RdKafka::ErrorCode err):
    key_(key), err_(err), value_(NULL), value_size_(value_size) {
      if (err == ERR_NO_ERROR)
        value_ = copy_value(value, value_size);
    }

    Header(const Header &other):
    key_(other.key_), err_(other.err_), value_size_(other.value_size_) {
      value_ = copy_value(other.value_, value_size_);
    }

    Header& operator=(const Header &other)
    {
      if (&other == this) {
        return *this;
      }

      key_ = other.key_;
      err_ = other.err_;
      value_size_ = other.value_size_;

      if (value_ != NULL)
        free(value_);

      value_ = copy_value(other.value_, value_size_);

      return *this;
    }

    ~Header() {
      if (value_ != NULL)
        free(value_);
    }

    std::string key() const {
      return key_;
    }

    const void *value() const {
      return value_;
    }

    const char *value_string() const {
      return static_cast<const char *>(value_);
    }

    size_t value_size() const {
      return value_size_;
    }

    RdKafka::ErrorCode err() const {
      return err_;
    }

 private:
    char *copy_value(const void *value, size_t value_size) {
      if (!value)
        return NULL;

      char *dest = (char *)malloc(value_size + 1);
      memcpy(dest, (const char *)value, value_size);
      dest[value_size] = '\0';

      return dest;
    }

    std::string key_;
    RdKafka::ErrorCode err_;
    char *value_;
    size_t value_size_;
    void *operator new(size_t); /* Prevent dynamic allocation */
  };

  static Headers *create();

  static Headers *create(const std::vector<Header> &headers);

  virtual ErrorCode add(const std::string &key, const void *value,
                        size_t value_size) = 0;

  virtual ErrorCode add(const std::string &key, const std::string &value) = 0;

  virtual ErrorCode add(const Header &header) = 0;

  virtual ErrorCode remove(const std::string &key) = 0;

  virtual std::vector<Header> get(const std::string &key) const = 0;

  virtual Header get_last(const std::string &key) const = 0;

  virtual std::vector<Header> get_all() const = 0;

  virtual size_t size() const = 0;
};


class RD_EXPORT Message {
 public:
  enum Status {
    MSG_STATUS_NOT_PERSISTED = 0,

    MSG_STATUS_POSSIBLY_PERSISTED = 1,

    MSG_STATUS_PERSISTED = 2,
  };

  virtual std::string         errstr() const = 0;

  virtual ErrorCode           err () const = 0;

  virtual Topic              *topic () const = 0;

  virtual std::string         topic_name () const = 0;

  virtual int32_t             partition () const = 0;

  virtual void               *payload () const = 0 ;

  virtual size_t              len () const = 0;

  virtual const std::string  *key () const = 0;

  virtual const void         *key_pointer () const = 0 ;

  virtual size_t              key_len () const = 0;

  virtual int64_t             offset () const = 0;

  virtual MessageTimestamp    timestamp () const = 0;

  virtual void               *msg_opaque () const = 0;

  virtual ~Message () = 0;

  virtual int64_t             latency () const = 0;

  virtual struct rd_kafka_message_s *c_ptr () = 0;

  virtual Status status () const = 0;

  virtual RdKafka::Headers   *headers () = 0;

  virtual RdKafka::Headers   *headers (RdKafka::ErrorCode *err) = 0;

  virtual int32_t broker_id () const = 0;
};

class RD_EXPORT Queue {
 public:
  static Queue *create (Handle *handle);

  virtual ErrorCode forward (Queue *dst) = 0;


  virtual Message *consume (int timeout_ms) = 0;

  virtual int poll (int timeout_ms) = 0;

  virtual ~Queue () = 0;

  virtual void io_event_enable (int fd, const void *payload, size_t size) = 0;
};

class RD_EXPORT ConsumerGroupMetadata {
public:
  virtual ~ConsumerGroupMetadata () = 0;
};

class RD_EXPORT KafkaConsumer : public virtual Handle {
public:
  static KafkaConsumer *create (const Conf *conf, std::string &errstr);

  virtual ~KafkaConsumer () = 0;


  virtual ErrorCode assignment (std::vector<RdKafka::TopicPartition*> &partitions) = 0;

  virtual ErrorCode subscription (std::vector<std::string> &topics) = 0;

  virtual ErrorCode subscribe (const std::vector<std::string> &topics) = 0;

  virtual ErrorCode unsubscribe () = 0;

  virtual ErrorCode assign (const std::vector<TopicPartition*> &partitions) = 0;

  virtual ErrorCode unassign () = 0;

  virtual Message *consume (int timeout_ms) = 0;

  virtual ErrorCode commitSync () = 0;

  virtual ErrorCode commitAsync () = 0;

  virtual ErrorCode commitSync (Message *message) = 0;

  virtual ErrorCode commitAsync (Message *message) = 0;

  virtual ErrorCode commitSync (std::vector<TopicPartition*> &offsets) = 0;

  virtual ErrorCode commitAsync (const std::vector<TopicPartition*> &offsets) = 0;

  virtual ErrorCode commitSync (OffsetCommitCb *offset_commit_cb) = 0;

  virtual ErrorCode commitSync (std::vector<TopicPartition*> &offsets,
                                OffsetCommitCb *offset_commit_cb) = 0;




  virtual ErrorCode committed (std::vector<TopicPartition*> &partitions,
                   int timeout_ms) = 0;

  virtual ErrorCode position (std::vector<TopicPartition*> &partitions) = 0;


  virtual ErrorCode close () = 0;


  virtual ErrorCode seek (const TopicPartition &partition, int timeout_ms) = 0;


  virtual ErrorCode offsets_store (std::vector<TopicPartition*> &offsets) = 0;


  virtual ConsumerGroupMetadata *groupMetadata () = 0;


};


class RD_EXPORT Consumer : public virtual Handle {
 public:
  static Consumer *create (const Conf *conf, std::string &errstr);

  virtual ~Consumer () = 0;


  virtual ErrorCode start (Topic *topic, int32_t partition, int64_t offset) = 0;

  virtual ErrorCode start (Topic *topic, int32_t partition, int64_t offset,
                           Queue *queue) = 0;

  virtual ErrorCode stop (Topic *topic, int32_t partition) = 0;

  virtual ErrorCode seek (Topic *topic, int32_t partition, int64_t offset,
              int timeout_ms) = 0;

  virtual Message *consume (Topic *topic, int32_t partition,
                            int timeout_ms) = 0;

  virtual Message *consume (Queue *queue, int timeout_ms) = 0;

  virtual int consume_callback (Topic *topic, int32_t partition,
                                int timeout_ms,
                                ConsumeCb *consume_cb,
                                void *opaque) = 0;

  virtual int consume_callback (Queue *queue, int timeout_ms,
                                RdKafka::ConsumeCb *consume_cb,
                                void *opaque) = 0;

  static int64_t OffsetTail(int64_t offset);
};

class RD_EXPORT Producer : public virtual Handle {
 public:
  static Producer *create (const Conf *conf, std::string &errstr);


  virtual ~Producer () = 0;

  enum {
    RK_MSG_FREE = 0x1, 
    RK_MSG_COPY = 0x2, 
    RK_MSG_BLOCK = 0x4  
  /* For backwards compatibility: */
#ifndef MSG_COPY /* defined in sys/msg.h */
    , 
    MSG_FREE = RK_MSG_FREE,
    MSG_COPY = RK_MSG_COPY
#endif

  };

  virtual ErrorCode produce (Topic *topic, int32_t partition,
                             int msgflags,
                             void *payload, size_t len,
                             const std::string *key,
                             void *msg_opaque) = 0;

  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;

  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;

  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;


  virtual ErrorCode produce (Topic *topic, int32_t partition,
                             const std::vector<char> *payload,
                             const std::vector<char> *key,
                             void *msg_opaque) = 0;


  virtual ErrorCode flush (int timeout_ms) = 0;


  virtual ErrorCode purge (int purge_flags) = 0;

  enum {
    PURGE_QUEUE = 0x1, 
    PURGE_INFLIGHT = 0x2, 
    PURGE_NON_BLOCKING = 0x4 /* Don't wait for background queue
                              * purging to finish. */
  };

  virtual Error *init_transactions (int timeout_ms) = 0;


  virtual Error *begin_transaction () = 0;

  virtual Error *send_offsets_to_transaction (
          const std::vector<TopicPartition*> &offsets,
          const ConsumerGroupMetadata *group_metadata,
          int timeout_ms) = 0;

  virtual Error *commit_transaction (int timeout_ms) = 0;

  virtual Error *abort_transaction (int timeout_ms) = 0;

};

class BrokerMetadata {
 public:
  virtual int32_t id() const = 0;

  virtual const std::string host() const = 0;

  virtual int port() const = 0;

  virtual ~BrokerMetadata() = 0;
};



class PartitionMetadata {
 public:
  typedef std::vector<int32_t> ReplicasVector;
  typedef std::vector<int32_t> ISRSVector;

  typedef ReplicasVector::const_iterator ReplicasIterator;
  typedef ISRSVector::const_iterator     ISRSIterator;


  virtual int32_t id() const = 0;

  virtual ErrorCode err() const = 0;

  virtual int32_t leader() const = 0;

  virtual const std::vector<int32_t> *replicas() const = 0;

  virtual const std::vector<int32_t> *isrs() const = 0;

  virtual ~PartitionMetadata() = 0;
};



class TopicMetadata {
 public:
  typedef std::vector<const PartitionMetadata*> PartitionMetadataVector;
  typedef PartitionMetadataVector::const_iterator PartitionMetadataIterator;

  virtual const std::string topic() const = 0;

  virtual const PartitionMetadataVector *partitions() const = 0;

  virtual ErrorCode err() const = 0;

  virtual ~TopicMetadata() = 0;
};


class Metadata {
 public:
  typedef std::vector<const BrokerMetadata*> BrokerMetadataVector;
  typedef std::vector<const TopicMetadata*>  TopicMetadataVector;

  typedef BrokerMetadataVector::const_iterator BrokerMetadataIterator;
  typedef TopicMetadataVector::const_iterator  TopicMetadataIterator;


  virtual const BrokerMetadataVector *brokers() const = 0;

  virtual const TopicMetadataVector  *topics() const = 0;

  virtual int32_t orig_broker_id() const = 0;

  virtual const std::string orig_broker_name() const = 0;

  virtual ~Metadata() = 0;
};

}

#endif /* _RDKAFKACPP_H_ */