src/rdkafka_broker.c
static int rd_kafka_broker_produce_toppars (rd_kafka_broker_t *rkb,
                                            rd_ts_t now,
                                            rd_ts_t *next_wakeup,
                                            int do_timeout_scan) {
        rd_kafka_toppar_t *rktp;
        int cnt = 0;
        rd_ts_t ret_next_wakeup = *next_wakeup;
        rd_kafka_pid_t pid = RD_KAFKA_PID_INITIALIZER;

        /* Round-robin serve each toppar. */
        rktp = rkb->rkb_active_toppar_next;
        if (unlikely(!rktp))
                return 0;

        if (rd_kafka_is_idempotent(rkb->rkb_rk)) {
                /* Idempotent producer: get a copy of the current pid. */
                pid = rd_kafka_idemp_get_pid(rkb->rkb_rk);

                /* If we don't have a valid pid return immedatiely,
                 * unless the per-partition timeout scan needs to run.
                 * The broker threads are woken up when a PID is acquired. */
                if (!rd_kafka_pid_valid(pid) && !do_timeout_scan)
                        return 0;
        }

        do {
                rd_ts_t this_next_wakeup = ret_next_wakeup;

                /* Try producing toppar */
                cnt += rd_kafka_toppar_producer_serve(
                        rkb, rktp, pid, now, &this_next_wakeup,
                        do_timeout_scan);

                if (this_next_wakeup < ret_next_wakeup)
                        ret_next_wakeup = this_next_wakeup;

        } while ((rktp = CIRCLEQ_LOOP_NEXT(&rkb->
                                           rkb_active_toppars,
                                           rktp, rktp_activelink)) !=
                 rkb->rkb_active_toppar_next);

        /* Update next starting toppar to produce in round-robin list. */
        rd_kafka_broker_active_toppar_next(
                rkb,
                CIRCLEQ_LOOP_NEXT(&rkb->rkb_active_toppars,
                                  rktp, rktp_activelink));

        *next_wakeup = ret_next_wakeup;

        return cnt;
}
/**
 * @brief Produce from all toppars assigned to this broker.
 *
 * @param next_wakeup is updated if the next IO/ops timeout should be
 *                    less than the input value.
 *
 * @returns the total number of messages produced.
 */
@brief Produce from all toppars assigned to this broker.
@param next_wakeup is updated if the next IO/ops timeout should be
less than the input value.

@returns the total number of messages produced.
@brief Produce from all toppars assigned to this broker . @param next_wakeup is updated if the next IO / ops timeout should be less than the input value . @returns the total number of messages produced .
static int rd_kafka_broker_produce_toppars (rd_kafka_broker_t *rkb,
                                            rd_ts_t now,
                                            rd_ts_t *next_wakeup,
                                            int do_timeout_scan) {
        rd_kafka_toppar_t *rktp;
        int cnt = 0;
        rd_ts_t ret_next_wakeup = *next_wakeup;
        rd_kafka_pid_t pid = RD_KAFKA_PID_INITIALIZER;
        rktp = rkb->rkb_active_toppar_next;
        if (unlikely(!rktp))
                return 0;
        if (rd_kafka_is_idempotent(rkb->rkb_rk)) {
                pid = rd_kafka_idemp_get_pid(rkb->rkb_rk);
                if (!rd_kafka_pid_valid(pid) && !do_timeout_scan)
                        return 0;
        }
        do {
                rd_ts_t this_next_wakeup = ret_next_wakeup;
                cnt += rd_kafka_toppar_producer_serve(
                        rkb, rktp, pid, now, &this_next_wakeup,
                        do_timeout_scan);
                if (this_next_wakeup < ret_next_wakeup)
                        ret_next_wakeup = this_next_wakeup;
        } while ((rktp = CIRCLEQ_LOOP_NEXT(&rkb->
                                           rkb_active_toppars,
                                           rktp, rktp_activelink)) !=
                 rkb->rkb_active_toppar_next);
        rd_kafka_broker_active_toppar_next(
                rkb,
                CIRCLEQ_LOOP_NEXT(&rkb->rkb_active_toppars,
                                  rktp, rktp_activelink));
        *next_wakeup = ret_next_wakeup;
        return cnt;
}
static int rd_kafka_broker_produce_toppars ( rd_kafka_broker_t * rkb , rd_ts_t now , rd_ts_t * next_wakeup , int do_timeout_scan ) { rd_kafka_toppar_t * rktp ; int cnt = 0 ; rd_ts_t ret_next_wakeup = * next_wakeup ; rd_kafka_pid_t pid = RD_KAFKA_PID_INITIALIZER ; rktp = rkb -> rkb_active_toppar_next ; if ( unlikely ( ! rktp ) ) return 0 ; if ( rd_kafka_is_idempotent ( rkb -> rkb_rk ) ) { pid = rd_kafka_idemp_get_pid ( rkb -> rkb_rk ) ; if ( ! rd_kafka_pid_valid ( pid ) && ! do_timeout_scan ) return 0 ; } do { rd_ts_t this_next_wakeup = ret_next_wakeup ; cnt += rd_kafka_toppar_producer_serve ( rkb , rktp , pid , now , & this_next_wakeup , do_timeout_scan ) ; if ( this_next_wakeup < ret_next_wakeup ) ret_next_wakeup = this_next_wakeup ; } while ( ( rktp = CIRCLEQ_LOOP_NEXT ( & rkb -> rkb_active_toppars , rktp , rktp_activelink ) ) != rkb -> rkb_active_toppar_next ) ; rd_kafka_broker_active_toppar_next ( rkb , CIRCLEQ_LOOP_NEXT ( & rkb -> rkb_active_toppars , rktp , rktp_activelink ) ) ; * next_wakeup = ret_next_wakeup ; return cnt ; }
@brief Produce from all toppars assigned to this broker.
@brief Produce from all toppars assigned to this broker .
/* Round-robin serve each toppar. */ /* Idempotent producer: get a copy of the current pid. */ /* If we don't have a valid pid return immedatiely,
                 * unless the per-partition timeout scan needs to run.
                 * The broker threads are woken up when a PID is acquired. */ /* Try producing toppar */ /* Update next starting toppar to produce in round-robin list. */