plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_cgrp.c
static rd_kafka_resp_err_t
rd_kafka_cgrp_assign (rd_kafka_cgrp_t *rkcg,
                      rd_kafka_topic_partition_list_t *assignment) {
        int i;
        rd_kafka_resp_err_t err = RD_KAFKA_RESP_ERR_NO_ERROR;

        rd_kafka_dbg(rkcg->rkcg_rk, CGRP|RD_KAFKA_DBG_CONSUMER, "ASSIGN",
                     "Group \"%s\": new assignment of %d partition(s) "
                     "in join state %s",
                     rkcg->rkcg_group_id->str,
                     assignment ? assignment->cnt : 0,
                     rd_kafka_cgrp_join_state_names[rkcg->rkcg_join_state]);

        /* Get toppar object for each partition.
         * This is to make sure the rktp stays alive during unassign(). */
        for (i = 0 ; assignment && i < assignment->cnt ; i++) {
                rd_kafka_topic_partition_t *rktpar;
                rd_kafka_toppar_t *rktp;

                rktpar = &assignment->elems[i];

                /* Use existing toppar if set */
                if (rktpar->_private)
                        continue;

                rktp = rd_kafka_toppar_get2(rkcg->rkcg_rk,
                                            rktpar->topic,
                                            rktpar->partition,
                                            0/*no-ua*/, 1/*create-on-miss*/);
                if (rktp)
                        rktpar->_private = rktp;
        }

        rd_kafka_cgrp_version_new_barrier(rkcg);

        rd_kafka_wrlock(rkcg->rkcg_rk);
        rkcg->rkcg_c.assignment_size = assignment ? assignment->cnt : 0;
        rd_kafka_wrunlock(rkcg->rkcg_rk);


        /* Remove existing assignment (async operation) */
	if (rkcg->rkcg_assignment)
		rd_kafka_cgrp_unassign(rkcg);

        /* If the consumer has raised a fatal error we treat all
         * assigns as unassigns */
        if (rd_kafka_fatal_error_code(rkcg->rkcg_rk)) {
                err = RD_KAFKA_RESP_ERR__FATAL;
                assignment = NULL;
        }

        rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "ASSIGN",
                     "Group \"%s\": assigning %d partition(s) in join state %s",
                     rkcg->rkcg_group_id->str, assignment ? assignment->cnt : 0,
                     rd_kafka_cgrp_join_state_names[rkcg->rkcg_join_state]);


	if (assignment) {
		rkcg->rkcg_assignment =
			rd_kafka_topic_partition_list_copy(assignment);

                /* Mark partition(s) as desired */
                for (i = 0 ; i < rkcg->rkcg_assignment->cnt ; i++) {
                        rd_kafka_topic_partition_t *rktpar =
                                &rkcg->rkcg_assignment->elems[i];
                        rd_kafka_toppar_t *rktp = rktpar->_private;
                        rd_kafka_toppar_lock(rktp);
                        rd_kafka_toppar_desired_add0(rktp);
                        rd_kafka_toppar_unlock(rktp);
                }
        }

        if (rkcg->rkcg_join_state == RD_KAFKA_CGRP_JOIN_STATE_WAIT_UNASSIGN)
                return err;

        rd_dassert(rkcg->rkcg_wait_unassign_cnt == 0);

        if (rkcg->rkcg_assignment) {
                /* No existing assignment that needs to be decommissioned,
                 * start partition fetchers right away */
		rd_kafka_cgrp_set_join_state(rkcg,
					     RD_KAFKA_CGRP_JOIN_STATE_ASSIGNED);
                if (RD_KAFKA_CGRP_CAN_FETCH_START(rkcg))
                        rd_kafka_cgrp_partitions_fetch_start(
                                rkcg, rkcg->rkcg_assignment, 0);
	} else {
                /* Skip the join backoff */
                rd_interval_reset(&rkcg->rkcg_join_intvl);

		rd_kafka_cgrp_set_join_state(rkcg,
					     RD_KAFKA_CGRP_JOIN_STATE_INIT);
	}

        return err;
}
/**
 * @brief Set new atomic partition assignment
 *        May update \p assignment but will not hold on to it.
 *
 * @returns 0 on success or an error if a fatal error has been raised.
 */
@brief Set new atomic partition assignment
May update \p assignment but will not hold on to it.

@returns 0 on success or an error if a fatal error has been raised.
@brief Set new atomic partition assignment May update \ p assignment but will not hold on to it . @returns 0 on success or an error if a fatal error has been raised .
static rd_kafka_resp_err_t
rd_kafka_cgrp_assign (rd_kafka_cgrp_t *rkcg,
                      rd_kafka_topic_partition_list_t *assignment) {
        int i;
        rd_kafka_resp_err_t err = RD_KAFKA_RESP_ERR_NO_ERROR;
        rd_kafka_dbg(rkcg->rkcg_rk, CGRP|RD_KAFKA_DBG_CONSUMER, "ASSIGN",
                     "Group \"%s\": new assignment of %d partition(s) "
                     "in join state %s",
                     rkcg->rkcg_group_id->str,
                     assignment ? assignment->cnt : 0,
                     rd_kafka_cgrp_join_state_names[rkcg->rkcg_join_state]);
        for (i = 0 ; assignment && i < assignment->cnt ; i++) {
                rd_kafka_topic_partition_t *rktpar;
                rd_kafka_toppar_t *rktp;
                rktpar = &assignment->elems[i];
                if (rktpar->_private)
                        continue;
                rktp = rd_kafka_toppar_get2(rkcg->rkcg_rk,
                                            rktpar->topic,
                                            rktpar->partition,
                                            0, 1);
                if (rktp)
                        rktpar->_private = rktp;
        }
        rd_kafka_cgrp_version_new_barrier(rkcg);
        rd_kafka_wrlock(rkcg->rkcg_rk);
        rkcg->rkcg_c.assignment_size = assignment ? assignment->cnt : 0;
        rd_kafka_wrunlock(rkcg->rkcg_rk);
	if (rkcg->rkcg_assignment)
		rd_kafka_cgrp_unassign(rkcg);
        if (rd_kafka_fatal_error_code(rkcg->rkcg_rk)) {
                err = RD_KAFKA_RESP_ERR__FATAL;
                assignment = NULL;
        }
        rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "ASSIGN",
                     "Group \"%s\": assigning %d partition(s) in join state %s",
                     rkcg->rkcg_group_id->str, assignment ? assignment->cnt : 0,
                     rd_kafka_cgrp_join_state_names[rkcg->rkcg_join_state]);
	if (assignment) {
		rkcg->rkcg_assignment =
			rd_kafka_topic_partition_list_copy(assignment);
                for (i = 0 ; i < rkcg->rkcg_assignment->cnt ; i++) {
                        rd_kafka_topic_partition_t *rktpar =
                                &rkcg->rkcg_assignment->elems[i];
                        rd_kafka_toppar_t *rktp = rktpar->_private;
                        rd_kafka_toppar_lock(rktp);
                        rd_kafka_toppar_desired_add0(rktp);
                        rd_kafka_toppar_unlock(rktp);
                }
        }
        if (rkcg->rkcg_join_state == RD_KAFKA_CGRP_JOIN_STATE_WAIT_UNASSIGN)
                return err;
        rd_dassert(rkcg->rkcg_wait_unassign_cnt == 0);
        if (rkcg->rkcg_assignment) {
		rd_kafka_cgrp_set_join_state(rkcg,
					     RD_KAFKA_CGRP_JOIN_STATE_ASSIGNED);
                if (RD_KAFKA_CGRP_CAN_FETCH_START(rkcg))
                        rd_kafka_cgrp_partitions_fetch_start(
                                rkcg, rkcg->rkcg_assignment, 0);
	} else {
                rd_interval_reset(&rkcg->rkcg_join_intvl);
		rd_kafka_cgrp_set_join_state(rkcg,
					     RD_KAFKA_CGRP_JOIN_STATE_INIT);
	}
        return err;
}
static rd_kafka_resp_err_t rd_kafka_cgrp_assign ( rd_kafka_cgrp_t * rkcg , rd_kafka_topic_partition_list_t * assignment ) { int i ; rd_kafka_resp_err_t err = RD_KAFKA_RESP_ERR_NO_ERROR ; rd_kafka_dbg ( rkcg -> rkcg_rk , CGRP | RD_KAFKA_DBG_CONSUMER , " " , " \" \" " " " , rkcg -> rkcg_group_id -> str , assignment ? assignment -> cnt : 0 , rd_kafka_cgrp_join_state_names [ rkcg -> rkcg_join_state ] ) ; for ( i = 0 ; assignment && i < assignment -> cnt ; i ++ ) { rd_kafka_topic_partition_t * rktpar ; rd_kafka_toppar_t * rktp ; rktpar = & assignment -> elems [ i ] ; if ( rktpar -> _private ) continue ; rktp = rd_kafka_toppar_get2 ( rkcg -> rkcg_rk , rktpar -> topic , rktpar -> partition , 0 , 1 ) ; if ( rktp ) rktpar -> _private = rktp ; } rd_kafka_cgrp_version_new_barrier ( rkcg ) ; rd_kafka_wrlock ( rkcg -> rkcg_rk ) ; rkcg -> rkcg_c . assignment_size = assignment ? assignment -> cnt : 0 ; rd_kafka_wrunlock ( rkcg -> rkcg_rk ) ; if ( rkcg -> rkcg_assignment ) rd_kafka_cgrp_unassign ( rkcg ) ; if ( rd_kafka_fatal_error_code ( rkcg -> rkcg_rk ) ) { err = RD_KAFKA_RESP_ERR__FATAL ; assignment = NULL ; } rd_kafka_dbg ( rkcg -> rkcg_rk , CGRP , " " , " \" \" " , rkcg -> rkcg_group_id -> str , assignment ? assignment -> cnt : 0 , rd_kafka_cgrp_join_state_names [ rkcg -> rkcg_join_state ] ) ; if ( assignment ) { rkcg -> rkcg_assignment = rd_kafka_topic_partition_list_copy ( assignment ) ; for ( i = 0 ; i < rkcg -> rkcg_assignment -> cnt ; i ++ ) { rd_kafka_topic_partition_t * rktpar = & rkcg -> rkcg_assignment -> elems [ i ] ; rd_kafka_toppar_t * rktp = rktpar -> _private ; rd_kafka_toppar_lock ( rktp ) ; rd_kafka_toppar_desired_add0 ( rktp ) ; rd_kafka_toppar_unlock ( rktp ) ; } } if ( rkcg -> rkcg_join_state == RD_KAFKA_CGRP_JOIN_STATE_WAIT_UNASSIGN ) return err ; rd_dassert ( rkcg -> rkcg_wait_unassign_cnt == 0 ) ; if ( rkcg -> rkcg_assignment ) { rd_kafka_cgrp_set_join_state ( rkcg , RD_KAFKA_CGRP_JOIN_STATE_ASSIGNED ) ; if ( RD_KAFKA_CGRP_CAN_FETCH_START ( rkcg ) ) rd_kafka_cgrp_partitions_fetch_start ( rkcg , rkcg -> rkcg_assignment , 0 ) ; } else { rd_interval_reset ( & rkcg -> rkcg_join_intvl ) ; rd_kafka_cgrp_set_join_state ( rkcg , RD_KAFKA_CGRP_JOIN_STATE_INIT ) ; } return err ; }
@brief Set new atomic partition assignment
        May update \p assignment but will not hold on to it.
@brief Set new atomic partition assignment May update \ p assignment but will not hold on to it .
/* Get toppar object for each partition.
         * This is to make sure the rktp stays alive during unassign(). */ /* Use existing toppar if set */ /*no-ua*/ /*create-on-miss*/ /* Remove existing assignment (async operation) */ /* If the consumer has raised a fatal error we treat all
         * assigns as unassigns */ /* Mark partition(s) as desired */ /* No existing assignment that needs to be decommissioned,
                 * start partition fetchers right away */ /* Skip the join backoff */