src/rdkafka_coord.c

/*
 * librdkafka - Apache Kafka C library
 *
 * Copyright (c) 2019-2022, 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.
 */


#include "rdkafka_int.h"
#include "rdkafka_request.h"
#include "rdkafka_coord.h"


/**
 * @name Coordinator cache
 * @{
 *
 */
void rd_kafka_coord_cache_entry_destroy(rd_kafka_coord_cache_t *cc,
                                        rd_kafka_coord_cache_entry_t *cce) {
        rd_assert(cc->cc_cnt > 0);
        rd_free(cce->cce_coordkey);
        rd_kafka_broker_destroy(cce->cce_rkb);
        TAILQ_REMOVE(&cc->cc_entries, cce, cce_link);
        cc->cc_cnt--;
        rd_free(cce);
}


/**
 * @brief Delete any expired cache entries
 *
 * @locality rdkafka main thread
 */
void rd_kafka_coord_cache_expire(rd_kafka_coord_cache_t *cc) {
        rd_kafka_coord_cache_entry_t *cce, *next;
        rd_ts_t expire = rd_clock() - cc->cc_expire_thres;

        next = TAILQ_LAST(&cc->cc_entries, rd_kafka_coord_cache_head_s);
        while (next) {
                cce = next;

                if (cce->cce_ts_used > expire)
                        break;

                next = TAILQ_PREV(cce, rd_kafka_coord_cache_head_s, cce_link);
                rd_kafka_coord_cache_entry_destroy(cc, cce);
        }
}


static rd_kafka_coord_cache_entry_t *
rd_kafka_coord_cache_find(rd_kafka_coord_cache_t *cc,
                          rd_kafka_coordtype_t coordtype,
                          const char *coordkey) {
        rd_kafka_coord_cache_entry_t *cce;

        TAILQ_FOREACH(cce, &cc->cc_entries, cce_link) {
                if (cce->cce_coordtype == coordtype &&
                    !strcmp(cce->cce_coordkey, coordkey)) {
                        /* Match */
                        cce->cce_ts_used = rd_clock();
                        if (TAILQ_FIRST(&cc->cc_entries) != cce) {
                                /* Move to head of list */
                                TAILQ_REMOVE(&cc->cc_entries, cce, cce_link);
                                TAILQ_INSERT_HEAD(&cc->cc_entries, cce,
                                                  cce_link);
                        }
                        return cce;
                }
        }

        return NULL;
}


rd_kafka_broker_t *rd_kafka_coord_cache_get(rd_kafka_coord_cache_t *cc,
                                            rd_kafka_coordtype_t coordtype,
                                            const char *coordkey) {
        rd_kafka_coord_cache_entry_t *cce;

        cce = rd_kafka_coord_cache_find(cc, coordtype, coordkey);
        if (!cce)
                return NULL;

        rd_kafka_broker_keep(cce->cce_rkb);
        return cce->cce_rkb;
}


static void rd_kafka_coord_cache_add(rd_kafka_coord_cache_t *cc,
                                     rd_kafka_coordtype_t coordtype,
                                     const char *coordkey,
                                     rd_kafka_broker_t *rkb) {
        rd_kafka_coord_cache_entry_t *cce;

        if (!(cce = rd_kafka_coord_cache_find(cc, coordtype, coordkey))) {
                if (cc->cc_cnt > 10) {
                        /* Not enough room in cache, remove least used entry */
                        rd_kafka_coord_cache_entry_t *rem = TAILQ_LAST(
                            &cc->cc_entries, rd_kafka_coord_cache_head_s);
                        rd_kafka_coord_cache_entry_destroy(cc, rem);
                }

                cce                = rd_calloc(1, sizeof(*cce));
                cce->cce_coordtype = coordtype;
                cce->cce_coordkey  = rd_strdup(coordkey);
                cce->cce_ts_used   = rd_clock();

                TAILQ_INSERT_HEAD(&cc->cc_entries, cce, cce_link);
                cc->cc_cnt++;
        }

        if (cce->cce_rkb != rkb) {
                if (cce->cce_rkb)
                        rd_kafka_broker_destroy(cce->cce_rkb);
                cce->cce_rkb = rkb;
                rd_kafka_broker_keep(rkb);
        }
}


/**
 * @brief Evict any cache entries for broker \p rkb.
 *
 * Use this when a request returns ERR_NOT_COORDINATOR_FOR...
 *
 * @locality rdkafka main thread
 * @locks none
 */
void rd_kafka_coord_cache_evict(rd_kafka_coord_cache_t *cc,
                                rd_kafka_broker_t *rkb) {
        rd_kafka_coord_cache_entry_t *cce, *tmp;

        TAILQ_FOREACH_SAFE(cce, &cc->cc_entries, cce_link, tmp) {
                if (cce->cce_rkb == rkb)
                        rd_kafka_coord_cache_entry_destroy(cc, cce);
        }
}

/**
 * @brief Destroy all coord cache entries.
 */
void rd_kafka_coord_cache_destroy(rd_kafka_coord_cache_t *cc) {
        rd_kafka_coord_cache_entry_t *cce;

        while ((cce = TAILQ_FIRST(&cc->cc_entries)))
                rd_kafka_coord_cache_entry_destroy(cc, cce);
}


/**
 * @brief Initialize the coord cache.
 *
 * Locking of the coord-cache is up to the owner.
 */
void rd_kafka_coord_cache_init(rd_kafka_coord_cache_t *cc,
                               int expire_thres_ms) {
        TAILQ_INIT(&cc->cc_entries);
        cc->cc_cnt          = 0;
        cc->cc_expire_thres = expire_thres_ms * 1000;
}

/**@}*/


/**
 * @name Asynchronous coordinator requests
 * @{
 *
 */


static void rd_kafka_coord_req_fsm(rd_kafka_t *rk, rd_kafka_coord_req_t *creq);

/**
 * @brief Timer callback for delayed coord requests.
 */
static void rd_kafka_coord_req_tmr_cb(rd_kafka_timers_t *rkts, void *arg) {
        rd_kafka_coord_req_t *creq = arg;

        rd_kafka_coord_req_fsm(rkts->rkts_rk, creq);
}


/**
 * @brief Look up coordinator for \p coordtype and \p coordkey
 *        (either from cache or by FindCoordinator), make sure there is
 *        a connection to the coordinator, and then call \p send_req_cb,
 *        passing the coordinator broker instance and \p rko
 *        to send the request.
 *        These steps may be performed by this function, or asynchronously
 *        at a later time.
 *
 * @param delay_ms If non-zero, delay scheduling of the coord request
 *                 for this long. The passed \p timeout_ms is automatically
 *                 adjusted to + \p delay_ms.
 *
 * Response, or error, is sent on \p replyq with callback \p rkbuf_cb.
 *
 * @locality rdkafka main thread
 * @locks none
 */
void rd_kafka_coord_req(rd_kafka_t *rk,
                        rd_kafka_coordtype_t coordtype,
                        const char *coordkey,
                        rd_kafka_send_req_cb_t *send_req_cb,
                        rd_kafka_op_t *rko,
                        int delay_ms,
                        int timeout_ms,
                        rd_kafka_replyq_t replyq,
                        rd_kafka_resp_cb_t *resp_cb,
                        void *reply_opaque) {
        rd_kafka_coord_req_t *creq;

        creq                    = rd_calloc(1, sizeof(*creq));
        creq->creq_coordtype    = coordtype;
        creq->creq_coordkey     = rd_strdup(coordkey);
        creq->creq_ts_timeout   = rd_timeout_init(delay_ms + timeout_ms);
        creq->creq_send_req_cb  = send_req_cb;
        creq->creq_rko          = rko;
        creq->creq_replyq       = replyq;
        creq->creq_resp_cb      = resp_cb;
        creq->creq_reply_opaque = reply_opaque;
        creq->creq_refcnt       = 1;
        creq->creq_done         = rd_false;
        rd_interval_init(&creq->creq_query_intvl);

        TAILQ_INSERT_TAIL(&rk->rk_coord_reqs, creq, creq_link);

        if (delay_ms)
                rd_kafka_timer_start_oneshot(&rk->rk_timers, &creq->creq_tmr,
                                             rd_true, (rd_ts_t)delay_ms * 1000,
                                             rd_kafka_coord_req_tmr_cb, creq);
        else
                rd_kafka_coord_req_fsm(rk, creq);
}


/**
 * @brief Decrease refcount of creq and free it if no more references.
 *
 * @param done Mark creq as done, having performed its duties. There may still
 *             be lingering references.
 *
 * @returns true if creq was destroyed, else false.
 */
static rd_bool_t rd_kafka_coord_req_destroy(rd_kafka_t *rk,
                                            rd_kafka_coord_req_t *creq,
                                            rd_bool_t done) {

        rd_assert(creq->creq_refcnt > 0);

        if (done) {
                /* Request has been performed, remove from rk_coord_reqs
                 * list so creq won't be triggered again by state broadcasts,
                 * etc. */
                rd_dassert(!creq->creq_done);
                TAILQ_REMOVE(&rk->rk_coord_reqs, creq, creq_link);
                creq->creq_done = rd_true;

                rd_kafka_timer_stop(&rk->rk_timers, &creq->creq_tmr,
                                    RD_DO_LOCK);
        }

        if (--creq->creq_refcnt > 0)
                return rd_false;

        rd_dassert(creq->creq_done);

        /* Clear out coordinator we were waiting for. */
        if (creq->creq_rkb) {
                rd_kafka_broker_persistent_connection_del(
                    creq->creq_rkb, &creq->creq_rkb->rkb_persistconn.coord);
                rd_kafka_broker_destroy(creq->creq_rkb);
                creq->creq_rkb = NULL;
        }

        rd_kafka_replyq_destroy(&creq->creq_replyq);
        rd_free(creq->creq_coordkey);
        rd_free(creq);

        return rd_true;
}

static void rd_kafka_coord_req_keep(rd_kafka_coord_req_t *creq) {
        creq->creq_refcnt++;
}

static void rd_kafka_coord_req_fail(rd_kafka_t *rk,
                                    rd_kafka_coord_req_t *creq,
                                    rd_kafka_resp_err_t err) {
        rd_kafka_op_t *reply;
        rd_kafka_buf_t *rkbuf;

        reply         = rd_kafka_op_new(RD_KAFKA_OP_RECV_BUF);
        reply->rko_rk = rk; /* Set rk since the rkbuf will not have a rkb
                             * to reach it. */
        reply->rko_err = err;

        /* Need a dummy rkbuf to pass state to the buf resp_cb */
        rkbuf                   = rd_kafka_buf_new(0, 0);
        rkbuf->rkbuf_cb         = creq->creq_resp_cb;
        rkbuf->rkbuf_opaque     = creq->creq_reply_opaque;
        reply->rko_u.xbuf.rkbuf = rkbuf;

        rd_kafka_replyq_enq(&creq->creq_replyq, reply, 0);

        rd_kafka_coord_req_destroy(rk, creq, rd_true /*done*/);
}


static void rd_kafka_coord_req_handle_FindCoordinator(rd_kafka_t *rk,
                                                      rd_kafka_broker_t *rkb,
                                                      rd_kafka_resp_err_t err,
                                                      rd_kafka_buf_t *rkbuf,
                                                      rd_kafka_buf_t *request,
                                                      void *opaque) {
        const int log_decode_errors = LOG_ERR;
        rd_kafka_coord_req_t *creq  = opaque;
        int16_t ErrorCode;
        rd_kafkap_str_t Host;
        int32_t NodeId, Port;
        char errstr[256] = "";
        int actions;
        rd_kafka_broker_t *coord;
        rd_kafka_metadata_broker_t mdb = RD_ZERO_INIT;

        /* If creq has finished (possibly because of an earlier FindCoordinator
         * response or a broker state broadcast we simply ignore the
         * response. */
        if (creq->creq_done)
                err = RD_KAFKA_RESP_ERR__DESTROY;

        if (err)
                goto err;

        if (request->rkbuf_reqhdr.ApiVersion >= 1)
                rd_kafka_buf_read_throttle_time(rkbuf);

        rd_kafka_buf_read_i16(rkbuf, &ErrorCode);

        if (request->rkbuf_reqhdr.ApiVersion >= 1) {
                rd_kafkap_str_t ErrorMsg;
                rd_kafka_buf_read_str(rkbuf, &ErrorMsg);
                if (ErrorCode)
                        rd_snprintf(errstr, sizeof(errstr), "%.*s",
                                    RD_KAFKAP_STR_PR(&ErrorMsg));
        }

        if ((err = ErrorCode))
                goto err;

        rd_kafka_buf_read_i32(rkbuf, &NodeId);
        rd_kafka_buf_read_str(rkbuf, &Host);
        rd_kafka_buf_read_i32(rkbuf, &Port);

        mdb.id = NodeId;
        RD_KAFKAP_STR_DUPA(&mdb.host, &Host);
        mdb.port = Port;

        /* Find, update or add broker */
        rd_kafka_broker_update(rk, rkb->rkb_proto, &mdb, &coord);

        if (!coord) {
                err = RD_KAFKA_RESP_ERR__FAIL;
                rd_snprintf(errstr, sizeof(errstr),
                            "Failed to add broker: "
                            "instance is probably terminating");
                goto err;
        }


        rd_kafka_coord_cache_add(&rk->rk_coord_cache, creq->creq_coordtype,
                                 creq->creq_coordkey, coord);
        rd_kafka_broker_destroy(coord); /* refcnt from broker_update() */

        rd_kafka_coord_req_fsm(rk, creq);

        /* Drop refcount from req_fsm() */
        rd_kafka_coord_req_destroy(rk, creq, rd_false /*!done*/);

        return;

err_parse:
        err = rkbuf->rkbuf_err;
err:
        actions = rd_kafka_err_action(
            rkb, err, request,

            RD_KAFKA_ERR_ACTION_SPECIAL, RD_KAFKA_RESP_ERR__DESTROY,

            RD_KAFKA_ERR_ACTION_PERMANENT,
            RD_KAFKA_RESP_ERR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED,

            RD_KAFKA_ERR_ACTION_PERMANENT,
            RD_KAFKA_RESP_ERR_CLUSTER_AUTHORIZATION_FAILED,

            RD_KAFKA_ERR_ACTION_REFRESH, RD_KAFKA_RESP_ERR__TRANSPORT,

            RD_KAFKA_ERR_ACTION_RETRY,
            RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE,

            RD_KAFKA_ERR_ACTION_END);

        if (actions & RD_KAFKA_ERR_ACTION_PERMANENT) {
                rd_kafka_coord_req_fail(rk, creq, err);
                return;

        } else if (actions & RD_KAFKA_ERR_ACTION_RETRY) {
                rd_kafka_buf_retry(rkb, request);
                return; /* Keep refcnt from req_fsm() and retry */
        }

        /* Rely on state broadcast to trigger retry */

        /* Drop refcount from req_fsm() */
        rd_kafka_coord_req_destroy(rk, creq, rd_false /*!done*/);
}


/**
 * @brief State machine for async coordinator requests.
 *
 * @remark May destroy the \p creq.
 *
 * @locality any
 * @locks none
 */
static void rd_kafka_coord_req_fsm(rd_kafka_t *rk, rd_kafka_coord_req_t *creq) {
        rd_kafka_broker_t *rkb;
        rd_kafka_resp_err_t err;

        if (creq->creq_done)
                /* crqeq has already performed its actions, this is a
                 * lingering reference, e.g., a late FindCoordinator response.
                 * Just ignore. */
                return;

        if (unlikely(rd_kafka_terminating(rk))) {
                rd_kafka_coord_req_fail(rk, creq, RD_KAFKA_RESP_ERR__DESTROY);
                return;
        }

        /* Do nothing if creq is delayed and the delay time hasn't expired yet.
         * We will be called again by the timer once it expires.*/
        if (rd_kafka_timer_next(&rk->rk_timers, &creq->creq_tmr, RD_DO_LOCK) >
            0)
                return;

        /* Check cache first */
        rkb = rd_kafka_coord_cache_get(
            &rk->rk_coord_cache, creq->creq_coordtype, creq->creq_coordkey);

        if (rkb) {
                if (rd_kafka_broker_is_up(rkb)) {
                        /* Cached coordinator is up, send request */
                        rd_kafka_replyq_t replyq;

                        /* Clear out previous coordinator we waited for. */
                        if (creq->creq_rkb) {
                                rd_kafka_broker_persistent_connection_del(
                                    creq->creq_rkb,
                                    &creq->creq_rkb->rkb_persistconn.coord);
                                rd_kafka_broker_destroy(creq->creq_rkb);
                                creq->creq_rkb = NULL;
                        }

                        rd_kafka_replyq_copy(&replyq, &creq->creq_replyq);
                        err = creq->creq_send_req_cb(rkb, creq->creq_rko,
                                                     replyq, creq->creq_resp_cb,
                                                     creq->creq_reply_opaque);

                        if (err) {
                                /* Permanent error, e.g., request not
                                 *  supported by broker. */
                                rd_kafka_replyq_destroy(&replyq);
                                rd_kafka_coord_req_fail(rk, creq, err);
                        } else {
                                rd_kafka_coord_req_destroy(rk, creq,
                                                           rd_true /*done*/);
                        }

                } else if (creq->creq_rkb == rkb) {
                        /* No change in coordinator, but it is still not up.
                         * Query for coordinator if at least a second has
                         * passed since this coord_req was created or the
                         * last time we queried. */
                        if (rd_interval(&creq->creq_query_intvl,
                                        1000 * 1000 /* 1s */, 0) > 0) {
                                rd_rkb_dbg(rkb, BROKER, "COORD",
                                           "Coordinator connection is "
                                           "still down: "
                                           "querying for new coordinator");
                                rd_kafka_broker_destroy(rkb);
                                goto query_coord;
                        }

                } else {
                        /* No connection yet.
                         * Let broker thread know we need a connection.
                         * We'll be re-triggered on broker state broadcast. */

                        if (creq->creq_rkb) {
                                /* Clear previous */
                                rd_kafka_broker_persistent_connection_del(
                                    creq->creq_rkb,
                                    &creq->creq_rkb->rkb_persistconn.coord);
                                rd_kafka_broker_destroy(creq->creq_rkb);
                        }

                        rd_kafka_broker_keep(rkb);
                        creq->creq_rkb = rkb;
                        rd_kafka_broker_persistent_connection_add(
                            rkb, &rkb->rkb_persistconn.coord);
                }

                rd_kafka_broker_destroy(rkb);
                return;

        } else if (creq->creq_rkb) {
                /* No coordinator information, clear out the previous
                 * coordinator we waited for. */
                rd_kafka_broker_persistent_connection_del(
                    creq->creq_rkb, &creq->creq_rkb->rkb_persistconn.coord);
                rd_kafka_broker_destroy(creq->creq_rkb);
                creq->creq_rkb = NULL;
        }

query_coord:
        /* Get any usable broker to look up the coordinator */
        rkb = rd_kafka_broker_any_usable(rk, RD_POLL_NOWAIT, RD_DO_LOCK,
                                         RD_KAFKA_FEATURE_BROKER_GROUP_COORD,
                                         "broker to look up coordinator");

        if (!rkb) {
                /* No available brokers yet, we'll be re-triggered on
                 * broker state broadcast. */
                return;
        }


        /* Send FindCoordinator request, the handler will continue
         * the state machine. */
        rd_kafka_coord_req_keep(creq);
        err = rd_kafka_FindCoordinatorRequest(
            rkb, creq->creq_coordtype, creq->creq_coordkey,
            RD_KAFKA_REPLYQ(rk->rk_ops, 0),
            rd_kafka_coord_req_handle_FindCoordinator, creq);

        rd_kafka_broker_destroy(rkb);

        if (err) {
                rd_kafka_coord_req_fail(rk, creq, err);
                /* from keep() above */
                rd_kafka_coord_req_destroy(rk, creq, rd_false /*!done*/);
        }
}


/**
 * @brief Callback called from rdkafka main thread on each
 *        broker state change from or to UP.
 *
 * @locality rdkafka main thread
 * @locks none
 */
void rd_kafka_coord_rkb_monitor_cb(rd_kafka_broker_t *rkb) {
        rd_kafka_t *rk = rkb->rkb_rk;
        rd_kafka_coord_req_t *creq, *tmp;

        /* Run through all coord_req fsms */
        TAILQ_FOREACH_SAFE(creq, &rk->rk_coord_reqs, creq_link, tmp) {
                rd_kafka_coord_req_fsm(rk, creq);
        }
}


/**
 * @brief Instance is terminating: destroy all coord reqs
 */
void rd_kafka_coord_reqs_term(rd_kafka_t *rk) {
        rd_kafka_coord_req_t *creq;

        while ((creq = TAILQ_FIRST(&rk->rk_coord_reqs)))
                rd_kafka_coord_req_fail(rk, creq, RD_KAFKA_RESP_ERR__DESTROY);
}


/**
 * @brief Initialize coord reqs list.
 */
void rd_kafka_coord_reqs_init(rd_kafka_t *rk) {
        TAILQ_INIT(&rk->rk_coord_reqs);
}

/**@}*/