src/rdkafka_broker.c
rd_kafka_broker_t *
rd_kafka_broker_controller_async (rd_kafka_t *rk, int state,
                                  rd_kafka_enq_once_t *eonce) {
        int version;
        do {
                rd_kafka_broker_t *rkb;

                version = rd_kafka_brokers_get_state_version(rk);

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

        } while (!rd_kafka_brokers_wait_state_change_async(rk, version, eonce));

        return NULL; /* eonce added to wait list */
}
/**
 * @returns the current controller using cached metadata information if
 *          available in state == \p state, with refcount increaesd.
 *
 *          Otherwise enqueues the \p eonce on the wait-controller queue
 *          which will be triggered on controller updates or broker state
 *          changes. It may also be triggered erroneously, so the caller
 *          should call rd_kafka_broker_controller_async() again when
 *          the eonce is triggered.
 *
 * @locks none
 * @locality any thread
 */
@returns the current controller using cached metadata information if
available in state == \p state, with refcount increaesd.

Otherwise enqueues the \p eonce on the wait-controller queue
which will be triggered on controller updates or broker state
changes. It may also be triggered erroneously, so the caller
should call rd_kafka_broker_controller_async() again when
the eonce is triggered.

@locks none
@locality any thread
@returns the current controller using cached metadata information if available in state == \ p state with refcount increaesd . Otherwise enqueues the \ p eonce on the wait - controller queue which will be triggered on controller updates or broker state changes . It may also be triggered erroneously so the caller should call rd_kafka_broker_controller_async () again when the eonce is triggered . @locks none @locality any thread
rd_kafka_broker_t *
rd_kafka_broker_controller_async (rd_kafka_t *rk, int state,
                                  rd_kafka_enq_once_t *eonce) {
        int version;
        do {
                rd_kafka_broker_t *rkb;
                version = rd_kafka_brokers_get_state_version(rk);
                rkb = rd_kafka_broker_controller_nowait(rk, state);
                if (rkb)
                        return rkb;
        } while (!rd_kafka_brokers_wait_state_change_async(rk, version, eonce));
        return NULL; 
}
rd_kafka_broker_t * rd_kafka_broker_controller_async ( rd_kafka_t * rk , int state , rd_kafka_enq_once_t * eonce ) { int version ; do { rd_kafka_broker_t * rkb ; version = rd_kafka_brokers_get_state_version ( rk ) ; rkb = rd_kafka_broker_controller_nowait ( rk , state ) ; if ( rkb ) return rkb ; } while ( ! rd_kafka_brokers_wait_state_change_async ( rk , version , eonce ) ) ; return NULL ; }
@returns the current controller using cached metadata information if
          available in state == \p state, with refcount increaesd.
@returns the current controller using cached metadata information if available in state == \ p state with refcount increaesd .
/* eonce added to wait list */