src/rdkafka_broker.c
static rd_kafka_broker_t *
rd_kafka_broker_random (rd_kafka_t *rk,
                        int state,
                        int (*filter) (rd_kafka_broker_t *rk, void *opaque),
                        void *opaque) {
        rd_kafka_broker_t *rkb, *good = NULL;
        int cnt = 0;

        TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {
                if (RD_KAFKA_BROKER_IS_LOGICAL(rkb))
                        continue;

                rd_kafka_broker_lock(rkb);
                if ((int)rkb->rkb_state == state &&
                    (!filter || !filter(rkb, opaque))) {
                        if (cnt < 1 || rd_jitter(0, cnt) < 1) {
                                if (good)
                                        rd_kafka_broker_destroy(good);
                                rd_kafka_broker_keep(rkb);
                                good = rkb;
                        }
                        cnt += 1;
                }
		rd_kafka_broker_unlock(rkb);
	}

        return good;
}
/**
 * @returns a random broker (with refcnt increased) with matching \p state
 *          and where the \p filter function returns 0.
 *
 * Uses reservoir sampling.
 *
 * @param filter is an optional callback used to filter out undesired brokers.
 *               The filter function should return 1 to filter out a broker,
 *               or 0 to keep it in the list of eligible brokers to return.
 *               rd_kafka_broker_lock() is held during the filter callback.
 *
 *
 * @locks rd_kafka_*lock() MUST be held
 * @locality any
 */
@returns a random broker (with refcnt increased) with matching \p state
and where the \p filter function returns 0.

Uses reservoir sampling.

@param filter is an optional callback used to filter out undesired brokers.
The filter function should return 1 to filter out a broker,
or 0 to keep it in the list of eligible brokers to return.
rd_kafka_broker_lock() is held during the filter callback.

@locks rd_kafka_*lock() MUST be held
@locality any
@returns a random broker ( with refcnt increased ) with matching \ p state and where the \ p filter function returns 0 . Uses reservoir sampling . @param filter is an optional callback used to filter out undesired brokers . The filter function should return 1 to filter out a broker or 0 to keep it in the list of eligible brokers to return . rd_kafka_broker_lock () is held during the filter callback . @locks rd_kafka_ * lock () MUST be held @locality any
static rd_kafka_broker_t *
rd_kafka_broker_random (rd_kafka_t *rk,
                        int state,
                        int (*filter) (rd_kafka_broker_t *rk, void *opaque),
                        void *opaque) {
        rd_kafka_broker_t *rkb, *good = NULL;
        int cnt = 0;
        TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {
                if (RD_KAFKA_BROKER_IS_LOGICAL(rkb))
                        continue;
                rd_kafka_broker_lock(rkb);
                if ((int)rkb->rkb_state == state &&
                    (!filter || !filter(rkb, opaque))) {
                        if (cnt < 1 || rd_jitter(0, cnt) < 1) {
                                if (good)
                                        rd_kafka_broker_destroy(good);
                                rd_kafka_broker_keep(rkb);
                                good = rkb;
                        }
                        cnt += 1;
                }
		rd_kafka_broker_unlock(rkb);
	}
        return good;
}
static rd_kafka_broker_t * rd_kafka_broker_random ( rd_kafka_t * rk , int state , int ( * filter ) ( rd_kafka_broker_t * rk , void * opaque ) , void * opaque ) { rd_kafka_broker_t * rkb , * good = NULL ; int cnt = 0 ; TAILQ_FOREACH ( rkb , & rk -> rk_brokers , rkb_link ) { if ( RD_KAFKA_BROKER_IS_LOGICAL ( rkb ) ) continue ; rd_kafka_broker_lock ( rkb ) ; if ( ( int ) rkb -> rkb_state == state && ( ! filter || ! filter ( rkb , opaque ) ) ) { if ( cnt < 1 || rd_jitter ( 0 , cnt ) < 1 ) { if ( good ) rd_kafka_broker_destroy ( good ) ; rd_kafka_broker_keep ( rkb ) ; good = rkb ; } cnt += 1 ; } rd_kafka_broker_unlock ( rkb ) ; } return good ; }
@returns a random broker (with refcnt increased) with matching \p state
          and where the \p filter function returns 0.
@returns a random broker ( with refcnt increased ) with matching \ p state and where the \ p filter function returns 0 .