src/rdkafka_request.c
void
rd_kafka_handle_InitProducerId (rd_kafka_t *rk,
                                rd_kafka_broker_t *rkb,
                                rd_kafka_resp_err_t err,
                                rd_kafka_buf_t *rkbuf,
                                rd_kafka_buf_t *request,
                                void *opaque) {
        const int log_decode_errors = LOG_ERR;
        int16_t error_code;
        rd_kafka_pid_t pid;

        if (err)
                goto err;

        rd_kafka_buf_read_throttle_time(rkbuf);

        rd_kafka_buf_read_i16(rkbuf, &error_code);
        if ((err = error_code))
                goto err;

        rd_kafka_buf_read_i64(rkbuf, &pid.id);
        rd_kafka_buf_read_i16(rkbuf, &pid.epoch);

        rd_kafka_idemp_pid_update(rkb, pid);

        return;

 err_parse:
        err = rkbuf->rkbuf_err;
 err:
        /* Retries are performed by idempotence state handler */
        rd_kafka_idemp_request_pid_failed(rkb, err);
}
/**
 * @brief Parses and handles an InitProducerId reply.
 *
 * @returns 0 on success, else an error.
 *
 * @locality rdkafka main thread
 * @locks none
 */
@brief Parses and handles an InitProducerId reply.
@returns 0 on success, else an error.

@locality rdkafka main thread
@locks none
@brief Parses and handles an InitProducerId reply . @returns 0 on success else an error . @locality rdkafka main thread @locks none
void
rd_kafka_handle_InitProducerId (rd_kafka_t *rk,
                                rd_kafka_broker_t *rkb,
                                rd_kafka_resp_err_t err,
                                rd_kafka_buf_t *rkbuf,
                                rd_kafka_buf_t *request,
                                void *opaque) {
        const int log_decode_errors = LOG_ERR;
        int16_t error_code;
        rd_kafka_pid_t pid;
        if (err)
                goto err;
        rd_kafka_buf_read_throttle_time(rkbuf);
        rd_kafka_buf_read_i16(rkbuf, &error_code);
        if ((err = error_code))
                goto err;
        rd_kafka_buf_read_i64(rkbuf, &pid.id);
        rd_kafka_buf_read_i16(rkbuf, &pid.epoch);
        rd_kafka_idemp_pid_update(rkb, pid);
        return;
 err_parse:
        err = rkbuf->rkbuf_err;
 err:
        rd_kafka_idemp_request_pid_failed(rkb, err);
}
void rd_kafka_handle_InitProducerId ( rd_kafka_t * rk , rd_kafka_broker_t * rkb , rd_kafka_resp_err_t err , rd_kafka_buf_t * rkbuf , rd_kafka_buf_t * request , void * opaque ) { const int log_decode_errors = LOG_ERR ; int16_t error_code ; rd_kafka_pid_t pid ; if ( err ) goto err ; rd_kafka_buf_read_throttle_time ( rkbuf ) ; rd_kafka_buf_read_i16 ( rkbuf , & error_code ) ; if ( ( err = error_code ) ) goto err ; rd_kafka_buf_read_i64 ( rkbuf , & pid . id ) ; rd_kafka_buf_read_i16 ( rkbuf , & pid . epoch ) ; rd_kafka_idemp_pid_update ( rkb , pid ) ; return ; err_parse : err = rkbuf -> rkbuf_err ; err : rd_kafka_idemp_request_pid_failed ( rkb , err ) ; }
@brief Parses and handles an InitProducerId reply.
@brief Parses and handles an InitProducerId reply .
/* Retries are performed by idempotence state handler */