librdkafka: 如何设置Kafka消费者订阅消息的起始偏移位置

/** * @brief Generic place holder for a specific Topic+Partition. * * @sa rd_kafka_topic_partition_list_new() */typedef struct rd_kafka_topic_partition_s {        char        *topic;             /**< Topic name */        int32_t      partition;         /**< Partition */      int64_t      offset;            /**< Offset */        void        *metadata;          /**< Metadata */        size_t       metadata_size;     /**< Metadata size */        void        *opaque;            /**< Application opaque */        rd_kafka_resp_err_t err;        /**< Error code, depending on use. */        void       *_private;           /**< INTERNAL USE ONLY,                                         *   INITIALIZE TO ZERO, DO NOT TOUCH */} rd_kafka_topic_partition_t;
////////////////////////////////////////////////////////////#define RD_KAFKA_OFFSET_BEGINNING -2  /**< Start consuming from beginning of               *   kafka partition queue: oldest msg */#define RD_KAFKA_OFFSET_END       -1  /**< Start consuming from end of kafka               *   partition queue: next msg */#define RD_KAFKA_OFFSET_STORED -1000  /**< Start consuming from offset retrieved               *   from offset store */#define RD_KAFKA_OFFSET_INVALID -1001 /**< Invalid offset */

/** @cond NO_DOC */#define RD_KAFKA_OFFSET_TAIL_BASE -2000 /* internal: do not use *//** @endcond */
/** * @brief Start consuming \p CNT messages from topic's current end offset. * * That is, if current end offset is 12345 and \p CNT is 200, it will start * consuming from offset \c 12345-200 = \c 12145. */#define RD_KAFKA_OFFSET_TAIL(CNT)  (RD_KAFKA_OFFSET_TAIL_BASE - (CNT))

/** * @brief Atomic assignment of partitions to consume. * * The new \p partitions will replace the existing assignment. * * When used from a rebalance callback the application shall pass the * partition list passed to the callback (or a copy of it) (even if the list * is empty) rather than NULL to maintain internal join state.
 * A zero-length \p partitions will treat the partitions as a valid, * albeit empty, assignment, and maintain internal state, while a \c NULL * value for \p partitions will reset and clear the internal state. */RD_EXPORT rd_kafka_resp_err_trd_kafka_assign (rd_kafka_t *rk,                 const rd_kafka_topic_partition_list_t *partitions);

rd_kafka_topic_partition_list_t *partitions;partitions = rd_kafka_topic_partition_list_new(0);rd_kafka_topic_partition_list_add(partitions, "mytopic", 3)->offset = 1234;rd_kafka_assign(rk, partitions);rd_kafka_topic_partition_list_destroy(partitions);  

void my_rebalance_cb (rd_kafka_t *rk, rd_kafka_resp_err_t err,                      rd_kafka_topic_partition_list_t *partitions, void *opaque) {   if (err == RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS) {       rd_kafka_topic_partition_t *part;       if ((part = rd_kafka_topic_partition_list_find(partitions, "mytopic", 3)))           part->offset = 1234;       rd_kafka_assign(rk, partitions);   }  else {       rd_kafka_assign(rk, NULL);   }}