src/rdkafka_broker.c
rd_kafka_broker_t *rd_kafka_broker_controller (rd_kafka_t *rk, int state,
                                               rd_ts_t abs_timeout) {

        while (1) {
                int version = rd_kafka_brokers_get_state_version(rk);
                rd_kafka_broker_t *rkb;
                int remains_ms;

                rkb = rd_kafka_broker_controller_nowait(rk, state);
                if (rkb)
                        return rkb;

                remains_ms = rd_timeout_remains(abs_timeout);
                if (rd_timeout_expired(remains_ms))
                        return NULL;

                rd_kafka_brokers_wait_state_change(rk, version, remains_ms);
        }
}
/**
 * @returns the current controller using cached metadata information,
 *          blocking up to \p abs_timeout for the controller to be known
 *          and to reach 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,
blocking up to \p abs_timeout for the controller to be known
and to reach 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 blocking up to \ p abs_timeout for the controller to be known and to reach state == \ p state . The reference count is increased for the returned broker . @locks none @locality any thread
rd_kafka_broker_t *rd_kafka_broker_controller (rd_kafka_t *rk, int state,
                                               rd_ts_t abs_timeout) {
        while (1) {
                int version = rd_kafka_brokers_get_state_version(rk);
                rd_kafka_broker_t *rkb;
                int remains_ms;
                rkb = rd_kafka_broker_controller_nowait(rk, state);
                if (rkb)
                        return rkb;
                remains_ms = rd_timeout_remains(abs_timeout);
                if (rd_timeout_expired(remains_ms))
                        return NULL;
                rd_kafka_brokers_wait_state_change(rk, version, remains_ms);
        }
}
rd_kafka_broker_t * rd_kafka_broker_controller ( rd_kafka_t * rk , int state , rd_ts_t abs_timeout ) { while ( 1 ) { int version = rd_kafka_brokers_get_state_version ( rk ) ; rd_kafka_broker_t * rkb ; int remains_ms ; rkb = rd_kafka_broker_controller_nowait ( rk , state ) ; if ( rkb ) return rkb ; remains_ms = rd_timeout_remains ( abs_timeout ) ; if ( rd_timeout_expired ( remains_ms ) ) return NULL ; rd_kafka_brokers_wait_state_change ( rk , version , remains_ms ) ; } }
@returns the current controller using cached metadata information,
          blocking up to \p abs_timeout for the controller to be known
          and to reach state == \p state.
@returns the current controller using cached metadata information blocking up to \ p abs_timeout for the controller to be known and to reach state == \ p state .