plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_cgrp.c
static void rd_kafka_cgrp_offsets_fetch_response (
	rd_kafka_t *rk,
	rd_kafka_broker_t *rkb,
	rd_kafka_resp_err_t err,
	rd_kafka_buf_t *reply,
	rd_kafka_buf_t *request,
	void *opaque) {
	rd_kafka_topic_partition_list_t *offsets = opaque;
	rd_kafka_cgrp_t *rkcg;

	if (err == RD_KAFKA_RESP_ERR__DESTROY) {
                /* Termination, quick cleanup. */
		rd_kafka_topic_partition_list_destroy(offsets);
                return;
        }

        rkcg = rd_kafka_cgrp_get(rk);

        if (rd_kafka_buf_version_outdated(request, rkcg->rkcg_version)) {
                rd_kafka_topic_partition_list_destroy(offsets);
                return;
        }

	rd_kafka_topic_partition_list_log(rk, "OFFSETFETCH",
                                          RD_KAFKA_DBG_TOPIC|RD_KAFKA_DBG_CGRP,
                                          offsets);
	/* If all partitions already had usable offsets then there
	 * was no request sent and thus no reply, the offsets list is
	 * good to go. */
	if (reply) {
		err = rd_kafka_handle_OffsetFetch(rk, rkb, err,
						  reply, request, offsets,
						  1/* Update toppars */);
                if (err == RD_KAFKA_RESP_ERR__IN_PROGRESS)
                        return; /* retrying */
        }
	if (err) {
		rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "OFFSET",
			     "Offset fetch error: %s",
			     rd_kafka_err2str(err));

		if (err != RD_KAFKA_RESP_ERR__WAIT_COORD)
                        rd_kafka_consumer_err(rkcg->rkcg_q,
                                              rd_kafka_broker_id(rkb),
                                              err, 0, NULL, NULL,
                                              RD_KAFKA_OFFSET_INVALID,
                                              "Failed to fetch offsets: %s",
                                              rd_kafka_err2str(err));
	} else {
		if (RD_KAFKA_CGRP_CAN_FETCH_START(rkcg))
			rd_kafka_cgrp_partitions_fetch_start(
				rkcg, offsets, 1 /* usable offsets */);
		else
			rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "OFFSET",
				     "Group \"%.*s\": "
				     "ignoring Offset fetch response for "
				     "%d partition(s): in state %s",
				     RD_KAFKAP_STR_PR(rkcg->rkcg_group_id),
				     offsets ? offsets->cnt : -1,
				     rd_kafka_cgrp_join_state_names[
					     rkcg->rkcg_join_state]);
	}

	rd_kafka_topic_partition_list_destroy(offsets);
}
/**
 * Reply for OffsetFetch from call below.
 */
Reply for OffsetFetch from call below.
Reply for OffsetFetch from call below .
static void rd_kafka_cgrp_offsets_fetch_response (
	rd_kafka_t *rk,
	rd_kafka_broker_t *rkb,
	rd_kafka_resp_err_t err,
	rd_kafka_buf_t *reply,
	rd_kafka_buf_t *request,
	void *opaque) {
	rd_kafka_topic_partition_list_t *offsets = opaque;
	rd_kafka_cgrp_t *rkcg;
	if (err == RD_KAFKA_RESP_ERR__DESTROY) {
		rd_kafka_topic_partition_list_destroy(offsets);
                return;
        }
        rkcg = rd_kafka_cgrp_get(rk);
        if (rd_kafka_buf_version_outdated(request, rkcg->rkcg_version)) {
                rd_kafka_topic_partition_list_destroy(offsets);
                return;
        }
	rd_kafka_topic_partition_list_log(rk, "OFFSETFETCH",
                                          RD_KAFKA_DBG_TOPIC|RD_KAFKA_DBG_CGRP,
                                          offsets);
	if (reply) {
		err = rd_kafka_handle_OffsetFetch(rk, rkb, err,
						  reply, request, offsets,
						  1);
                if (err == RD_KAFKA_RESP_ERR__IN_PROGRESS)
                        return; 
        }
	if (err) {
		rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "OFFSET",
			     "Offset fetch error: %s",
			     rd_kafka_err2str(err));
		if (err != RD_KAFKA_RESP_ERR__WAIT_COORD)
                        rd_kafka_consumer_err(rkcg->rkcg_q,
                                              rd_kafka_broker_id(rkb),
                                              err, 0, NULL, NULL,
                                              RD_KAFKA_OFFSET_INVALID,
                                              "Failed to fetch offsets: %s",
                                              rd_kafka_err2str(err));
	} else {
		if (RD_KAFKA_CGRP_CAN_FETCH_START(rkcg))
			rd_kafka_cgrp_partitions_fetch_start(
				rkcg, offsets, 1 );
		else
			rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "OFFSET",
				     "Group \"%.*s\": "
				     "ignoring Offset fetch response for "
				     "%d partition(s): in state %s",
				     RD_KAFKAP_STR_PR(rkcg->rkcg_group_id),
				     offsets ? offsets->cnt : -1,
				     rd_kafka_cgrp_join_state_names[
					     rkcg->rkcg_join_state]);
	}
	rd_kafka_topic_partition_list_destroy(offsets);
}
static void rd_kafka_cgrp_offsets_fetch_response ( rd_kafka_t * rk , rd_kafka_broker_t * rkb , rd_kafka_resp_err_t err , rd_kafka_buf_t * reply , rd_kafka_buf_t * request , void * opaque ) { rd_kafka_topic_partition_list_t * offsets = opaque ; rd_kafka_cgrp_t * rkcg ; if ( err == RD_KAFKA_RESP_ERR__DESTROY ) { rd_kafka_topic_partition_list_destroy ( offsets ) ; return ; } rkcg = rd_kafka_cgrp_get ( rk ) ; if ( rd_kafka_buf_version_outdated ( request , rkcg -> rkcg_version ) ) { rd_kafka_topic_partition_list_destroy ( offsets ) ; return ; } rd_kafka_topic_partition_list_log ( rk , " " , RD_KAFKA_DBG_TOPIC | RD_KAFKA_DBG_CGRP , offsets ) ; if ( reply ) { err = rd_kafka_handle_OffsetFetch ( rk , rkb , err , reply , request , offsets , 1 ) ; if ( err == RD_KAFKA_RESP_ERR__IN_PROGRESS ) return ; } if ( err ) { rd_kafka_dbg ( rkcg -> rkcg_rk , CGRP , " " , " " , rd_kafka_err2str ( err ) ) ; if ( err != RD_KAFKA_RESP_ERR__WAIT_COORD ) rd_kafka_consumer_err ( rkcg -> rkcg_q , rd_kafka_broker_id ( rkb ) , err , 0 , NULL , NULL , RD_KAFKA_OFFSET_INVALID , " " , rd_kafka_err2str ( err ) ) ; } else { if ( RD_KAFKA_CGRP_CAN_FETCH_START ( rkcg ) ) rd_kafka_cgrp_partitions_fetch_start ( rkcg , offsets , 1 ) ; else rd_kafka_dbg ( rkcg -> rkcg_rk , CGRP , " " , " \" \" " " " " " , RD_KAFKAP_STR_PR ( rkcg -> rkcg_group_id ) , offsets ? offsets -> cnt : -1 , rd_kafka_cgrp_join_state_names [ rkcg -> rkcg_join_state ] ) ; } rd_kafka_topic_partition_list_destroy ( offsets ) ; }
Reply for OffsetFetch from call below.
Reply for OffsetFetch from call below .
/* Termination, quick cleanup. */ /* If all partitions already had usable offsets then there
	 * was no request sent and thus no reply, the offsets list is
	 * good to go. */ /* Update toppars */ /* retrying */ /* usable offsets */