plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_assignor.c
static rd_kafka_resp_err_t
rd_kafka_assignor_add (rd_kafka_t *rk,
		       rd_kafka_assignor_t **rkasp,
                       const char *protocol_type,
                       const char *protocol_name,
                       rd_kafka_resp_err_t (*assign_cb) (
                               rd_kafka_t *rk,
                               const char *member_id,
                               const char *protocol_name,
                               const rd_kafka_metadata_t *metadata,
                               rd_kafka_group_member_t *members,
                               size_t member_cnt,
                               rd_kafka_assignor_topic_t **eligible_topics,
                               size_t eligible_topic_cnt,
                               char *errstr, size_t errstr_size, void *opaque),
                       void *opaque) {
        rd_kafka_assignor_t *rkas;

	if (rkasp)
		*rkasp = NULL;

        if (rd_kafkap_str_cmp_str(rk->rk_conf.group_protocol_type,
                                  protocol_type))
                return RD_KAFKA_RESP_ERR__UNKNOWN_PROTOCOL;

        /* Dont overwrite application assignors */
        if ((rkas = rd_kafka_assignor_find(rk, protocol_name))) {
		if (rkasp)
			*rkasp = rkas;
		return RD_KAFKA_RESP_ERR__CONFLICT;
	}

        rkas = rd_calloc(1, sizeof(*rkas));

        rkas->rkas_protocol_name    = rd_kafkap_str_new(protocol_name, -1);
        rkas->rkas_protocol_type    = rd_kafkap_str_new(protocol_type, -1);
        rkas->rkas_assign_cb        = assign_cb;
        rkas->rkas_get_metadata_cb  = rd_kafka_assignor_get_metadata;
        rkas->rkas_opaque = opaque;

        rd_list_add(&rk->rk_conf.partition_assignors, rkas);

	if (rkasp)
		*rkasp = rkas;

        return RD_KAFKA_RESP_ERR_NO_ERROR;
}
/**
 * Add an assignor, overwriting any previous one with the same protocol_name.
 */
Add an assignor, overwriting any previous one with the same protocol_name.
Add an assignor overwriting any previous one with the same protocol_name .
static rd_kafka_resp_err_t
rd_kafka_assignor_add (rd_kafka_t *rk,
		       rd_kafka_assignor_t **rkasp,
                       const char *protocol_type,
                       const char *protocol_name,
                       rd_kafka_resp_err_t (*assign_cb) (
                               rd_kafka_t *rk,
                               const char *member_id,
                               const char *protocol_name,
                               const rd_kafka_metadata_t *metadata,
                               rd_kafka_group_member_t *members,
                               size_t member_cnt,
                               rd_kafka_assignor_topic_t **eligible_topics,
                               size_t eligible_topic_cnt,
                               char *errstr, size_t errstr_size, void *opaque),
                       void *opaque) {
        rd_kafka_assignor_t *rkas;
	if (rkasp)
		*rkasp = NULL;
        if (rd_kafkap_str_cmp_str(rk->rk_conf.group_protocol_type,
                                  protocol_type))
                return RD_KAFKA_RESP_ERR__UNKNOWN_PROTOCOL;
        if ((rkas = rd_kafka_assignor_find(rk, protocol_name))) {
		if (rkasp)
			*rkasp = rkas;
		return RD_KAFKA_RESP_ERR__CONFLICT;
	}
        rkas = rd_calloc(1, sizeof(*rkas));
        rkas->rkas_protocol_name    = rd_kafkap_str_new(protocol_name, -1);
        rkas->rkas_protocol_type    = rd_kafkap_str_new(protocol_type, -1);
        rkas->rkas_assign_cb        = assign_cb;
        rkas->rkas_get_metadata_cb  = rd_kafka_assignor_get_metadata;
        rkas->rkas_opaque = opaque;
        rd_list_add(&rk->rk_conf.partition_assignors, rkas);
	if (rkasp)
		*rkasp = rkas;
        return RD_KAFKA_RESP_ERR_NO_ERROR;
}
static rd_kafka_resp_err_t rd_kafka_assignor_add ( rd_kafka_t * rk , rd_kafka_assignor_t * * rkasp , const char * protocol_type , const char * protocol_name , rd_kafka_resp_err_t ( * assign_cb ) ( rd_kafka_t * rk , const char * member_id , const char * protocol_name , const rd_kafka_metadata_t * metadata , rd_kafka_group_member_t * members , size_t member_cnt , rd_kafka_assignor_topic_t * * eligible_topics , size_t eligible_topic_cnt , char * errstr , size_t errstr_size , void * opaque ) , void * opaque ) { rd_kafka_assignor_t * rkas ; if ( rkasp ) * rkasp = NULL ; if ( rd_kafkap_str_cmp_str ( rk -> rk_conf . group_protocol_type , protocol_type ) ) return RD_KAFKA_RESP_ERR__UNKNOWN_PROTOCOL ; if ( ( rkas = rd_kafka_assignor_find ( rk , protocol_name ) ) ) { if ( rkasp ) * rkasp = rkas ; return RD_KAFKA_RESP_ERR__CONFLICT ; } rkas = rd_calloc ( 1 , sizeof ( * rkas ) ) ; rkas -> rkas_protocol_name = rd_kafkap_str_new ( protocol_name , -1 ) ; rkas -> rkas_protocol_type = rd_kafkap_str_new ( protocol_type , -1 ) ; rkas -> rkas_assign_cb = assign_cb ; rkas -> rkas_get_metadata_cb = rd_kafka_assignor_get_metadata ; rkas -> rkas_opaque = opaque ; rd_list_add ( & rk -> rk_conf . partition_assignors , rkas ) ; if ( rkasp ) * rkasp = rkas ; return RD_KAFKA_RESP_ERR_NO_ERROR ; }
Add an assignor, overwriting any previous one with the same protocol_name.
Add an assignor overwriting any previous one with the same protocol_name .
/* Dont overwrite application assignors */