src/rdkafka_broker.c
static void rd_kafka_broker_retry_bufs_move (rd_kafka_broker_t *rkb) {
	rd_ts_t now = rd_clock();
	rd_kafka_buf_t *rkbuf;
        int cnt = 0;

	while ((rkbuf = TAILQ_FIRST(&rkb->rkb_retrybufs.rkbq_bufs))) {
		if (rkbuf->rkbuf_ts_retry > now)
			break;

		rd_kafka_bufq_deq(&rkb->rkb_retrybufs, rkbuf);

                rd_kafka_broker_buf_enq0(rkb, rkbuf);
                cnt++;
	}

        if (cnt > 0)
                rd_rkb_dbg(rkb, BROKER, "RETRY",
                           "Moved %d retry buffer(s) to output queue", cnt);
}
/**
 * Move buffers that have expired their retry backoff time from the 
 * retry queue to the outbuf.
 */
Move buffers that have expired their retry backoff time from the
retry queue to the outbuf.
Move buffers that have expired their retry backoff time from the retry queue to the outbuf .
static void rd_kafka_broker_retry_bufs_move (rd_kafka_broker_t *rkb) {
	rd_ts_t now = rd_clock();
	rd_kafka_buf_t *rkbuf;
        int cnt = 0;
	while ((rkbuf = TAILQ_FIRST(&rkb->rkb_retrybufs.rkbq_bufs))) {
		if (rkbuf->rkbuf_ts_retry > now)
			break;
		rd_kafka_bufq_deq(&rkb->rkb_retrybufs, rkbuf);
                rd_kafka_broker_buf_enq0(rkb, rkbuf);
                cnt++;
	}
        if (cnt > 0)
                rd_rkb_dbg(rkb, BROKER, "RETRY",
                           "Moved %d retry buffer(s) to output queue", cnt);
}
static void rd_kafka_broker_retry_bufs_move ( rd_kafka_broker_t * rkb ) { rd_ts_t now = rd_clock ( ) ; rd_kafka_buf_t * rkbuf ; int cnt = 0 ; while ( ( rkbuf = TAILQ_FIRST ( & rkb -> rkb_retrybufs . rkbq_bufs ) ) ) { if ( rkbuf -> rkbuf_ts_retry > now ) break ; rd_kafka_bufq_deq ( & rkb -> rkb_retrybufs , rkbuf ) ; rd_kafka_broker_buf_enq0 ( rkb , rkbuf ) ; cnt ++ ; } if ( cnt > 0 ) rd_rkb_dbg ( rkb , BROKER , " " , " " , cnt ) ; }
Move buffers that have expired their retry backoff time from the
 retry queue to the outbuf.
Move buffers that have expired their retry backoff time from the retry queue to the outbuf .