plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_cgrp.c
static void rd_kafka_cgrp_op_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,
						  void *opaque) {
	rd_kafka_cgrp_t *rkcg = rk->rk_cgrp;
        rd_kafka_op_t *rko_orig = opaque;
	rd_kafka_topic_partition_list_t *offsets =
		rko_orig->rko_u.offset_commit.partitions; /* maybe NULL */
        int errcnt;
        int offset_commit_cb_served = 0;

	RD_KAFKA_OP_TYPE_ASSERT(rko_orig, RD_KAFKA_OP_OFFSET_COMMIT);

        if (rd_kafka_buf_version_outdated(request, rkcg->rkcg_version))
                err = RD_KAFKA_RESP_ERR__DESTROY;

	err = rd_kafka_handle_OffsetCommit(rk, rkb, err, rkbuf,
					   request, offsets);

        if (rkb)
                rd_rkb_dbg(rkb, CGRP, "COMMIT",
                           "OffsetCommit for %d partition(s): %s: returned: %s",
                           offsets ? offsets->cnt : -1,
                           rko_orig->rko_u.offset_commit.reason,
                           rd_kafka_err2str(err));
        else
                rd_kafka_dbg(rk, CGRP, "COMMIT",
                             "OffsetCommit for %d partition(s): %s: returned: %s",
                             offsets ? offsets->cnt : -1,
                             rko_orig->rko_u.offset_commit.reason,
                             rd_kafka_err2str(err));

        if (err == RD_KAFKA_RESP_ERR__IN_PROGRESS)
                return; /* Retrying */
        else if (err == RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP ||
                 err == RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE ||
                 err == RD_KAFKA_RESP_ERR__TRANSPORT) {
                /* The coordinator is not available, defer the offset commit
                 * to when the coordinator is back up again. */

                /* future-proofing, see timeout_scan(). */
                rd_kafka_assert(NULL, err != RD_KAFKA_RESP_ERR__WAIT_COORD);

                if (rd_kafka_cgrp_defer_offset_commit(rkcg, rko_orig,
                                                      rd_kafka_err2str(err)))
                        return;

                /* FALLTHRU and error out */
        }

	rd_kafka_assert(NULL, rkcg->rkcg_wait_commit_cnt > 0);
	rkcg->rkcg_wait_commit_cnt--;

        if (err == RD_KAFKA_RESP_ERR_NO_ERROR) {
                if (rkcg->rkcg_wait_commit_cnt == 0 &&
                    rkcg->rkcg_assignment &&
                    RD_KAFKA_CGRP_CAN_FETCH_START(rkcg))
                        rd_kafka_cgrp_partitions_fetch_start(rkcg,
                                                             rkcg->rkcg_assignment, 0);
	}

	if (err == RD_KAFKA_RESP_ERR__DESTROY ||
            (err == RD_KAFKA_RESP_ERR__NO_OFFSET &&
             rko_orig->rko_u.offset_commit.silent_empty)) {
		rd_kafka_op_destroy(rko_orig);
                rd_kafka_cgrp_check_unassign_done(
                        rkcg,
                        err == RD_KAFKA_RESP_ERR__DESTROY ?
                        "OffsetCommit done (__DESTROY)" :
                        "OffsetCommit done (__NO_OFFSET)");
		return;
	}

        /* Call on_commit interceptors */
        if (err != RD_KAFKA_RESP_ERR__NO_OFFSET &&
            err != RD_KAFKA_RESP_ERR__DESTROY &&
            offsets && offsets->cnt > 0)
                rd_kafka_interceptors_on_commit(rk, offsets, err);


	/* If no special callback is set but a offset_commit_cb has
	 * been set in conf then post an event for the latter. */
	if (!rko_orig->rko_u.offset_commit.cb && rk->rk_conf.offset_commit_cb) {
                rd_kafka_op_t *rko_reply = rd_kafka_op_new_reply(rko_orig, err);

                rd_kafka_op_set_prio(rko_reply, RD_KAFKA_PRIO_HIGH);

		if (offsets)
			rko_reply->rko_u.offset_commit.partitions =
				rd_kafka_topic_partition_list_copy(offsets);

		rko_reply->rko_u.offset_commit.cb =
			rk->rk_conf.offset_commit_cb;
		rko_reply->rko_u.offset_commit.opaque = rk->rk_conf.opaque;

                rd_kafka_q_enq(rk->rk_rep, rko_reply);
                offset_commit_cb_served++;
	}


	/* Enqueue reply to requester's queue, if any. */
	if (rko_orig->rko_replyq.q) {
                rd_kafka_op_t *rko_reply = rd_kafka_op_new_reply(rko_orig, err);

                rd_kafka_op_set_prio(rko_reply, RD_KAFKA_PRIO_HIGH);

		/* Copy offset & partitions & callbacks to reply op */
		rko_reply->rko_u.offset_commit = rko_orig->rko_u.offset_commit;
		if (offsets)
			rko_reply->rko_u.offset_commit.partitions =
				rd_kafka_topic_partition_list_copy(offsets);
                if (rko_reply->rko_u.offset_commit.reason)
                        rko_reply->rko_u.offset_commit.reason =
                        rd_strdup(rko_reply->rko_u.offset_commit.reason);

                rd_kafka_replyq_enq(&rko_orig->rko_replyq, rko_reply, 0);
                offset_commit_cb_served++;
        }

        errcnt = rd_kafka_cgrp_handle_OffsetCommit(rkcg, err, offsets);

        if (!offset_commit_cb_served &&
            offsets &&
            (errcnt > 0 ||
             (err != RD_KAFKA_RESP_ERR_NO_ERROR &&
              err != RD_KAFKA_RESP_ERR__NO_OFFSET))) {
                /* If there is no callback or handler for this (auto)
                 * commit then raise an error to the application (#1043) */
                char tmp[512];

                rd_kafka_topic_partition_list_str(
                        offsets, tmp, sizeof(tmp),
                        /* Print per-partition errors unless there was a
                         * request-level error. */
                        RD_KAFKA_FMT_F_OFFSET |
                        (errcnt ? RD_KAFKA_FMT_F_ONLY_ERR : 0));

                rd_kafka_log(rkcg->rkcg_rk, LOG_WARNING, "COMMITFAIL",
                             "Offset commit (%s) failed "
                             "for %d/%d partition(s): "
                             "%s%s%s",
                             rko_orig->rko_u.offset_commit.reason,
                             errcnt ? offsets->cnt : errcnt, offsets->cnt,
                             errcnt ? rd_kafka_err2str(err) : "",
                             errcnt ? ": " : "",
                             tmp);
        }

        rd_kafka_op_destroy(rko_orig);
}
/**
 * Handle OffsetCommitResponse
 * Takes the original 'rko' as opaque argument.
 * @remark \p rkb, rkbuf, and request may be NULL in a number of
 *         error cases (e.g., _NO_OFFSET, _WAIT_COORD)
 */
Handle OffsetCommitResponse
Takes the original 'rko' as opaque argument.
@remark \p rkb, rkbuf, and request may be NULL in a number of
error cases
Handle OffsetCommitResponse Takes the original ' rko ' as opaque argument . @remark \ p rkb rkbuf and request may be NULL in a number of error cases
static void rd_kafka_cgrp_op_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,
						  void *opaque) {
	rd_kafka_cgrp_t *rkcg = rk->rk_cgrp;
        rd_kafka_op_t *rko_orig = opaque;
	rd_kafka_topic_partition_list_t *offsets =
		rko_orig->rko_u.offset_commit.partitions; 
        int errcnt;
        int offset_commit_cb_served = 0;
	RD_KAFKA_OP_TYPE_ASSERT(rko_orig, RD_KAFKA_OP_OFFSET_COMMIT);
        if (rd_kafka_buf_version_outdated(request, rkcg->rkcg_version))
                err = RD_KAFKA_RESP_ERR__DESTROY;
	err = rd_kafka_handle_OffsetCommit(rk, rkb, err, rkbuf,
					   request, offsets);
        if (rkb)
                rd_rkb_dbg(rkb, CGRP, "COMMIT",
                           "OffsetCommit for %d partition(s): %s: returned: %s",
                           offsets ? offsets->cnt : -1,
                           rko_orig->rko_u.offset_commit.reason,
                           rd_kafka_err2str(err));
        else
                rd_kafka_dbg(rk, CGRP, "COMMIT",
                             "OffsetCommit for %d partition(s): %s: returned: %s",
                             offsets ? offsets->cnt : -1,
                             rko_orig->rko_u.offset_commit.reason,
                             rd_kafka_err2str(err));
        if (err == RD_KAFKA_RESP_ERR__IN_PROGRESS)
                return; 
        else if (err == RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP ||
                 err == RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE ||
                 err == RD_KAFKA_RESP_ERR__TRANSPORT) {
                rd_kafka_assert(NULL, err != RD_KAFKA_RESP_ERR__WAIT_COORD);
                if (rd_kafka_cgrp_defer_offset_commit(rkcg, rko_orig,
                                                      rd_kafka_err2str(err)))
                        return;
        }
	rd_kafka_assert(NULL, rkcg->rkcg_wait_commit_cnt > 0);
	rkcg->rkcg_wait_commit_cnt--;
        if (err == RD_KAFKA_RESP_ERR_NO_ERROR) {
                if (rkcg->rkcg_wait_commit_cnt == 0 &&
                    rkcg->rkcg_assignment &&
                    RD_KAFKA_CGRP_CAN_FETCH_START(rkcg))
                        rd_kafka_cgrp_partitions_fetch_start(rkcg,
                                                             rkcg->rkcg_assignment, 0);
	}
	if (err == RD_KAFKA_RESP_ERR__DESTROY ||
            (err == RD_KAFKA_RESP_ERR__NO_OFFSET &&
             rko_orig->rko_u.offset_commit.silent_empty)) {
		rd_kafka_op_destroy(rko_orig);
                rd_kafka_cgrp_check_unassign_done(
                        rkcg,
                        err == RD_KAFKA_RESP_ERR__DESTROY ?
                        "OffsetCommit done (__DESTROY)" :
                        "OffsetCommit done (__NO_OFFSET)");
		return;
	}
        if (err != RD_KAFKA_RESP_ERR__NO_OFFSET &&
            err != RD_KAFKA_RESP_ERR__DESTROY &&
            offsets && offsets->cnt > 0)
                rd_kafka_interceptors_on_commit(rk, offsets, err);
	if (!rko_orig->rko_u.offset_commit.cb && rk->rk_conf.offset_commit_cb) {
                rd_kafka_op_t *rko_reply = rd_kafka_op_new_reply(rko_orig, err);
                rd_kafka_op_set_prio(rko_reply, RD_KAFKA_PRIO_HIGH);
		if (offsets)
			rko_reply->rko_u.offset_commit.partitions =
				rd_kafka_topic_partition_list_copy(offsets);
		rko_reply->rko_u.offset_commit.cb =
			rk->rk_conf.offset_commit_cb;
		rko_reply->rko_u.offset_commit.opaque = rk->rk_conf.opaque;
                rd_kafka_q_enq(rk->rk_rep, rko_reply);
                offset_commit_cb_served++;
	}
	if (rko_orig->rko_replyq.q) {
                rd_kafka_op_t *rko_reply = rd_kafka_op_new_reply(rko_orig, err);
                rd_kafka_op_set_prio(rko_reply, RD_KAFKA_PRIO_HIGH);
		rko_reply->rko_u.offset_commit = rko_orig->rko_u.offset_commit;
		if (offsets)
			rko_reply->rko_u.offset_commit.partitions =
				rd_kafka_topic_partition_list_copy(offsets);
                if (rko_reply->rko_u.offset_commit.reason)
                        rko_reply->rko_u.offset_commit.reason =
                        rd_strdup(rko_reply->rko_u.offset_commit.reason);
                rd_kafka_replyq_enq(&rko_orig->rko_replyq, rko_reply, 0);
                offset_commit_cb_served++;
        }
        errcnt = rd_kafka_cgrp_handle_OffsetCommit(rkcg, err, offsets);
        if (!offset_commit_cb_served &&
            offsets &&
            (errcnt > 0 ||
             (err != RD_KAFKA_RESP_ERR_NO_ERROR &&
              err != RD_KAFKA_RESP_ERR__NO_OFFSET))) {
                char tmp[512];
                rd_kafka_topic_partition_list_str(
                        offsets, tmp, sizeof(tmp),
                        RD_KAFKA_FMT_F_OFFSET |
                        (errcnt ? RD_KAFKA_FMT_F_ONLY_ERR : 0));
                rd_kafka_log(rkcg->rkcg_rk, LOG_WARNING, "COMMITFAIL",
                             "Offset commit (%s) failed "
                             "for %d/%d partition(s): "
                             "%s%s%s",
                             rko_orig->rko_u.offset_commit.reason,
                             errcnt ? offsets->cnt : errcnt, offsets->cnt,
                             errcnt ? rd_kafka_err2str(err) : "",
                             errcnt ? ": " : "",
                             tmp);
        }
        rd_kafka_op_destroy(rko_orig);
}
static void rd_kafka_cgrp_op_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 , void * opaque ) { rd_kafka_cgrp_t * rkcg = rk -> rk_cgrp ; rd_kafka_op_t * rko_orig = opaque ; rd_kafka_topic_partition_list_t * offsets = rko_orig -> rko_u . offset_commit . partitions ; int errcnt ; int offset_commit_cb_served = 0 ; RD_KAFKA_OP_TYPE_ASSERT ( rko_orig , RD_KAFKA_OP_OFFSET_COMMIT ) ; if ( rd_kafka_buf_version_outdated ( request , rkcg -> rkcg_version ) ) err = RD_KAFKA_RESP_ERR__DESTROY ; err = rd_kafka_handle_OffsetCommit ( rk , rkb , err , rkbuf , request , offsets ) ; if ( rkb ) rd_rkb_dbg ( rkb , CGRP , " " , " " , offsets ? offsets -> cnt : -1 , rko_orig -> rko_u . offset_commit . reason , rd_kafka_err2str ( err ) ) ; else rd_kafka_dbg ( rk , CGRP , " " , " " , offsets ? offsets -> cnt : -1 , rko_orig -> rko_u . offset_commit . reason , rd_kafka_err2str ( err ) ) ; if ( err == RD_KAFKA_RESP_ERR__IN_PROGRESS ) return ; else if ( err == RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP || err == RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE || err == RD_KAFKA_RESP_ERR__TRANSPORT ) { rd_kafka_assert ( NULL , err != RD_KAFKA_RESP_ERR__WAIT_COORD ) ; if ( rd_kafka_cgrp_defer_offset_commit ( rkcg , rko_orig , rd_kafka_err2str ( err ) ) ) return ; } rd_kafka_assert ( NULL , rkcg -> rkcg_wait_commit_cnt > 0 ) ; rkcg -> rkcg_wait_commit_cnt -- ; if ( err == RD_KAFKA_RESP_ERR_NO_ERROR ) { if ( rkcg -> rkcg_wait_commit_cnt == 0 && rkcg -> rkcg_assignment && RD_KAFKA_CGRP_CAN_FETCH_START ( rkcg ) ) rd_kafka_cgrp_partitions_fetch_start ( rkcg , rkcg -> rkcg_assignment , 0 ) ; } if ( err == RD_KAFKA_RESP_ERR__DESTROY || ( err == RD_KAFKA_RESP_ERR__NO_OFFSET && rko_orig -> rko_u . offset_commit . silent_empty ) ) { rd_kafka_op_destroy ( rko_orig ) ; rd_kafka_cgrp_check_unassign_done ( rkcg , err == RD_KAFKA_RESP_ERR__DESTROY ? " " : " " ) ; return ; } if ( err != RD_KAFKA_RESP_ERR__NO_OFFSET && err != RD_KAFKA_RESP_ERR__DESTROY && offsets && offsets -> cnt > 0 ) rd_kafka_interceptors_on_commit ( rk , offsets , err ) ; if ( ! rko_orig -> rko_u . offset_commit . cb && rk -> rk_conf . offset_commit_cb ) { rd_kafka_op_t * rko_reply = rd_kafka_op_new_reply ( rko_orig , err ) ; rd_kafka_op_set_prio ( rko_reply , RD_KAFKA_PRIO_HIGH ) ; if ( offsets ) rko_reply -> rko_u . offset_commit . partitions = rd_kafka_topic_partition_list_copy ( offsets ) ; rko_reply -> rko_u . offset_commit . cb = rk -> rk_conf . offset_commit_cb ; rko_reply -> rko_u . offset_commit . opaque = rk -> rk_conf . opaque ; rd_kafka_q_enq ( rk -> rk_rep , rko_reply ) ; offset_commit_cb_served ++ ; } if ( rko_orig -> rko_replyq . q ) { rd_kafka_op_t * rko_reply = rd_kafka_op_new_reply ( rko_orig , err ) ; rd_kafka_op_set_prio ( rko_reply , RD_KAFKA_PRIO_HIGH ) ; rko_reply -> rko_u . offset_commit = rko_orig -> rko_u . offset_commit ; if ( offsets ) rko_reply -> rko_u . offset_commit . partitions = rd_kafka_topic_partition_list_copy ( offsets ) ; if ( rko_reply -> rko_u . offset_commit . reason ) rko_reply -> rko_u . offset_commit . reason = rd_strdup ( rko_reply -> rko_u . offset_commit . reason ) ; rd_kafka_replyq_enq ( & rko_orig -> rko_replyq , rko_reply , 0 ) ; offset_commit_cb_served ++ ; } errcnt = rd_kafka_cgrp_handle_OffsetCommit ( rkcg , err , offsets ) ; if ( ! offset_commit_cb_served && offsets && ( errcnt > 0 || ( err != RD_KAFKA_RESP_ERR_NO_ERROR && err != RD_KAFKA_RESP_ERR__NO_OFFSET ) ) ) { char tmp [ 512 ] ; rd_kafka_topic_partition_list_str ( offsets , tmp , sizeof ( tmp ) , RD_KAFKA_FMT_F_OFFSET | ( errcnt ? RD_KAFKA_FMT_F_ONLY_ERR : 0 ) ) ; rd_kafka_log ( rkcg -> rkcg_rk , LOG_WARNING , " " , " " " " " " , rko_orig -> rko_u . offset_commit . reason , errcnt ? offsets -> cnt : errcnt , offsets -> cnt , errcnt ? rd_kafka_err2str ( err ) : " " , errcnt ? " " : " " , tmp ) ; } rd_kafka_op_destroy ( rko_orig ) ; }
Handle OffsetCommitResponse
 Takes the original 'rko' as opaque argument.
Handle OffsetCommitResponse Takes the original ' rko ' as opaque argument .
/* maybe NULL */ /* Retrying */ /* The coordinator is not available, defer the offset commit
                 * to when the coordinator is back up again. */ /* future-proofing, see timeout_scan(). */ /* FALLTHRU and error out */ /* Call on_commit interceptors */ /* If no special callback is set but a offset_commit_cb has
	 * been set in conf then post an event for the latter. */ /* Enqueue reply to requester's queue, if any. */ /* Copy offset & partitions & callbacks to reply op */ /* If there is no callback or handler for this (auto)
                 * commit then raise an error to the application (#1043) */ /* Print per-partition errors unless there was a
                         * request-level error. */