src/rdkafka_request.c
rd_kafka_resp_err_t rd_kafka_handle_Offset (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,
                                            rd_kafka_topic_partition_list_t
                                            *offsets) {

        const int log_decode_errors = LOG_ERR;
        int16_t ErrorCode = 0;
        int32_t TopicArrayCnt;
        int actions;
        int16_t api_version;

        if (err) {
                ErrorCode = err;
                goto err;
        }

        api_version = request->rkbuf_reqhdr.ApiVersion;

        /* NOTE:
         * Broker may return offsets in a different constellation than
         * in the original request .*/

        rd_kafka_buf_read_i32(rkbuf, &TopicArrayCnt);
        while (TopicArrayCnt-- > 0) {
                rd_kafkap_str_t ktopic;
                int32_t PartArrayCnt;
                char *topic_name;

                rd_kafka_buf_read_str(rkbuf, &ktopic);
                rd_kafka_buf_read_i32(rkbuf, &PartArrayCnt);

                RD_KAFKAP_STR_DUPA(&topic_name, &ktopic);

                while (PartArrayCnt-- > 0) {
                        int32_t kpartition;
                        int32_t OffsetArrayCnt;
                        int64_t Offset = -1;
                        rd_kafka_topic_partition_t *rktpar;

                        rd_kafka_buf_read_i32(rkbuf, &kpartition);
                        rd_kafka_buf_read_i16(rkbuf, &ErrorCode);

                        if (api_version == 1) {
                                int64_t Timestamp;
                                rd_kafka_buf_read_i64(rkbuf, &Timestamp);
                                rd_kafka_buf_read_i64(rkbuf, &Offset);
                        } else if (api_version == 0) {
                                rd_kafka_buf_read_i32(rkbuf, &OffsetArrayCnt);
                                /* We only request one offset so just grab
                                 * the first one. */
                                while (OffsetArrayCnt-- > 0)
                                        rd_kafka_buf_read_i64(rkbuf, &Offset);
                        } else {
                                rd_kafka_assert(NULL, !*"NOTREACHED");
                        }

                        rktpar = rd_kafka_topic_partition_list_add(
                                offsets, topic_name, kpartition);
                        rktpar->err = ErrorCode;
                        rktpar->offset = Offset;
                }
        }

        goto done;

 err_parse:
        ErrorCode = rkbuf->rkbuf_err;
 err:
        actions = rd_kafka_err_action(
                rkb, ErrorCode, request,
                RD_KAFKA_ERR_ACTION_PERMANENT,
                RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART,

                RD_KAFKA_ERR_ACTION_REFRESH|RD_KAFKA_ERR_ACTION_RETRY,
                RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION,

                RD_KAFKA_ERR_ACTION_END);

        if (actions & RD_KAFKA_ERR_ACTION_REFRESH) {
                char tmp[256];
                /* Re-query for leader */
                rd_snprintf(tmp, sizeof(tmp),
                            "OffsetRequest failed: %s",
                            rd_kafka_err2str(ErrorCode));
                rd_kafka_metadata_refresh_known_topics(rk, NULL, 1/*force*/,
                                                       tmp);
        }

        if (actions & RD_KAFKA_ERR_ACTION_RETRY) {
                if (rd_kafka_buf_retry(rkb, request))
                        return RD_KAFKA_RESP_ERR__IN_PROGRESS;
                /* FALLTHRU */
        }

done:
        return ErrorCode;
}
/**
 * @brief Parses and handles Offset replies.
 *
 * Returns the parsed offsets (and errors) in \p offsets
 *
 * @returns 0 on success, else an error.
 */
@brief Parses and handles Offset replies.
Returns the parsed offsets (and errors) in \p offsets

@returns 0 on success, else an error.
@brief Parses and handles Offset replies . Returns the parsed offsets ( and errors ) in \ p offsets @returns 0 on success else an error .
rd_kafka_resp_err_t rd_kafka_handle_Offset (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,
                                            rd_kafka_topic_partition_list_t
                                            *offsets) {
        const int log_decode_errors = LOG_ERR;
        int16_t ErrorCode = 0;
        int32_t TopicArrayCnt;
        int actions;
        int16_t api_version;
        if (err) {
                ErrorCode = err;
                goto err;
        }
        api_version = request->rkbuf_reqhdr.ApiVersion;
        rd_kafka_buf_read_i32(rkbuf, &TopicArrayCnt);
        while (TopicArrayCnt-- > 0) {
                rd_kafkap_str_t ktopic;
                int32_t PartArrayCnt;
                char *topic_name;
                rd_kafka_buf_read_str(rkbuf, &ktopic);
                rd_kafka_buf_read_i32(rkbuf, &PartArrayCnt);
                RD_KAFKAP_STR_DUPA(&topic_name, &ktopic);
                while (PartArrayCnt-- > 0) {
                        int32_t kpartition;
                        int32_t OffsetArrayCnt;
                        int64_t Offset = -1;
                        rd_kafka_topic_partition_t *rktpar;
                        rd_kafka_buf_read_i32(rkbuf, &kpartition);
                        rd_kafka_buf_read_i16(rkbuf, &ErrorCode);
                        if (api_version == 1) {
                                int64_t Timestamp;
                                rd_kafka_buf_read_i64(rkbuf, &Timestamp);
                                rd_kafka_buf_read_i64(rkbuf, &Offset);
                        } else if (api_version == 0) {
                                rd_kafka_buf_read_i32(rkbuf, &OffsetArrayCnt);
                                while (OffsetArrayCnt-- > 0)
                                        rd_kafka_buf_read_i64(rkbuf, &Offset);
                        } else {
                                rd_kafka_assert(NULL, !*"NOTREACHED");
                        }
                        rktpar = rd_kafka_topic_partition_list_add(
                                offsets, topic_name, kpartition);
                        rktpar->err = ErrorCode;
                        rktpar->offset = Offset;
                }
        }
        goto done;
 err_parse:
        ErrorCode = rkbuf->rkbuf_err;
 err:
        actions = rd_kafka_err_action(
                rkb, ErrorCode, request,
                RD_KAFKA_ERR_ACTION_PERMANENT,
                RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART,
                RD_KAFKA_ERR_ACTION_REFRESH|RD_KAFKA_ERR_ACTION_RETRY,
                RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION,
                RD_KAFKA_ERR_ACTION_END);
        if (actions & RD_KAFKA_ERR_ACTION_REFRESH) {
                char tmp[256];
                rd_snprintf(tmp, sizeof(tmp),
                            "OffsetRequest failed: %s",
                            rd_kafka_err2str(ErrorCode));
                rd_kafka_metadata_refresh_known_topics(rk, NULL, 1,
                                                       tmp);
        }
        if (actions & RD_KAFKA_ERR_ACTION_RETRY) {
                if (rd_kafka_buf_retry(rkb, request))
                        return RD_KAFKA_RESP_ERR__IN_PROGRESS;
        }
done:
        return ErrorCode;
}
rd_kafka_resp_err_t rd_kafka_handle_Offset ( 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 , rd_kafka_topic_partition_list_t * offsets ) { const int log_decode_errors = LOG_ERR ; int16_t ErrorCode = 0 ; int32_t TopicArrayCnt ; int actions ; int16_t api_version ; if ( err ) { ErrorCode = err ; goto err ; } api_version = request -> rkbuf_reqhdr . ApiVersion ; rd_kafka_buf_read_i32 ( rkbuf , & TopicArrayCnt ) ; while ( TopicArrayCnt -- > 0 ) { rd_kafkap_str_t ktopic ; int32_t PartArrayCnt ; char * topic_name ; rd_kafka_buf_read_str ( rkbuf , & ktopic ) ; rd_kafka_buf_read_i32 ( rkbuf , & PartArrayCnt ) ; RD_KAFKAP_STR_DUPA ( & topic_name , & ktopic ) ; while ( PartArrayCnt -- > 0 ) { int32_t kpartition ; int32_t OffsetArrayCnt ; int64_t Offset = -1 ; rd_kafka_topic_partition_t * rktpar ; rd_kafka_buf_read_i32 ( rkbuf , & kpartition ) ; rd_kafka_buf_read_i16 ( rkbuf , & ErrorCode ) ; if ( api_version == 1 ) { int64_t Timestamp ; rd_kafka_buf_read_i64 ( rkbuf , & Timestamp ) ; rd_kafka_buf_read_i64 ( rkbuf , & Offset ) ; } else if ( api_version == 0 ) { rd_kafka_buf_read_i32 ( rkbuf , & OffsetArrayCnt ) ; while ( OffsetArrayCnt -- > 0 ) rd_kafka_buf_read_i64 ( rkbuf , & Offset ) ; } else { rd_kafka_assert ( NULL , ! * " " ) ; } rktpar = rd_kafka_topic_partition_list_add ( offsets , topic_name , kpartition ) ; rktpar -> err = ErrorCode ; rktpar -> offset = Offset ; } } goto done ; err_parse : ErrorCode = rkbuf -> rkbuf_err ; err : actions = rd_kafka_err_action ( rkb , ErrorCode , request , RD_KAFKA_ERR_ACTION_PERMANENT , RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART , RD_KAFKA_ERR_ACTION_REFRESH | RD_KAFKA_ERR_ACTION_RETRY , RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION , RD_KAFKA_ERR_ACTION_END ) ; if ( actions & RD_KAFKA_ERR_ACTION_REFRESH ) { char tmp [ 256 ] ; rd_snprintf ( tmp , sizeof ( tmp ) , " " , rd_kafka_err2str ( ErrorCode ) ) ; rd_kafka_metadata_refresh_known_topics ( rk , NULL , 1 , tmp ) ; } if ( actions & RD_KAFKA_ERR_ACTION_RETRY ) { if ( rd_kafka_buf_retry ( rkb , request ) ) return RD_KAFKA_RESP_ERR__IN_PROGRESS ; } done : return ErrorCode ; }
@brief Parses and handles Offset replies.
@brief Parses and handles Offset replies .
/* NOTE:
         * Broker may return offsets in a different constellation than
         * in the original request .*/ /* We only request one offset so just grab
                                 * the first one. */ /* Re-query for leader */ /*force*/ /* FALLTHRU */