src/rdkafka_broker.c
static rd_kafka_broker_t *rd_kafka_broker_controller_nowait (rd_kafka_t *rk,
                                                             int state) {
        rd_kafka_broker_t *rkb;

        rd_kafka_rdlock(rk);

        if (rk->rk_controllerid == -1) {
                rd_kafka_rdunlock(rk);
                rd_kafka_metadata_refresh_brokers(rk, NULL,
                                                  "lookup controller");
                return NULL;
        }

        rkb = rd_kafka_broker_find_by_nodeid0(rk, rk->rk_controllerid, state,
                                              rd_true);

        rd_kafka_rdunlock(rk);

        return rkb;
}
/**
 * @returns the current controller using cached metadata information,
 *          and only if the broker's state == \p state.
 *          The reference count is increased for the returned broker.
 *
 * @locks none
 * @locality any thread
 */
@returns the current controller using cached metadata information,
and only if the broker's state == \p state.
The reference count is increased for the returned broker.

@locks none
@locality any thread
@returns the current controller using cached metadata information and only if the broker ' s state == \ p state . The reference count is increased for the returned broker . @locks none @locality any thread
static rd_kafka_broker_t *rd_kafka_broker_controller_nowait (rd_kafka_t *rk,
                                                             int state) {
        rd_kafka_broker_t *rkb;
        rd_kafka_rdlock(rk);
        if (rk->rk_controllerid == -1) {
                rd_kafka_rdunlock(rk);
                rd_kafka_metadata_refresh_brokers(rk, NULL,
                                                  "lookup controller");
                return NULL;
        }
        rkb = rd_kafka_broker_find_by_nodeid0(rk, rk->rk_controllerid, state,
                                              rd_true);
        rd_kafka_rdunlock(rk);
        return rkb;
}
static rd_kafka_broker_t * rd_kafka_broker_controller_nowait ( rd_kafka_t * rk , int state ) { rd_kafka_broker_t * rkb ; rd_kafka_rdlock ( rk ) ; if ( rk -> rk_controllerid == -1 ) { rd_kafka_rdunlock ( rk ) ; rd_kafka_metadata_refresh_brokers ( rk , NULL , " " ) ; return NULL ; } rkb = rd_kafka_broker_find_by_nodeid0 ( rk , rk -> rk_controllerid , state , rd_true ) ; rd_kafka_rdunlock ( rk ) ; return rkb ; }
@returns the current controller using cached metadata information,
          and only if the broker's state == \p state.
@returns the current controller using cached metadata information and only if the broker ' s state == \ p state .