path: root/fluent-bit/lib/librdkafka-2.1.0/src/rdkafka_mock_int.h

/*
 * librdkafka - Apache Kafka C library
 *
 * Copyright (c) 2019 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 _RDKAFKA_MOCK_INT_H_
#define _RDKAFKA_MOCK_INT_H_

/**
 * @name Mock cluster - internal data types
 *
 */


/**
 * @struct Response error and/or RTT-delay to return to client.
 */
typedef struct rd_kafka_mock_error_rtt_s {
        rd_kafka_resp_err_t err; /**< Error response (or 0) */
        rd_ts_t rtt;             /**< RTT/delay in microseconds (or 0) */
} rd_kafka_mock_error_rtt_t;

/**
 * @struct A stack of errors or rtt latencies to return to the client,
 *         one by one until the stack is depleted.
 */
typedef struct rd_kafka_mock_error_stack_s {
        TAILQ_ENTRY(rd_kafka_mock_error_stack_s) link;
        int16_t ApiKey; /**< Optional ApiKey for which this stack
                         *   applies to, else -1. */
        size_t cnt;     /**< Current number of errors in .errs */
        size_t size;    /**< Current allocated size for .errs (in elements) */
        rd_kafka_mock_error_rtt_t *errs; /**< Array of errors/rtts */
} rd_kafka_mock_error_stack_t;

typedef TAILQ_HEAD(rd_kafka_mock_error_stack_head_s,
                   rd_kafka_mock_error_stack_s)
    rd_kafka_mock_error_stack_head_t;


/**
 * @struct Consumer group protocol name and metadata.
 */
typedef struct rd_kafka_mock_cgrp_proto_s {
        rd_kafkap_str_t *name;
        rd_kafkap_bytes_t *metadata;
} rd_kafka_mock_cgrp_proto_t;

/**
 * @struct Consumer group member
 */
typedef struct rd_kafka_mock_cgrp_member_s {
        TAILQ_ENTRY(rd_kafka_mock_cgrp_member_s) link;
        char *id;                 /**< MemberId */
        char *group_instance_id;  /**< Group instance id */
        rd_ts_t ts_last_activity; /**< Last activity, e.g., Heartbeat */
        rd_kafka_mock_cgrp_proto_t *protos;      /**< Protocol names */
        int proto_cnt;                           /**< Number of protocols */
        rd_kafkap_bytes_t *assignment;           /**< Current assignment */
        rd_kafka_buf_t *resp;                    /**< Current response buffer */
        struct rd_kafka_mock_connection_s *conn; /**< Connection, may be NULL
                                                  *   if there is no ongoing
                                                  *   request. */
} rd_kafka_mock_cgrp_member_t;

/**
 * @struct Consumer group.
 */
typedef struct rd_kafka_mock_cgrp_s {
        TAILQ_ENTRY(rd_kafka_mock_cgrp_s) link;
        struct rd_kafka_mock_cluster_s *cluster; /**< Cluster */
        struct rd_kafka_mock_connection_s *conn; /**< Connection */
        char *id;                                /**< Group Id */
        char *protocol_type;                     /**< Protocol type */
        char *protocol_name;                     /**< Elected protocol name */
        int32_t generation_id;                   /**< Generation Id */
        int session_timeout_ms;                  /**< Session timeout */
        enum { RD_KAFKA_MOCK_CGRP_STATE_EMPTY,   /* No members */
               RD_KAFKA_MOCK_CGRP_STATE_JOINING, /* Members are joining */
               RD_KAFKA_MOCK_CGRP_STATE_SYNCING, /* Syncing assignments */
               RD_KAFKA_MOCK_CGRP_STATE_REBALANCING, /* Rebalance triggered */
               RD_KAFKA_MOCK_CGRP_STATE_UP,          /* Group is operational */
        } state;                        /**< Consumer group state */
        rd_kafka_timer_t session_tmr;   /**< Session timeout timer */
        rd_kafka_timer_t rebalance_tmr; /**< Rebalance state timer */
        TAILQ_HEAD(, rd_kafka_mock_cgrp_member_s) members; /**< Group members */
        int member_cnt;      /**< Number of group members */
        int last_member_cnt; /**< Mumber of group members at last election */
        int assignment_cnt;  /**< Number of member assignments in last Sync */
        rd_kafka_mock_cgrp_member_t *leader; /**< Elected leader */
} rd_kafka_mock_cgrp_t;


/**
 * @struct TransactionalId + PID (+ optional sequence state)
 */
typedef struct rd_kafka_mock_pid_s {
        rd_kafka_pid_t pid;

        /* BaseSequence tracking (partition) */
        int8_t window;  /**< increases up to 5 */
        int8_t lo;      /**< Window low bucket: oldest */
        int8_t hi;      /**< Window high bucket: most recent */
        int32_t seq[5]; /**< Next expected BaseSequence for each bucket */

        char TransactionalId[1]; /**< Allocated after this structure */
} rd_kafka_mock_pid_t;

/**
 * @brief rd_kafka_mock_pid_t.pid Pid (not epoch) comparator
 */
static RD_UNUSED int rd_kafka_mock_pid_cmp_pid(const void *_a, const void *_b) {
        const rd_kafka_mock_pid_t *a = _a, *b = _b;

        if (a->pid.id < b->pid.id)
                return -1;
        else if (a->pid.id > b->pid.id)
                return 1;

        return 0;
}

/**
 * @brief rd_kafka_mock_pid_t.pid TransactionalId,Pid,epoch comparator
 */
static RD_UNUSED int rd_kafka_mock_pid_cmp(const void *_a, const void *_b) {
        const rd_kafka_mock_pid_t *a = _a, *b = _b;
        int r;

        r = strcmp(a->TransactionalId, b->TransactionalId);
        if (r)
                return r;

        if (a->pid.id < b->pid.id)
                return -1;
        else if (a->pid.id > b->pid.id)
                return 1;

        if (a->pid.epoch < b->pid.epoch)
                return -1;
        if (a->pid.epoch > b->pid.epoch)
                return 1;

        return 0;
}



/**
 * @struct A real TCP connection from the client to a mock broker.
 */
typedef struct rd_kafka_mock_connection_s {
        TAILQ_ENTRY(rd_kafka_mock_connection_s) link;
        rd_kafka_transport_t *transport; /**< Socket transport */
        rd_kafka_buf_t *rxbuf;           /**< Receive buffer */
        rd_kafka_bufq_t outbufs;         /**< Send buffers */
        short *poll_events;              /**< Events to poll, points to
                                          *   the broker's pfd array */
        struct sockaddr_in peer;         /**< Peer address */
        struct rd_kafka_mock_broker_s *broker;
        rd_kafka_timer_t write_tmr; /**< Socket write delay timer */
} rd_kafka_mock_connection_t;


/**
 * @struct Mock broker
 */
typedef struct rd_kafka_mock_broker_s {
        TAILQ_ENTRY(rd_kafka_mock_broker_s) link;
        int32_t id;
        char advertised_listener[128];
        struct sockaddr_in sin; /**< Bound address:port */
        uint16_t port;
        char *rack;
        rd_bool_t up;
        rd_ts_t rtt; /**< RTT in microseconds */

        rd_socket_t listen_s; /**< listen() socket */

        TAILQ_HEAD(, rd_kafka_mock_connection_s) connections;

        /**< Per-protocol request error stack.
         *   @locks mcluster->lock */
        rd_kafka_mock_error_stack_head_t errstacks;

        struct rd_kafka_mock_cluster_s *cluster;
} rd_kafka_mock_broker_t;


/**
 * @struct A Kafka-serialized MessageSet
 */
typedef struct rd_kafka_mock_msgset_s {
        TAILQ_ENTRY(rd_kafka_mock_msgset_s) link;
        int64_t first_offset; /**< First offset in batch */
        int64_t last_offset;  /**< Last offset in batch */
        int32_t leader_epoch; /**< Msgset leader epoch */
        rd_kafkap_bytes_t bytes;
        /* Space for bytes.data is allocated after the msgset_t */
} rd_kafka_mock_msgset_t;


/**
 * @struct Committed offset for a group and partition.
 */
typedef struct rd_kafka_mock_committed_offset_s {
        /**< mpart.committed_offsets */
        TAILQ_ENTRY(rd_kafka_mock_committed_offset_s) link;
        char *group;               /**< Allocated along with the struct */
        int64_t offset;            /**< Committed offset */
        rd_kafkap_str_t *metadata; /**< Metadata, allocated separately */
} rd_kafka_mock_committed_offset_t;


TAILQ_HEAD(rd_kafka_mock_msgset_tailq_s, rd_kafka_mock_msgset_s);

/**
 * @struct Mock partition
 */
typedef struct rd_kafka_mock_partition_s {
        TAILQ_ENTRY(rd_kafka_mock_partition_s) leader_link;
        int32_t id;

        int32_t leader_epoch;          /**< Leader epoch, bumped on each
                                        *   partition leader change. */
        int64_t start_offset;          /**< Actual/leader start offset */
        int64_t end_offset;            /**< Actual/leader end offset */
        int64_t follower_start_offset; /**< Follower's start offset */
        int64_t follower_end_offset;   /**< Follower's end offset */
        rd_bool_t update_follower_start_offset; /**< Keep follower_start_offset
                                                 *   in synch with start_offset
                                                 */
        rd_bool_t update_follower_end_offset;   /**< Keep follower_end_offset
                                                 *   in synch with end_offset
                                                 */

        struct rd_kafka_mock_msgset_tailq_s msgsets;
        size_t size;     /**< Total size of all .msgsets */
        size_t cnt;      /**< Total count of .msgsets */
        size_t max_size; /**< Maximum size of all .msgsets, may be overshot. */
        size_t max_cnt;  /**< Maximum number of .msgsets */

        /**< Committed offsets */
        TAILQ_HEAD(, rd_kafka_mock_committed_offset_s) committed_offsets;

        rd_kafka_mock_broker_t *leader;
        rd_kafka_mock_broker_t **replicas;
        int replica_cnt;

        rd_list_t pidstates; /**< PID states */

        int32_t follower_id; /**< Preferred replica/follower */

        struct rd_kafka_mock_topic_s *topic;
} rd_kafka_mock_partition_t;


/**
 * @struct Mock topic
 */
typedef struct rd_kafka_mock_topic_s {
        TAILQ_ENTRY(rd_kafka_mock_topic_s) link;
        char *name;

        rd_kafka_mock_partition_t *partitions;
        int partition_cnt;

        rd_kafka_resp_err_t err; /**< Error to return in protocol requests
                                  *   for this topic. */

        struct rd_kafka_mock_cluster_s *cluster;
} rd_kafka_mock_topic_t;

/**
 * @struct Explicitly set coordinator.
 */
typedef struct rd_kafka_mock_coord_s {
        TAILQ_ENTRY(rd_kafka_mock_coord_s) link;
        rd_kafka_coordtype_t type;
        char *key;
        int32_t broker_id;
} rd_kafka_mock_coord_t;


typedef void(rd_kafka_mock_io_handler_t)(
    struct rd_kafka_mock_cluster_s *mcluster,
    rd_socket_t fd,
    int events,
    void *opaque);

struct rd_kafka_mock_api_handler {
        int16_t MinVersion;
        int16_t MaxVersion;
        int16_t FlexVersion; /**< First Flexible version */
        int (*cb)(rd_kafka_mock_connection_t *mconn, rd_kafka_buf_t *rkbuf);
};

extern const struct rd_kafka_mock_api_handler
    rd_kafka_mock_api_handlers[RD_KAFKAP__NUM];



/**
 * @struct Mock cluster.
 *
 * The cluster IO loop runs in a separate thread where all
 * broker IO is handled.
 *
 * No locking is needed.
 */
struct rd_kafka_mock_cluster_s {
        char id[32]; /**< Generated cluster id */

        rd_kafka_t *rk;

        int32_t controller_id; /**< Current controller */

        TAILQ_HEAD(, rd_kafka_mock_broker_s) brokers;
        int broker_cnt;

        TAILQ_HEAD(, rd_kafka_mock_topic_s) topics;
        int topic_cnt;

        TAILQ_HEAD(, rd_kafka_mock_cgrp_s) cgrps;

        /** Explicit coordinators (set with mock_set_coordinator()) */
        TAILQ_HEAD(, rd_kafka_mock_coord_s) coords;

        /** Current transactional producer PIDs.
         *  Element type is a malloced rd_kafka_mock_pid_t*. */
        rd_list_t pids;

        char *bootstraps; /**< bootstrap.servers */

        thrd_t thread; /**< Mock thread */

        rd_kafka_q_t *ops; /**< Control ops queue for interacting with the
                            *   cluster. */

        rd_socket_t wakeup_fds[2]; /**< Wake-up fds for use with .ops */

        rd_bool_t run; /**< Cluster will run while this value is true */

        int fd_cnt;         /**< Number of file descriptors */
        int fd_size;        /**< Allocated size of .fds
                             *   and .handlers */
        struct pollfd *fds; /**< Dynamic array */

        rd_kafka_broker_t *dummy_rkb; /**< Some internal librdkafka APIs
                                       *   that we are reusing requires a
                                       *   broker object, we use the
                                       *   internal broker and store it
                                       *   here for convenient access. */

        struct {
                int partition_cnt;      /**< Auto topic create part cnt */
                int replication_factor; /**< Auto topic create repl factor */
        } defaults;

        /**< Dynamic array of IO handlers for corresponding fd in .fds */
        struct {
                rd_kafka_mock_io_handler_t *cb; /**< Callback */
                void *opaque;                   /**< Callbacks' opaque */
        } * handlers;

        /**< Per-protocol request error stack. */
        rd_kafka_mock_error_stack_head_t errstacks;

        /**< Request handlers */
        struct rd_kafka_mock_api_handler api_handlers[RD_KAFKAP__NUM];

        /**< Mutex for:
         *   .errstacks
         *   .apiversions
         */
        mtx_t lock;

        rd_kafka_timers_t timers; /**< Timers */
};



rd_kafka_buf_t *rd_kafka_mock_buf_new_response(const rd_kafka_buf_t *request);
void rd_kafka_mock_connection_send_response(rd_kafka_mock_connection_t *mconn,
                                            rd_kafka_buf_t *resp);
void rd_kafka_mock_connection_set_blocking(rd_kafka_mock_connection_t *mconn,
                                           rd_bool_t blocking);

rd_kafka_mock_partition_t *
rd_kafka_mock_partition_find(const rd_kafka_mock_topic_t *mtopic,
                             int32_t partition);
rd_kafka_mock_topic_t *
rd_kafka_mock_topic_auto_create(rd_kafka_mock_cluster_t *mcluster,
                                const char *topic,
                                int partition_cnt,
                                rd_kafka_resp_err_t *errp);
rd_kafka_mock_topic_t *
rd_kafka_mock_topic_find(const rd_kafka_mock_cluster_t *mcluster,
                         const char *name);
rd_kafka_mock_topic_t *
rd_kafka_mock_topic_find_by_kstr(const rd_kafka_mock_cluster_t *mcluster,
                                 const rd_kafkap_str_t *kname);
rd_kafka_mock_broker_t *
rd_kafka_mock_cluster_get_coord(rd_kafka_mock_cluster_t *mcluster,
                                rd_kafka_coordtype_t KeyType,
                                const rd_kafkap_str_t *Key);

rd_kafka_mock_committed_offset_t *
rd_kafka_mock_committed_offset_find(const rd_kafka_mock_partition_t *mpart,
                                    const rd_kafkap_str_t *group);
rd_kafka_mock_committed_offset_t *
rd_kafka_mock_commit_offset(rd_kafka_mock_partition_t *mpart,
                            const rd_kafkap_str_t *group,
                            int64_t offset,
                            const rd_kafkap_str_t *metadata);

const rd_kafka_mock_msgset_t *
rd_kafka_mock_msgset_find(const rd_kafka_mock_partition_t *mpart,
                          int64_t offset,
                          rd_bool_t on_follower);

rd_kafka_resp_err_t
rd_kafka_mock_next_request_error(rd_kafka_mock_connection_t *mconn,
                                 rd_kafka_buf_t *resp);

rd_kafka_resp_err_t
rd_kafka_mock_partition_log_append(rd_kafka_mock_partition_t *mpart,
                                   const rd_kafkap_bytes_t *records,
                                   const rd_kafkap_str_t *TransactionalId,
                                   int64_t *BaseOffset);

rd_kafka_resp_err_t rd_kafka_mock_partition_leader_epoch_check(
    const rd_kafka_mock_partition_t *mpart,
    int32_t leader_epoch);

int64_t rd_kafka_mock_partition_offset_for_leader_epoch(
    const rd_kafka_mock_partition_t *mpart,
    int32_t leader_epoch);


/**
 * @returns true if the ApiVersion is supported, else false.
 */
static RD_UNUSED rd_bool_t
rd_kafka_mock_cluster_ApiVersion_check(const rd_kafka_mock_cluster_t *mcluster,
                                       int16_t ApiKey,
                                       int16_t ApiVersion) {
        return (ApiVersion >= mcluster->api_handlers[ApiKey].MinVersion &&
                ApiVersion <= mcluster->api_handlers[ApiKey].MaxVersion);
}


rd_kafka_resp_err_t
rd_kafka_mock_pid_find(rd_kafka_mock_cluster_t *mcluster,
                       const rd_kafkap_str_t *TransactionalId,
                       const rd_kafka_pid_t pid,
                       rd_kafka_mock_pid_t **mpidp);


/**
 * @name Mock consumer group (rdkafka_mock_cgrp.c)
 * @{
 */
void rd_kafka_mock_cgrp_member_active(rd_kafka_mock_cgrp_t *mcgrp,
                                      rd_kafka_mock_cgrp_member_t *member);
void rd_kafka_mock_cgrp_member_assignment_set(
    rd_kafka_mock_cgrp_t *mcgrp,
    rd_kafka_mock_cgrp_member_t *member,
    const rd_kafkap_bytes_t *Metadata);
rd_kafka_resp_err_t
rd_kafka_mock_cgrp_member_sync_set(rd_kafka_mock_cgrp_t *mcgrp,
                                   rd_kafka_mock_cgrp_member_t *member,
                                   rd_kafka_mock_connection_t *mconn,
                                   rd_kafka_buf_t *resp);
rd_kafka_resp_err_t
rd_kafka_mock_cgrp_member_leave(rd_kafka_mock_cgrp_t *mcgrp,
                                rd_kafka_mock_cgrp_member_t *member);
void rd_kafka_mock_cgrp_protos_destroy(rd_kafka_mock_cgrp_proto_t *protos,
                                       int proto_cnt);
rd_kafka_resp_err_t
rd_kafka_mock_cgrp_member_add(rd_kafka_mock_cgrp_t *mcgrp,
                              rd_kafka_mock_connection_t *mconn,
                              rd_kafka_buf_t *resp,
                              const rd_kafkap_str_t *MemberId,
                              const rd_kafkap_str_t *ProtocolType,
                              rd_kafka_mock_cgrp_proto_t *protos,
                              int proto_cnt,
                              int session_timeout_ms);
rd_kafka_resp_err_t
rd_kafka_mock_cgrp_check_state(rd_kafka_mock_cgrp_t *mcgrp,
                               rd_kafka_mock_cgrp_member_t *member,
                               const rd_kafka_buf_t *request,
                               int32_t generation_id);
rd_kafka_mock_cgrp_member_t *
rd_kafka_mock_cgrp_member_find(const rd_kafka_mock_cgrp_t *mcgrp,
                               const rd_kafkap_str_t *MemberId);
void rd_kafka_mock_cgrp_destroy(rd_kafka_mock_cgrp_t *mcgrp);
rd_kafka_mock_cgrp_t *rd_kafka_mock_cgrp_find(rd_kafka_mock_cluster_t *mcluster,
                                              const rd_kafkap_str_t *GroupId);
rd_kafka_mock_cgrp_t *
rd_kafka_mock_cgrp_get(rd_kafka_mock_cluster_t *mcluster,
                       const rd_kafkap_str_t *GroupId,
                       const rd_kafkap_str_t *ProtocolType);
void rd_kafka_mock_cgrps_connection_closed(rd_kafka_mock_cluster_t *mcluster,
                                           rd_kafka_mock_connection_t *mconn);


/**
 *@}
 */


#include "rdkafka_mock.h"

#endif /* _RDKAFKA_MOCK_INT_H_ */