src/rdkafka_broker.c
static rd_ts_t rd_kafka_broker_toppars_serve (rd_kafka_broker_t *rkb) {
        rd_kafka_toppar_t *rktp, *rktp_tmp;
        rd_ts_t min_backoff = RD_TS_MAX;

        TAILQ_FOREACH_SAFE(rktp, &rkb->rkb_toppars, rktp_rkblink, rktp_tmp) {
                rd_ts_t backoff;

                /* Serve toppar to update desired rktp state */
                backoff = rd_kafka_broker_consumer_toppar_serve(rkb, rktp);
                if (backoff < min_backoff)
                        min_backoff = backoff;
        }

        return min_backoff;
}
/**
 * @brief Serve the toppar's assigned to this broker.
 *
 * @returns the minimum Fetch backoff time (abs timestamp) for the
 *          partitions to fetch.
 *
 * @locality broker thread
 */
@brief Serve the toppar's assigned to this broker.
@returns the minimum Fetch backoff time (abs timestamp) for the
partitions to fetch.

@locality broker thread
@brief Serve the toppar ' s assigned to this broker . @returns the minimum Fetch backoff time ( abs timestamp ) for the partitions to fetch . @locality broker thread
static rd_ts_t rd_kafka_broker_toppars_serve (rd_kafka_broker_t *rkb) {
        rd_kafka_toppar_t *rktp, *rktp_tmp;
        rd_ts_t min_backoff = RD_TS_MAX;
        TAILQ_FOREACH_SAFE(rktp, &rkb->rkb_toppars, rktp_rkblink, rktp_tmp) {
                rd_ts_t backoff;
                backoff = rd_kafka_broker_consumer_toppar_serve(rkb, rktp);
                if (backoff < min_backoff)
                        min_backoff = backoff;
        }
        return min_backoff;
}
static rd_ts_t rd_kafka_broker_toppars_serve ( rd_kafka_broker_t * rkb ) { rd_kafka_toppar_t * rktp , * rktp_tmp ; rd_ts_t min_backoff = RD_TS_MAX ; TAILQ_FOREACH_SAFE ( rktp , & rkb -> rkb_toppars , rktp_rkblink , rktp_tmp ) { rd_ts_t backoff ; backoff = rd_kafka_broker_consumer_toppar_serve ( rkb , rktp ) ; if ( backoff < min_backoff ) min_backoff = backoff ; } return min_backoff ; }
@brief Serve the toppar's assigned to this broker.
@brief Serve the toppar ' s assigned to this broker .
/* Serve toppar to update desired rktp state */