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 0x010900ff

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);

RD_EXPORT
void *mem_malloc(size_t size);

RD_EXPORT
void mem_free(void *ptr);

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__ASSIGNMENT_LOST = -142,
  ERR__NOOP = -141,
  ERR__AUTO_OFFSET_RESET = -140,

  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,
  ERR_THROTTLING_QUOTA_EXCEEDED = 89,
  ERR_PRODUCER_FENCED = 90,
  ERR_RESOURCE_NOT_FOUND = 91,
  ERR_DUPLICATE_RESOURCE = 92,
  ERR_UNACCEPTABLE_CREDENTIAL = 93,
  ERR_INCONSISTENT_VOTER_SET = 94,
  ERR_INVALID_UPDATE_VERSION = 95,
  ERR_FEATURE_UPDATE_FAILED = 96,
  ERR_PRINCIPAL_DESERIALIZATION_FAILURE = 97
};


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 Handle;
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(
      RdKafka::Handle *handle,
      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;

  virtual Conf::ConfResult set_engine_callback_data(void *value,
                                                    std::string &errstr) = 0;


  virtual Conf::ConfResult enable_sasl_queue(bool enable,
                                             std::string &errstr) = 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;

  virtual Error *sasl_background_callbacks_enable() = 0;


  virtual Queue *get_sasl_queue() = 0;

  virtual Queue *get_background_queue() = 0;



  virtual void *mem_malloc(size_t size) = 0;

  virtual void mem_free(void *ptr) = 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)
        mem_free(value_);

      value_ = copy_value(other.value_, value_size_);

      return *this;
    }

    ~Header() {
      if (value_ != NULL)
        mem_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 *)mem_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;


  virtual bool assignment_lost() = 0;

  virtual std::string rebalance_protocol() = 0;


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


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

  virtual Error *close(Queue *queue) = 0;


  virtual bool closed() = 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;
};

}  // namespace RdKafka


#endif /* _RDKAFKACPP_H_ */