plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_admin.c
static rd_kafka_broker_t *
rd_kafka_admin_common_get_controller (rd_kafka_t *rk,
                                      rd_kafka_op_t *rko) {
        rd_kafka_broker_t *rkb;

        rd_kafka_dbg(rk, ADMIN, "ADMIN", "%s: looking up controller",
                     rd_kafka_op2str(rko->rko_type));

        /* Since we're iterating over this controller_async() call
         * (asynchronously) until a controller is availabe (or timeout)
         * we need to re-enable the eonce to be triggered again (which
         * is not necessary the first time we get here, but there
         * is no harm doing it then either). */
        rd_kafka_enq_once_reenable(rko->rko_u.admin_request.eonce,
                                   rko, RD_KAFKA_REPLYQ(rk->rk_ops, 0));

        /* Look up the controller asynchronously, if the controller
         * is not available the eonce is registered for broker
         * state changes which will cause our function to be called
         * again as soon as (any) broker state changes.
         * When we are called again we perform the controller lookup
         * again and hopefully get an rkb back, otherwise defer a new
         * async wait. Repeat until success or timeout. */
        if (!(rkb = rd_kafka_broker_controller_async(
                      rk, RD_KAFKA_BROKER_STATE_UP,
                      rko->rko_u.admin_request.eonce))) {
                /* Controller not available, wait asynchronously
                 * for controller code to trigger eonce. */
                return NULL;
        }

        rd_kafka_dbg(rk, ADMIN, "ADMIN", "%s: controller %s",
                     rd_kafka_op2str(rko->rko_type), rkb->rkb_name);

        return rkb;
}
/**
 * @brief Asynchronously look up the controller.
 *        To be called repeatedly from each invocation of the worker
 *        when in state RD_KAFKA_ADMIN_STATE_WAIT_CONTROLLER until
 *        a valid rkb is returned.
 *
 * @returns the controller rkb with refcount increased, or NULL if not yet
 *          available.
 */
@brief Asynchronously look up the controller.
To be called repeatedly from each invocation of the worker
when in state RD_KAFKA_ADMIN_STATE_WAIT_CONTROLLER until
a valid rkb is returned.

@returns the controller rkb with refcount increased, or NULL if not yet
available.
@brief Asynchronously look up the controller . To be called repeatedly from each invocation of the worker when in state RD_KAFKA_ADMIN_STATE_WAIT_CONTROLLER until a valid rkb is returned . @returns the controller rkb with refcount increased or NULL if not yet available .
static rd_kafka_broker_t *
rd_kafka_admin_common_get_controller (rd_kafka_t *rk,
                                      rd_kafka_op_t *rko) {
        rd_kafka_broker_t *rkb;
        rd_kafka_dbg(rk, ADMIN, "ADMIN", "%s: looking up controller",
                     rd_kafka_op2str(rko->rko_type));
        rd_kafka_enq_once_reenable(rko->rko_u.admin_request.eonce,
                                   rko, RD_KAFKA_REPLYQ(rk->rk_ops, 0));
        if (!(rkb = rd_kafka_broker_controller_async(
                      rk, RD_KAFKA_BROKER_STATE_UP,
                      rko->rko_u.admin_request.eonce))) {
                return NULL;
        }
        rd_kafka_dbg(rk, ADMIN, "ADMIN", "%s: controller %s",
                     rd_kafka_op2str(rko->rko_type), rkb->rkb_name);
        return rkb;
}
static rd_kafka_broker_t * rd_kafka_admin_common_get_controller ( rd_kafka_t * rk , rd_kafka_op_t * rko ) { rd_kafka_broker_t * rkb ; rd_kafka_dbg ( rk , ADMIN , " " , " " , rd_kafka_op2str ( rko -> rko_type ) ) ; rd_kafka_enq_once_reenable ( rko -> rko_u . admin_request . eonce , rko , RD_KAFKA_REPLYQ ( rk -> rk_ops , 0 ) ) ; if ( ! ( rkb = rd_kafka_broker_controller_async ( rk , RD_KAFKA_BROKER_STATE_UP , rko -> rko_u . admin_request . eonce ) ) ) { return NULL ; } rd_kafka_dbg ( rk , ADMIN , " " , " " , rd_kafka_op2str ( rko -> rko_type ) , rkb -> rkb_name ) ; return rkb ; }
@brief Asynchronously look up the controller.
@brief Asynchronously look up the controller .
/* Since we're iterating over this controller_async() call
         * (asynchronously) until a controller is availabe (or timeout)
         * we need to re-enable the eonce to be triggered again (which
         * is not necessary the first time we get here, but there
         * is no harm doing it then either). */ /* Look up the controller asynchronously, if the controller
         * is not available the eonce is registered for broker
         * state changes which will cause our function to be called
         * again as soon as (any) broker state changes.
         * When we are called again we perform the controller lookup
         * again and hopefully get an rkb back, otherwise defer a new
         * async wait. Repeat until success or timeout. */ /* Controller not available, wait asynchronously
                 * for controller code to trigger eonce. */