plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_cgrp.c
static rd_kafka_resp_err_t
rd_kafka_cgrp_subscribe (rd_kafka_cgrp_t *rkcg,
                         rd_kafka_topic_partition_list_t *rktparlist) {

	rd_kafka_dbg(rkcg->rkcg_rk, CGRP|RD_KAFKA_DBG_CONSUMER, "SUBSCRIBE",
		     "Group \"%.*s\": subscribe to new %ssubscription "
		     "of %d topics (join state %s)",
		     RD_KAFKAP_STR_PR(rkcg->rkcg_group_id),
		     rktparlist ? "" : "unset ",
		     rktparlist ? rktparlist->cnt : 0,
		     rd_kafka_cgrp_join_state_names[rkcg->rkcg_join_state]);

        if (rkcg->rkcg_rk->rk_conf.enabled_assignor_cnt == 0)
                return RD_KAFKA_RESP_ERR__INVALID_ARG;

        /* Remove existing subscription first */
        rd_kafka_cgrp_unsubscribe(rkcg,
                                  rktparlist ?
                                  0/* dont leave group if new subscription */ :
                                  1/* leave group if no new subscription */);

        /* If the consumer has raised a fatal error we treat all
         * subscribes as unsubscribe */
        if (rd_kafka_fatal_error_code(rkcg->rkcg_rk))
                return RD_KAFKA_RESP_ERR__FATAL;

        if (!rktparlist)
                return RD_KAFKA_RESP_ERR_NO_ERROR;

        rkcg->rkcg_flags |= RD_KAFKA_CGRP_F_SUBSCRIPTION;

        if (rd_kafka_topic_partition_list_regex_cnt(rktparlist) > 0)
                rkcg->rkcg_flags |= RD_KAFKA_CGRP_F_WILDCARD_SUBSCRIPTION;

        rkcg->rkcg_subscription = rktparlist;

        rd_kafka_cgrp_join(rkcg);

        return RD_KAFKA_RESP_ERR_NO_ERROR;
}
/**
 * Set new atomic topic subscription.
 */
Set new atomic topic subscription.
Set new atomic topic subscription .
static rd_kafka_resp_err_t
rd_kafka_cgrp_subscribe (rd_kafka_cgrp_t *rkcg,
                         rd_kafka_topic_partition_list_t *rktparlist) {
	rd_kafka_dbg(rkcg->rkcg_rk, CGRP|RD_KAFKA_DBG_CONSUMER, "SUBSCRIBE",
		     "Group \"%.*s\": subscribe to new %ssubscription "
		     "of %d topics (join state %s)",
		     RD_KAFKAP_STR_PR(rkcg->rkcg_group_id),
		     rktparlist ? "" : "unset ",
		     rktparlist ? rktparlist->cnt : 0,
		     rd_kafka_cgrp_join_state_names[rkcg->rkcg_join_state]);
        if (rkcg->rkcg_rk->rk_conf.enabled_assignor_cnt == 0)
                return RD_KAFKA_RESP_ERR__INVALID_ARG;
        rd_kafka_cgrp_unsubscribe(rkcg,
                                  rktparlist ?
                                  0 :
                                  1);
        if (rd_kafka_fatal_error_code(rkcg->rkcg_rk))
                return RD_KAFKA_RESP_ERR__FATAL;
        if (!rktparlist)
                return RD_KAFKA_RESP_ERR_NO_ERROR;
        rkcg->rkcg_flags |= RD_KAFKA_CGRP_F_SUBSCRIPTION;
        if (rd_kafka_topic_partition_list_regex_cnt(rktparlist) > 0)
                rkcg->rkcg_flags |= RD_KAFKA_CGRP_F_WILDCARD_SUBSCRIPTION;
        rkcg->rkcg_subscription = rktparlist;
        rd_kafka_cgrp_join(rkcg);
        return RD_KAFKA_RESP_ERR_NO_ERROR;
}
static rd_kafka_resp_err_t rd_kafka_cgrp_subscribe ( rd_kafka_cgrp_t * rkcg , rd_kafka_topic_partition_list_t * rktparlist ) { rd_kafka_dbg ( rkcg -> rkcg_rk , CGRP | RD_KAFKA_DBG_CONSUMER , " " , " \" \" " " " , RD_KAFKAP_STR_PR ( rkcg -> rkcg_group_id ) , rktparlist ? " " : " " , rktparlist ? rktparlist -> cnt : 0 , rd_kafka_cgrp_join_state_names [ rkcg -> rkcg_join_state ] ) ; if ( rkcg -> rkcg_rk -> rk_conf . enabled_assignor_cnt == 0 ) return RD_KAFKA_RESP_ERR__INVALID_ARG ; rd_kafka_cgrp_unsubscribe ( rkcg , rktparlist ? 0 : 1 ) ; if ( rd_kafka_fatal_error_code ( rkcg -> rkcg_rk ) ) return RD_KAFKA_RESP_ERR__FATAL ; if ( ! rktparlist ) return RD_KAFKA_RESP_ERR_NO_ERROR ; rkcg -> rkcg_flags |= RD_KAFKA_CGRP_F_SUBSCRIPTION ; if ( rd_kafka_topic_partition_list_regex_cnt ( rktparlist ) > 0 ) rkcg -> rkcg_flags |= RD_KAFKA_CGRP_F_WILDCARD_SUBSCRIPTION ; rkcg -> rkcg_subscription = rktparlist ; rd_kafka_cgrp_join ( rkcg ) ; return RD_KAFKA_RESP_ERR_NO_ERROR ; }
Set new atomic topic subscription.
Set new atomic topic subscription .
/* Remove existing subscription first */ /* dont leave group if new subscription */ /* leave group if no new subscription */ /* If the consumer has raised a fatal error we treat all
         * subscribes as unsubscribe */