plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_cgrp.c
static void rd_kafka_cgrp_unassign_done (rd_kafka_cgrp_t *rkcg,
                                         const char *reason) {
	rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "UNASSIGN",
		     "Group \"%s\": unassign done in state %s (join state %s): "
		     "%s: %s",
		     rkcg->rkcg_group_id->str,
		     rd_kafka_cgrp_state_names[rkcg->rkcg_state],
		     rd_kafka_cgrp_join_state_names[rkcg->rkcg_join_state],
		     rkcg->rkcg_assignment ?
		     "with new assignment" : "without new assignment",
                     reason);

        /* Don't send Leave when termating with NO_CONSUMER_CLOSE flag */
        if (rd_kafka_destroy_flags_no_consumer_close(rkcg->rkcg_rk))
                rkcg->rkcg_flags &= ~RD_KAFKA_CGRP_F_LEAVE_ON_UNASSIGN;

        /*
         * KIP-345: Static group members must not send a LeaveGroupRequest
         * on termination.
         */
        if (RD_KAFKA_CGRP_IS_STATIC_MEMBER(rkcg) &&
            rkcg->rkcg_flags & RD_KAFKA_CGRP_F_TERMINATE)
                rkcg->rkcg_flags &= ~RD_KAFKA_CGRP_F_LEAVE_ON_UNASSIGN;

	if (rkcg->rkcg_flags & RD_KAFKA_CGRP_F_LEAVE_ON_UNASSIGN) {
		rd_kafka_cgrp_leave(rkcg);
		rkcg->rkcg_flags &= ~RD_KAFKA_CGRP_F_LEAVE_ON_UNASSIGN;
	}

        if (rkcg->rkcg_join_state != RD_KAFKA_CGRP_JOIN_STATE_WAIT_UNASSIGN) {
                rd_kafka_cgrp_try_terminate(rkcg);
                return;
        }

        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 {
                /* Skip the join backoff */
                rd_interval_reset(&rkcg->rkcg_join_intvl);

		rd_kafka_cgrp_set_join_state(rkcg,
					     RD_KAFKA_CGRP_JOIN_STATE_INIT);
	}

	rd_kafka_cgrp_try_terminate(rkcg);
}
/**
 * Call when all unassign operations are done to transition to the next state
 */
Call when all unassign operations are done to transition to the next state
Call when all unassign operations are done to transition to the next state
static void rd_kafka_cgrp_unassign_done (rd_kafka_cgrp_t *rkcg,
                                         const char *reason) {
	rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "UNASSIGN",
		     "Group \"%s\": unassign done in state %s (join state %s): "
		     "%s: %s",
		     rkcg->rkcg_group_id->str,
		     rd_kafka_cgrp_state_names[rkcg->rkcg_state],
		     rd_kafka_cgrp_join_state_names[rkcg->rkcg_join_state],
		     rkcg->rkcg_assignment ?
		     "with new assignment" : "without new assignment",
                     reason);
        if (rd_kafka_destroy_flags_no_consumer_close(rkcg->rkcg_rk))
                rkcg->rkcg_flags &= ~RD_KAFKA_CGRP_F_LEAVE_ON_UNASSIGN;
        if (RD_KAFKA_CGRP_IS_STATIC_MEMBER(rkcg) &&
            rkcg->rkcg_flags & RD_KAFKA_CGRP_F_TERMINATE)
                rkcg->rkcg_flags &= ~RD_KAFKA_CGRP_F_LEAVE_ON_UNASSIGN;
	if (rkcg->rkcg_flags & RD_KAFKA_CGRP_F_LEAVE_ON_UNASSIGN) {
		rd_kafka_cgrp_leave(rkcg);
		rkcg->rkcg_flags &= ~RD_KAFKA_CGRP_F_LEAVE_ON_UNASSIGN;
	}
        if (rkcg->rkcg_join_state != RD_KAFKA_CGRP_JOIN_STATE_WAIT_UNASSIGN) {
                rd_kafka_cgrp_try_terminate(rkcg);
                return;
        }
        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);
	}
	rd_kafka_cgrp_try_terminate(rkcg);
}
static void rd_kafka_cgrp_unassign_done ( rd_kafka_cgrp_t * rkcg , const char * reason ) { rd_kafka_dbg ( rkcg -> rkcg_rk , CGRP , " " , " \" \" " " " , rkcg -> rkcg_group_id -> str , rd_kafka_cgrp_state_names [ rkcg -> rkcg_state ] , rd_kafka_cgrp_join_state_names [ rkcg -> rkcg_join_state ] , rkcg -> rkcg_assignment ? " " : " " , reason ) ; if ( rd_kafka_destroy_flags_no_consumer_close ( rkcg -> rkcg_rk ) ) rkcg -> rkcg_flags &= ~ RD_KAFKA_CGRP_F_LEAVE_ON_UNASSIGN ; if ( RD_KAFKA_CGRP_IS_STATIC_MEMBER ( rkcg ) && rkcg -> rkcg_flags & RD_KAFKA_CGRP_F_TERMINATE ) rkcg -> rkcg_flags &= ~ RD_KAFKA_CGRP_F_LEAVE_ON_UNASSIGN ; if ( rkcg -> rkcg_flags & RD_KAFKA_CGRP_F_LEAVE_ON_UNASSIGN ) { rd_kafka_cgrp_leave ( rkcg ) ; rkcg -> rkcg_flags &= ~ RD_KAFKA_CGRP_F_LEAVE_ON_UNASSIGN ; } if ( rkcg -> rkcg_join_state != RD_KAFKA_CGRP_JOIN_STATE_WAIT_UNASSIGN ) { rd_kafka_cgrp_try_terminate ( rkcg ) ; return ; } 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 ) ; } rd_kafka_cgrp_try_terminate ( rkcg ) ; }
Call when all unassign operations are done to transition to the next state
Call when all unassign operations are done to transition to the next state
/* Don't send Leave when termating with NO_CONSUMER_CLOSE flag */ /*
         * KIP-345: Static group members must not send a LeaveGroupRequest
         * on termination.
         */ /* Skip the join backoff */