src/rdkafka_broker.c
void rd_kafka_dr_msgq (rd_kafka_itopic_t *rkt,
		       rd_kafka_msgq_t *rkmq,
                       rd_kafka_resp_err_t err) {
        rd_kafka_t *rk = rkt->rkt_rk;

	if (unlikely(rd_kafka_msgq_len(rkmq) == 0))
	    return;

        /* Call on_acknowledgement() interceptors */
        rd_kafka_interceptors_on_acknowledgement_queue(rk, rkmq, err);

        if ((rk->rk_conf.enabled_events & RD_KAFKA_EVENT_DR) &&
	    (!rk->rk_conf.dr_err_only || err)) {
		/* Pass all messages to application thread in one op. */
		rd_kafka_op_t *rko;

		rko = rd_kafka_op_new(RD_KAFKA_OP_DR);
		rko->rko_err = err;
		rko->rko_u.dr.s_rkt = rd_kafka_topic_keep(rkt);
		rd_kafka_msgq_init(&rko->rko_u.dr.msgq);

		/* Move all messages to op's msgq */
		rd_kafka_msgq_move(&rko->rko_u.dr.msgq, rkmq);

		rd_kafka_q_enq(rk->rk_rep, rko);

	} else {
		/* No delivery report callback. */

                /* Destroy the messages right away. */
                rd_kafka_msgq_purge(rk, rkmq);
	}
}
/**
 * @brief Propagate delivery report for entire message queue.
 *
 * @param err The error which will be set on each message.
 * @param status The status which will be set on each message.
 *
 * To avoid extra iterations, the \p err and \p status are set on
 * the message as they are popped off the OP_DR msgq in rd_kafka_poll() et.al
 */
@brief Propagate delivery report for entire message queue.
@param err The error which will be set on each message.
@param status The status which will be set on each message.

To avoid extra iterations, the \p err and \p status are set on
the message as they are popped off the OP_DR msgq in rd_kafka_poll() et.al
@brief Propagate delivery report for entire message queue . @param err The error which will be set on each message . @param status The status which will be set on each message . To avoid extra iterations the \ p err and \ p status are set on the message as they are popped off the OP_DR msgq in rd_kafka_poll () et . al
void rd_kafka_dr_msgq (rd_kafka_itopic_t *rkt,
		       rd_kafka_msgq_t *rkmq,
                       rd_kafka_resp_err_t err) {
        rd_kafka_t *rk = rkt->rkt_rk;
	if (unlikely(rd_kafka_msgq_len(rkmq) == 0))
	    return;
        rd_kafka_interceptors_on_acknowledgement_queue(rk, rkmq, err);
        if ((rk->rk_conf.enabled_events & RD_KAFKA_EVENT_DR) &&
	    (!rk->rk_conf.dr_err_only || err)) {
		rd_kafka_op_t *rko;
		rko = rd_kafka_op_new(RD_KAFKA_OP_DR);
		rko->rko_err = err;
		rko->rko_u.dr.s_rkt = rd_kafka_topic_keep(rkt);
		rd_kafka_msgq_init(&rko->rko_u.dr.msgq);
		rd_kafka_msgq_move(&rko->rko_u.dr.msgq, rkmq);
		rd_kafka_q_enq(rk->rk_rep, rko);
	} else {
                rd_kafka_msgq_purge(rk, rkmq);
	}
}
void rd_kafka_dr_msgq ( rd_kafka_itopic_t * rkt , rd_kafka_msgq_t * rkmq , rd_kafka_resp_err_t err ) { rd_kafka_t * rk = rkt -> rkt_rk ; if ( unlikely ( rd_kafka_msgq_len ( rkmq ) == 0 ) ) return ; rd_kafka_interceptors_on_acknowledgement_queue ( rk , rkmq , err ) ; if ( ( rk -> rk_conf . enabled_events & RD_KAFKA_EVENT_DR ) && ( ! rk -> rk_conf . dr_err_only || err ) ) { rd_kafka_op_t * rko ; rko = rd_kafka_op_new ( RD_KAFKA_OP_DR ) ; rko -> rko_err = err ; rko -> rko_u . dr . s_rkt = rd_kafka_topic_keep ( rkt ) ; rd_kafka_msgq_init ( & rko -> rko_u . dr . msgq ) ; rd_kafka_msgq_move ( & rko -> rko_u . dr . msgq , rkmq ) ; rd_kafka_q_enq ( rk -> rk_rep , rko ) ; } else { rd_kafka_msgq_purge ( rk , rkmq ) ; } }
@brief Propagate delivery report for entire message queue.
@brief Propagate delivery report for entire message queue .
/* Call on_acknowledgement() interceptors */ /* Pass all messages to application thread in one op. */ /* Move all messages to op's msgq */ /* No delivery report callback. */ /* Destroy the messages right away. */