src/rdkafka_request.c
int rd_kafka_ProduceRequest (rd_kafka_broker_t *rkb, rd_kafka_toppar_t *rktp,
                             const rd_kafka_pid_t pid) {
        rd_kafka_buf_t *rkbuf;
        rd_kafka_itopic_t *rkt = rktp->rktp_rkt;
        size_t MessageSetSize = 0;
        int cnt;
        rd_ts_t now;
        int64_t first_msg_timeout;
        int tmout;

        /**
         * Create ProduceRequest with as many messages from the toppar
         * transmit queue as possible.
         */
        rkbuf = rd_kafka_msgset_create_ProduceRequest(rkb, rktp,
                                                      &rktp->rktp_xmit_msgq,
                                                      pid, &MessageSetSize);
        if (unlikely(!rkbuf))
                return 0;

        cnt = rd_kafka_msgq_len(&rkbuf->rkbuf_batch.msgq);
        rd_dassert(cnt > 0);

        rd_avg_add(&rktp->rktp_rkt->rkt_avg_batchcnt, (int64_t)cnt);
        rd_avg_add(&rktp->rktp_rkt->rkt_avg_batchsize, (int64_t)MessageSetSize);

        if (!rkt->rkt_conf.required_acks)
                rkbuf->rkbuf_flags |= RD_KAFKA_OP_F_NO_RESPONSE;

        /* Use timeout from first message in batch */
        now = rd_clock();
        first_msg_timeout = (rd_kafka_msgq_first(&rkbuf->rkbuf_batch.msgq)->
                             rkm_ts_timeout - now) / 1000;

        if (unlikely(first_msg_timeout <= 0)) {
                /* Message has already timed out, allow 100 ms
                 * to produce anyway */
                tmout = 100;
        } else {
                tmout = (int)RD_MIN(INT_MAX, first_msg_timeout);
        }

        /* Set absolute timeout (including retries), the
         * effective timeout for this specific request will be
         * capped by socket.timeout.ms */
        rd_kafka_buf_set_abs_timeout(rkbuf, tmout, now);

        rd_kafka_broker_buf_enq_replyq(rkb, rkbuf,
                                       RD_KAFKA_NO_REPLYQ,
                                       rd_kafka_handle_Produce, NULL);

        return cnt;
}
/**
 * @brief Send ProduceRequest for messages in toppar queue.
 *
 * @returns the number of messages included, or 0 on error / no messages.
 *
 * @locality broker thread
 */
@brief Send ProduceRequest for messages in toppar queue.
@returns the number of messages included, or 0 on error / no messages.

@locality broker thread
@brief Send ProduceRequest for messages in toppar queue . @returns the number of messages included or 0 on error / no messages . @locality broker thread
int rd_kafka_ProduceRequest (rd_kafka_broker_t *rkb, rd_kafka_toppar_t *rktp,
                             const rd_kafka_pid_t pid) {
        rd_kafka_buf_t *rkbuf;
        rd_kafka_itopic_t *rkt = rktp->rktp_rkt;
        size_t MessageSetSize = 0;
        int cnt;
        rd_ts_t now;
        int64_t first_msg_timeout;
        int tmout;
        rkbuf = rd_kafka_msgset_create_ProduceRequest(rkb, rktp,
                                                      &rktp->rktp_xmit_msgq,
                                                      pid, &MessageSetSize);
        if (unlikely(!rkbuf))
                return 0;
        cnt = rd_kafka_msgq_len(&rkbuf->rkbuf_batch.msgq);
        rd_dassert(cnt > 0);
        rd_avg_add(&rktp->rktp_rkt->rkt_avg_batchcnt, (int64_t)cnt);
        rd_avg_add(&rktp->rktp_rkt->rkt_avg_batchsize, (int64_t)MessageSetSize);
        if (!rkt->rkt_conf.required_acks)
                rkbuf->rkbuf_flags |= RD_KAFKA_OP_F_NO_RESPONSE;
        now = rd_clock();
        first_msg_timeout = (rd_kafka_msgq_first(&rkbuf->rkbuf_batch.msgq)->
                             rkm_ts_timeout - now) / 1000;
        if (unlikely(first_msg_timeout <= 0)) {
                tmout = 100;
        } else {
                tmout = (int)RD_MIN(INT_MAX, first_msg_timeout);
        }
        rd_kafka_buf_set_abs_timeout(rkbuf, tmout, now);
        rd_kafka_broker_buf_enq_replyq(rkb, rkbuf,
                                       RD_KAFKA_NO_REPLYQ,
                                       rd_kafka_handle_Produce, NULL);
        return cnt;
}
int rd_kafka_ProduceRequest ( rd_kafka_broker_t * rkb , rd_kafka_toppar_t * rktp , const rd_kafka_pid_t pid ) { rd_kafka_buf_t * rkbuf ; rd_kafka_itopic_t * rkt = rktp -> rktp_rkt ; size_t MessageSetSize = 0 ; int cnt ; rd_ts_t now ; int64_t first_msg_timeout ; int tmout ; rkbuf = rd_kafka_msgset_create_ProduceRequest ( rkb , rktp , & rktp -> rktp_xmit_msgq , pid , & MessageSetSize ) ; if ( unlikely ( ! rkbuf ) ) return 0 ; cnt = rd_kafka_msgq_len ( & rkbuf -> rkbuf_batch . msgq ) ; rd_dassert ( cnt > 0 ) ; rd_avg_add ( & rktp -> rktp_rkt -> rkt_avg_batchcnt , ( int64_t ) cnt ) ; rd_avg_add ( & rktp -> rktp_rkt -> rkt_avg_batchsize , ( int64_t ) MessageSetSize ) ; if ( ! rkt -> rkt_conf . required_acks ) rkbuf -> rkbuf_flags |= RD_KAFKA_OP_F_NO_RESPONSE ; now = rd_clock ( ) ; first_msg_timeout = ( rd_kafka_msgq_first ( & rkbuf -> rkbuf_batch . msgq ) -> rkm_ts_timeout - now ) / 1000 ; if ( unlikely ( first_msg_timeout <= 0 ) ) { tmout = 100 ; } else { tmout = ( int ) RD_MIN ( INT_MAX , first_msg_timeout ) ; } rd_kafka_buf_set_abs_timeout ( rkbuf , tmout , now ) ; rd_kafka_broker_buf_enq_replyq ( rkb , rkbuf , RD_KAFKA_NO_REPLYQ , rd_kafka_handle_Produce , NULL ) ; return cnt ; }
@brief Send ProduceRequest for messages in toppar queue.
@brief Send ProduceRequest for messages in toppar queue .
/**
         * Create ProduceRequest with as many messages from the toppar
         * transmit queue as possible.
         */ /* Use timeout from first message in batch */ /* Message has already timed out, allow 100 ms
                 * to produce anyway */ /* Set absolute timeout (including retries), the
         * effective timeout for this specific request will be
         * capped by socket.timeout.ms */