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

#pragma once

#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

namespace RdKafka {


#define RD_KAFKA_VERSION  0x000b00ff

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__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_GROUP_LOAD_IN_PROGRESS = 14,
        ERR_GROUP_COORDINATOR_NOT_AVAILABLE = 15,
        ERR_NOT_COORDINATOR_FOR_GROUP = 16,
        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
};


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


/* Forward declarations */
class Producer;
class Message;
class Queue;
class Event;
class Topic;
class TopicPartition;
class Metadata;
class KafkaConsumer;
class RD_EXPORT DeliveryReportCb {
 public:
  virtual void dr_cb (Message &message) = 0;

  virtual ~DeliveryReportCb() { }
};


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



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 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,
                                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 get(const std::string &name,
      std::string &value) const = 0;

  virtual Conf::ConfResult get(DeliveryReportCb *&dr_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 std::list<std::string> *dump () = 0;

  virtual Conf::ConfResult set (const std::string &name, ConsumeCb *consume_cb,
                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;
};


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,
                        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;
};


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 Message {
 public:
  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;
};

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

class RD_EXPORT KafkaConsumer : public virtual Handle {
public:
  static KafkaConsumer *create (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;
};


class RD_EXPORT Consumer : public virtual Handle {
 public:
  static Consumer *create (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 (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 (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;
};

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

}