src/rdkafka_request.c
static rd_kafka_resp_err_t
rd_kafka_handle_Produce_parse (rd_kafka_broker_t *rkb,
                               rd_kafka_toppar_t *rktp,
                               rd_kafka_buf_t *rkbuf,
                               rd_kafka_buf_t *request,
                               struct rd_kafka_Produce_result *result) {
        int32_t TopicArrayCnt;
        int32_t PartitionArrayCnt;
        struct {
                int32_t Partition;
                int16_t ErrorCode;
                int64_t Offset;
        } hdr;
        const int log_decode_errors = LOG_ERR;
        int64_t log_start_offset = -1;

        rd_kafka_buf_read_i32(rkbuf, &TopicArrayCnt);
        if (TopicArrayCnt != 1)
                goto err;

        /* Since we only produce to one single topic+partition in each
         * request we assume that the reply only contains one topic+partition
         * and that it is the same that we requested.
         * If not the broker is buggy. */
        rd_kafka_buf_skip_str(rkbuf);
        rd_kafka_buf_read_i32(rkbuf, &PartitionArrayCnt);

        if (PartitionArrayCnt != 1)
                goto err;

        rd_kafka_buf_read_i32(rkbuf, &hdr.Partition);
        rd_kafka_buf_read_i16(rkbuf, &hdr.ErrorCode);
        rd_kafka_buf_read_i64(rkbuf, &hdr.Offset);

        result->offset = hdr.Offset;

        result->timestamp = -1;
        if (request->rkbuf_reqhdr.ApiVersion >= 2)
                rd_kafka_buf_read_i64(rkbuf, &result->timestamp);

        if (request->rkbuf_reqhdr.ApiVersion >= 5)
                rd_kafka_buf_read_i64(rkbuf, &log_start_offset);

        if (request->rkbuf_reqhdr.ApiVersion >= 1) {
                int32_t Throttle_Time;
                rd_kafka_buf_read_i32(rkbuf, &Throttle_Time);

                rd_kafka_op_throttle_time(rkb, rkb->rkb_rk->rk_rep,
                                          Throttle_Time);
        }


        return hdr.ErrorCode;

 err_parse:
        return rkbuf->rkbuf_err;
 err:
        return RD_KAFKA_RESP_ERR__BAD_MSG;
}
/**
 * @brief Parses a Produce reply.
 * @returns 0 on success or an error code on failure.
 * @locality broker thread
 */
@brief Parses a Produce reply.
@returns 0 on success or an error code on failure.
@locality broker thread
@brief Parses a Produce reply . @returns 0 on success or an error code on failure . @locality broker thread
static rd_kafka_resp_err_t
rd_kafka_handle_Produce_parse (rd_kafka_broker_t *rkb,
                               rd_kafka_toppar_t *rktp,
                               rd_kafka_buf_t *rkbuf,
                               rd_kafka_buf_t *request,
                               struct rd_kafka_Produce_result *result) {
        int32_t TopicArrayCnt;
        int32_t PartitionArrayCnt;
        struct {
                int32_t Partition;
                int16_t ErrorCode;
                int64_t Offset;
        } hdr;
        const int log_decode_errors = LOG_ERR;
        int64_t log_start_offset = -1;
        rd_kafka_buf_read_i32(rkbuf, &TopicArrayCnt);
        if (TopicArrayCnt != 1)
                goto err;
        rd_kafka_buf_skip_str(rkbuf);
        rd_kafka_buf_read_i32(rkbuf, &PartitionArrayCnt);
        if (PartitionArrayCnt != 1)
                goto err;
        rd_kafka_buf_read_i32(rkbuf, &hdr.Partition);
        rd_kafka_buf_read_i16(rkbuf, &hdr.ErrorCode);
        rd_kafka_buf_read_i64(rkbuf, &hdr.Offset);
        result->offset = hdr.Offset;
        result->timestamp = -1;
        if (request->rkbuf_reqhdr.ApiVersion >= 2)
                rd_kafka_buf_read_i64(rkbuf, &result->timestamp);
        if (request->rkbuf_reqhdr.ApiVersion >= 5)
                rd_kafka_buf_read_i64(rkbuf, &log_start_offset);
        if (request->rkbuf_reqhdr.ApiVersion >= 1) {
                int32_t Throttle_Time;
                rd_kafka_buf_read_i32(rkbuf, &Throttle_Time);
                rd_kafka_op_throttle_time(rkb, rkb->rkb_rk->rk_rep,
                                          Throttle_Time);
        }
        return hdr.ErrorCode;
 err_parse:
        return rkbuf->rkbuf_err;
 err:
        return RD_KAFKA_RESP_ERR__BAD_MSG;
}
static rd_kafka_resp_err_t rd_kafka_handle_Produce_parse ( rd_kafka_broker_t * rkb , rd_kafka_toppar_t * rktp , rd_kafka_buf_t * rkbuf , rd_kafka_buf_t * request , struct rd_kafka_Produce_result * result ) { int32_t TopicArrayCnt ; int32_t PartitionArrayCnt ; struct { int32_t Partition ; int16_t ErrorCode ; int64_t Offset ; } hdr ; const int log_decode_errors = LOG_ERR ; int64_t log_start_offset = -1 ; rd_kafka_buf_read_i32 ( rkbuf , & TopicArrayCnt ) ; if ( TopicArrayCnt != 1 ) goto err ; rd_kafka_buf_skip_str ( rkbuf ) ; rd_kafka_buf_read_i32 ( rkbuf , & PartitionArrayCnt ) ; if ( PartitionArrayCnt != 1 ) goto err ; rd_kafka_buf_read_i32 ( rkbuf , & hdr . Partition ) ; rd_kafka_buf_read_i16 ( rkbuf , & hdr . ErrorCode ) ; rd_kafka_buf_read_i64 ( rkbuf , & hdr . Offset ) ; result -> offset = hdr . Offset ; result -> timestamp = -1 ; if ( request -> rkbuf_reqhdr . ApiVersion >= 2 ) rd_kafka_buf_read_i64 ( rkbuf , & result -> timestamp ) ; if ( request -> rkbuf_reqhdr . ApiVersion >= 5 ) rd_kafka_buf_read_i64 ( rkbuf , & log_start_offset ) ; if ( request -> rkbuf_reqhdr . ApiVersion >= 1 ) { int32_t Throttle_Time ; rd_kafka_buf_read_i32 ( rkbuf , & Throttle_Time ) ; rd_kafka_op_throttle_time ( rkb , rkb -> rkb_rk -> rk_rep , Throttle_Time ) ; } return hdr . ErrorCode ; err_parse : return rkbuf -> rkbuf_err ; err : return RD_KAFKA_RESP_ERR__BAD_MSG ; }
@brief Parses a Produce reply.
@brief Parses a Produce reply .
/* Since we only produce to one single topic+partition in each
         * request we assume that the reply only contains one topic+partition
         * and that it is the same that we requested.
         * If not the broker is buggy. */