src/rdkafka_request.c
rd_kafka_resp_err_t
rd_kafka_handle_OffsetCommit (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;
        int32_t TopicArrayCnt;
        int16_t ErrorCode = 0, last_ErrorCode = 0;
	int errcnt = 0;
        int i;
	int actions;

        if (err)
		goto err;

        rd_kafka_buf_read_i32(rkbuf, &TopicArrayCnt);
        for (i = 0 ; i < TopicArrayCnt ; i++) {
                rd_kafkap_str_t topic;
                char *topic_str;
                int32_t PartArrayCnt;
                int j;

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

                RD_KAFKAP_STR_DUPA(&topic_str, &topic);

                for (j = 0 ; j < PartArrayCnt ; j++) {
                        int32_t partition;
                        rd_kafka_topic_partition_t *rktpar;

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

                        rktpar = rd_kafka_topic_partition_list_find(
                                offsets, topic_str, partition);

                        if (!rktpar) {
                                /* Received offset for topic/partition we didn't
                                 * ask for, this shouldn't really happen. */
                                continue;
                        }

                        rktpar->err = ErrorCode;
			if (ErrorCode) {
				last_ErrorCode = ErrorCode;
				errcnt++;
			}
                }
        }

	/* If all partitions failed use error code
	 * from last partition as the global error. */
	if (offsets && errcnt == offsets->cnt)
		err = last_ErrorCode;
	goto done;

 err_parse:
        err = rkbuf->rkbuf_err;

 err:
        actions = rd_kafka_err_action(
		rkb, err, request,

		RD_KAFKA_ERR_ACTION_PERMANENT,
		RD_KAFKA_RESP_ERR_OFFSET_METADATA_TOO_LARGE,

		RD_KAFKA_ERR_ACTION_RETRY,
		RD_KAFKA_RESP_ERR_GROUP_LOAD_IN_PROGRESS,

		RD_KAFKA_ERR_ACTION_REFRESH|RD_KAFKA_ERR_ACTION_SPECIAL,
		RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE,

		RD_KAFKA_ERR_ACTION_REFRESH|RD_KAFKA_ERR_ACTION_SPECIAL,
		RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP,

		RD_KAFKA_ERR_ACTION_REFRESH|RD_KAFKA_ERR_ACTION_RETRY,
		RD_KAFKA_RESP_ERR_ILLEGAL_GENERATION,

		RD_KAFKA_ERR_ACTION_REFRESH|RD_KAFKA_ERR_ACTION_RETRY,
		RD_KAFKA_RESP_ERR_UNKNOWN_MEMBER_ID,

		RD_KAFKA_ERR_ACTION_RETRY,
		RD_KAFKA_RESP_ERR_REBALANCE_IN_PROGRESS,

		RD_KAFKA_ERR_ACTION_PERMANENT,
		RD_KAFKA_RESP_ERR_INVALID_COMMIT_OFFSET_SIZE,

		RD_KAFKA_ERR_ACTION_PERMANENT,
		RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED,

		RD_KAFKA_ERR_ACTION_PERMANENT,
		RD_KAFKA_RESP_ERR_GROUP_AUTHORIZATION_FAILED,

		RD_KAFKA_ERR_ACTION_END);

	if (actions & RD_KAFKA_ERR_ACTION_REFRESH && rk->rk_cgrp) {
		/* Mark coordinator dead or re-query for coordinator.
		 * ..dead() will trigger a re-query. */
		if (actions & RD_KAFKA_ERR_ACTION_SPECIAL)
			rd_kafka_cgrp_coord_dead(rk->rk_cgrp, err,
						 "OffsetCommitRequest failed");
		else
			rd_kafka_cgrp_coord_query(rk->rk_cgrp,
						  "OffsetCommitRequest failed");
	}
	if (actions & RD_KAFKA_ERR_ACTION_RETRY) {
		if (rd_kafka_buf_retry(rkb, request))
			return RD_KAFKA_RESP_ERR__IN_PROGRESS;
		/* FALLTHRU */
	}

 done:
	return err;
}
/**
 * @remark \p offsets may be NULL if \p err is set
 */
@remark \p offsets may be NULL if \p err is set
@remark \ p offsets may be NULL if \ p err is set
rd_kafka_resp_err_t
rd_kafka_handle_OffsetCommit (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;
        int32_t TopicArrayCnt;
        int16_t ErrorCode = 0, last_ErrorCode = 0;
	int errcnt = 0;
        int i;
	int actions;
        if (err)
		goto err;
        rd_kafka_buf_read_i32(rkbuf, &TopicArrayCnt);
        for (i = 0 ; i < TopicArrayCnt ; i++) {
                rd_kafkap_str_t topic;
                char *topic_str;
                int32_t PartArrayCnt;
                int j;
                rd_kafka_buf_read_str(rkbuf, &topic);
                rd_kafka_buf_read_i32(rkbuf, &PartArrayCnt);
                RD_KAFKAP_STR_DUPA(&topic_str, &topic);
                for (j = 0 ; j < PartArrayCnt ; j++) {
                        int32_t partition;
                        rd_kafka_topic_partition_t *rktpar;
                        rd_kafka_buf_read_i32(rkbuf, &partition);
                        rd_kafka_buf_read_i16(rkbuf, &ErrorCode);
                        rktpar = rd_kafka_topic_partition_list_find(
                                offsets, topic_str, partition);
                        if (!rktpar) {
                                continue;
                        }
                        rktpar->err = ErrorCode;
			if (ErrorCode) {
				last_ErrorCode = ErrorCode;
				errcnt++;
			}
                }
        }
	if (offsets && errcnt == offsets->cnt)
		err = last_ErrorCode;
	goto done;
 err_parse:
        err = rkbuf->rkbuf_err;
 err:
        actions = rd_kafka_err_action(
		rkb, err, request,
		RD_KAFKA_ERR_ACTION_PERMANENT,
		RD_KAFKA_RESP_ERR_OFFSET_METADATA_TOO_LARGE,
		RD_KAFKA_ERR_ACTION_RETRY,
		RD_KAFKA_RESP_ERR_GROUP_LOAD_IN_PROGRESS,
		RD_KAFKA_ERR_ACTION_REFRESH|RD_KAFKA_ERR_ACTION_SPECIAL,
		RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE,
		RD_KAFKA_ERR_ACTION_REFRESH|RD_KAFKA_ERR_ACTION_SPECIAL,
		RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP,
		RD_KAFKA_ERR_ACTION_REFRESH|RD_KAFKA_ERR_ACTION_RETRY,
		RD_KAFKA_RESP_ERR_ILLEGAL_GENERATION,
		RD_KAFKA_ERR_ACTION_REFRESH|RD_KAFKA_ERR_ACTION_RETRY,
		RD_KAFKA_RESP_ERR_UNKNOWN_MEMBER_ID,
		RD_KAFKA_ERR_ACTION_RETRY,
		RD_KAFKA_RESP_ERR_REBALANCE_IN_PROGRESS,
		RD_KAFKA_ERR_ACTION_PERMANENT,
		RD_KAFKA_RESP_ERR_INVALID_COMMIT_OFFSET_SIZE,
		RD_KAFKA_ERR_ACTION_PERMANENT,
		RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED,
		RD_KAFKA_ERR_ACTION_PERMANENT,
		RD_KAFKA_RESP_ERR_GROUP_AUTHORIZATION_FAILED,
		RD_KAFKA_ERR_ACTION_END);
	if (actions & RD_KAFKA_ERR_ACTION_REFRESH && rk->rk_cgrp) {
		if (actions & RD_KAFKA_ERR_ACTION_SPECIAL)
			rd_kafka_cgrp_coord_dead(rk->rk_cgrp, err,
						 "OffsetCommitRequest failed");
		else
			rd_kafka_cgrp_coord_query(rk->rk_cgrp,
						  "OffsetCommitRequest failed");
	}
	if (actions & RD_KAFKA_ERR_ACTION_RETRY) {
		if (rd_kafka_buf_retry(rkb, request))
			return RD_KAFKA_RESP_ERR__IN_PROGRESS;
	}
 done:
	return err;
}
rd_kafka_resp_err_t rd_kafka_handle_OffsetCommit ( 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 ; int32_t TopicArrayCnt ; int16_t ErrorCode = 0 , last_ErrorCode = 0 ; int errcnt = 0 ; int i ; int actions ; if ( err ) goto err ; rd_kafka_buf_read_i32 ( rkbuf , & TopicArrayCnt ) ; for ( i = 0 ; i < TopicArrayCnt ; i ++ ) { rd_kafkap_str_t topic ; char * topic_str ; int32_t PartArrayCnt ; int j ; rd_kafka_buf_read_str ( rkbuf , & topic ) ; rd_kafka_buf_read_i32 ( rkbuf , & PartArrayCnt ) ; RD_KAFKAP_STR_DUPA ( & topic_str , & topic ) ; for ( j = 0 ; j < PartArrayCnt ; j ++ ) { int32_t partition ; rd_kafka_topic_partition_t * rktpar ; rd_kafka_buf_read_i32 ( rkbuf , & partition ) ; rd_kafka_buf_read_i16 ( rkbuf , & ErrorCode ) ; rktpar = rd_kafka_topic_partition_list_find ( offsets , topic_str , partition ) ; if ( ! rktpar ) { continue ; } rktpar -> err = ErrorCode ; if ( ErrorCode ) { last_ErrorCode = ErrorCode ; errcnt ++ ; } } } if ( offsets && errcnt == offsets -> cnt ) err = last_ErrorCode ; goto done ; err_parse : err = rkbuf -> rkbuf_err ; err : actions = rd_kafka_err_action ( rkb , err , request , RD_KAFKA_ERR_ACTION_PERMANENT , RD_KAFKA_RESP_ERR_OFFSET_METADATA_TOO_LARGE , RD_KAFKA_ERR_ACTION_RETRY , RD_KAFKA_RESP_ERR_GROUP_LOAD_IN_PROGRESS , RD_KAFKA_ERR_ACTION_REFRESH | RD_KAFKA_ERR_ACTION_SPECIAL , RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE , RD_KAFKA_ERR_ACTION_REFRESH | RD_KAFKA_ERR_ACTION_SPECIAL , RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP , RD_KAFKA_ERR_ACTION_REFRESH | RD_KAFKA_ERR_ACTION_RETRY , RD_KAFKA_RESP_ERR_ILLEGAL_GENERATION , RD_KAFKA_ERR_ACTION_REFRESH | RD_KAFKA_ERR_ACTION_RETRY , RD_KAFKA_RESP_ERR_UNKNOWN_MEMBER_ID , RD_KAFKA_ERR_ACTION_RETRY , RD_KAFKA_RESP_ERR_REBALANCE_IN_PROGRESS , RD_KAFKA_ERR_ACTION_PERMANENT , RD_KAFKA_RESP_ERR_INVALID_COMMIT_OFFSET_SIZE , RD_KAFKA_ERR_ACTION_PERMANENT , RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED , RD_KAFKA_ERR_ACTION_PERMANENT , RD_KAFKA_RESP_ERR_GROUP_AUTHORIZATION_FAILED , RD_KAFKA_ERR_ACTION_END ) ; if ( actions & RD_KAFKA_ERR_ACTION_REFRESH && rk -> rk_cgrp ) { if ( actions & RD_KAFKA_ERR_ACTION_SPECIAL ) rd_kafka_cgrp_coord_dead ( rk -> rk_cgrp , err , " " ) ; else rd_kafka_cgrp_coord_query ( rk -> rk_cgrp , " " ) ; } if ( actions & RD_KAFKA_ERR_ACTION_RETRY ) { if ( rd_kafka_buf_retry ( rkb , request ) ) return RD_KAFKA_RESP_ERR__IN_PROGRESS ; } done : return err ; }
@remark \p offsets may be NULL if \p err is set
@remark \ p offsets may be NULL if \ p err is set
/* Received offset for topic/partition we didn't
                                 * ask for, this shouldn't really happen. */ /* If all partitions failed use error code
	 * from last partition as the global error. */ /* Mark coordinator dead or re-query for coordinator.
		 * ..dead() will trigger a re-query. */ /* FALLTHRU */