src/rdkafka_request.c
static void rd_kafka_handle_Produce (rd_kafka_t *rk,
                                     rd_kafka_broker_t *rkb,
                                     rd_kafka_resp_err_t err,
                                     rd_kafka_buf_t *reply,
                                     rd_kafka_buf_t *request,
                                     void *opaque) {
        rd_kafka_msgbatch_t *batch = &request->rkbuf_batch;
        rd_kafka_toppar_t *rktp = rd_kafka_toppar_s2i(batch->s_rktp);
        struct rd_kafka_Produce_result result = {
                .offset = RD_KAFKA_OFFSET_INVALID,
                .timestamp = -1
        };

        /* Unit test interface: inject errors */
        if (unlikely(rk->rk_conf.ut.handle_ProduceResponse != NULL)) {
                err = rk->rk_conf.ut.handle_ProduceResponse(
                        rkb->rkb_rk,
                        rkb->rkb_nodeid,
                        batch->first_msgid,
                        err);
        }

        /* Parse Produce reply (unless the request errored) */
        if (!err && reply)
                err = rd_kafka_handle_Produce_parse(rkb, rktp,
                                                    reply, request,
                                                    &result);

        rd_kafka_msgbatch_handle_Produce_result(rkb, batch, err,
                                                &result, request);
}
/**
 * @brief Handle ProduceResponse
 *
 * @param reply is NULL when `acks=0` and on various local errors.
 *
 * @remark ProduceRequests are never retried, retriable errors are
 *         instead handled by re-enqueuing the request's messages back
 *         on the partition queue to have a new ProduceRequest constructed
 *         eventually.
 *
 * @warning May be called on the old leader thread. Lock rktp appropriately!
 *
 * @locality broker thread (but not necessarily the leader broker thread)
 */
@brief Handle ProduceResponse
@param reply is NULL when `acks=0` and on various local errors.

@remark ProduceRequests are never retried, retriable errors are
instead handled by re-enqueuing the request's messages back
on the partition queue to have a new ProduceRequest constructed
eventually.

@warning May be called on the old leader thread. Lock rktp appropriately!

@locality broker thread (but not necessarily the leader broker thread)
@brief Handle ProduceResponse @param reply is NULL when ` acks = 0 ` and on various local errors . @remark ProduceRequests are never retried retriable errors are instead handled by re - enqueuing the request ' s messages back on the partition queue to have a new ProduceRequest constructed eventually . @warning May be called on the old leader thread . Lock rktp appropriately! @locality broker thread ( but not necessarily the leader broker thread )
static void rd_kafka_handle_Produce (rd_kafka_t *rk,
                                     rd_kafka_broker_t *rkb,
                                     rd_kafka_resp_err_t err,
                                     rd_kafka_buf_t *reply,
                                     rd_kafka_buf_t *request,
                                     void *opaque) {
        rd_kafka_msgbatch_t *batch = &request->rkbuf_batch;
        rd_kafka_toppar_t *rktp = rd_kafka_toppar_s2i(batch->s_rktp);
        struct rd_kafka_Produce_result result = {
                .offset = RD_KAFKA_OFFSET_INVALID,
                .timestamp = -1
        };
        if (unlikely(rk->rk_conf.ut.handle_ProduceResponse != NULL)) {
                err = rk->rk_conf.ut.handle_ProduceResponse(
                        rkb->rkb_rk,
                        rkb->rkb_nodeid,
                        batch->first_msgid,
                        err);
        }
        if (!err && reply)
                err = rd_kafka_handle_Produce_parse(rkb, rktp,
                                                    reply, request,
                                                    &result);
        rd_kafka_msgbatch_handle_Produce_result(rkb, batch, err,
                                                &result, request);
}
static void rd_kafka_handle_Produce ( rd_kafka_t * rk , rd_kafka_broker_t * rkb , rd_kafka_resp_err_t err , rd_kafka_buf_t * reply , rd_kafka_buf_t * request , void * opaque ) { rd_kafka_msgbatch_t * batch = & request -> rkbuf_batch ; rd_kafka_toppar_t * rktp = rd_kafka_toppar_s2i ( batch -> s_rktp ) ; struct rd_kafka_Produce_result result = { . offset = RD_KAFKA_OFFSET_INVALID , . timestamp = -1 } ; if ( unlikely ( rk -> rk_conf . ut . handle_ProduceResponse != NULL ) ) { err = rk -> rk_conf . ut . handle_ProduceResponse ( rkb -> rkb_rk , rkb -> rkb_nodeid , batch -> first_msgid , err ) ; } if ( ! err && reply ) err = rd_kafka_handle_Produce_parse ( rkb , rktp , reply , request , & result ) ; rd_kafka_msgbatch_handle_Produce_result ( rkb , batch , err , & result , request ) ; }
@brief Handle ProduceResponse
 @param reply is NULL when `acks=0` and on various local errors.
@brief Handle ProduceResponse @param reply is NULL when ` acks = 0 ` and on various local errors .
/* Unit test interface: inject errors */ /* Parse Produce reply (unless the request errored) */