plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_assignor.c
static int rd_kafka_member_subscription_match (
        rd_kafka_cgrp_t *rkcg,
        rd_kafka_group_member_t *rkgm,
        const rd_kafka_metadata_topic_t *topic_metadata,
        rd_kafka_assignor_topic_t *eligible_topic) {
        int i;
        int has_regex = 0;
        int matched = 0;

        /* Match against member's subscription. */
        for (i = 0 ; i < rkgm->rkgm_subscription->cnt ; i++) {
                const rd_kafka_topic_partition_t *rktpar =
                        &rkgm->rkgm_subscription->elems[i];
		int matched_by_regex = 0;

		if (rd_kafka_topic_partition_match(rkcg->rkcg_rk, rkgm, rktpar,
						   topic_metadata->topic,
						   &matched_by_regex)) {
			rd_list_add(&rkgm->rkgm_eligible,
				    (void *)topic_metadata);
			matched++;
			has_regex += matched_by_regex;
		}
	}

        if (matched)
                rd_list_add(&eligible_topic->members, rkgm);

        if (!has_regex &&
            rd_list_cnt(&rkgm->rkgm_eligible) == rkgm->rkgm_subscription->cnt)
                return 1; /* All subscriptions matched */
        else
                return 0;
}
/**
 * Returns 1 if all subscriptions are satifised for this member, else 0.
 */
Returns 1 if all subscriptions are satifised for this member, else 0.
Returns 1 if all subscriptions are satifised for this member else 0 .
static int rd_kafka_member_subscription_match (
        rd_kafka_cgrp_t *rkcg,
        rd_kafka_group_member_t *rkgm,
        const rd_kafka_metadata_topic_t *topic_metadata,
        rd_kafka_assignor_topic_t *eligible_topic) {
        int i;
        int has_regex = 0;
        int matched = 0;
        for (i = 0 ; i < rkgm->rkgm_subscription->cnt ; i++) {
                const rd_kafka_topic_partition_t *rktpar =
                        &rkgm->rkgm_subscription->elems[i];
		int matched_by_regex = 0;
		if (rd_kafka_topic_partition_match(rkcg->rkcg_rk, rkgm, rktpar,
						   topic_metadata->topic,
						   &matched_by_regex)) {
			rd_list_add(&rkgm->rkgm_eligible,
				    (void *)topic_metadata);
			matched++;
			has_regex += matched_by_regex;
		}
	}
        if (matched)
                rd_list_add(&eligible_topic->members, rkgm);
        if (!has_regex &&
            rd_list_cnt(&rkgm->rkgm_eligible) == rkgm->rkgm_subscription->cnt)
                return 1; 
        else
                return 0;
}
static int rd_kafka_member_subscription_match ( rd_kafka_cgrp_t * rkcg , rd_kafka_group_member_t * rkgm , const rd_kafka_metadata_topic_t * topic_metadata , rd_kafka_assignor_topic_t * eligible_topic ) { int i ; int has_regex = 0 ; int matched = 0 ; for ( i = 0 ; i < rkgm -> rkgm_subscription -> cnt ; i ++ ) { const rd_kafka_topic_partition_t * rktpar = & rkgm -> rkgm_subscription -> elems [ i ] ; int matched_by_regex = 0 ; if ( rd_kafka_topic_partition_match ( rkcg -> rkcg_rk , rkgm , rktpar , topic_metadata -> topic , & matched_by_regex ) ) { rd_list_add ( & rkgm -> rkgm_eligible , ( void * ) topic_metadata ) ; matched ++ ; has_regex += matched_by_regex ; } } if ( matched ) rd_list_add ( & eligible_topic -> members , rkgm ) ; if ( ! has_regex && rd_list_cnt ( & rkgm -> rkgm_eligible ) == rkgm -> rkgm_subscription -> cnt ) return 1 ; else return 0 ; }
Returns 1 if all subscriptions are satifised for this member, else 0.
Returns 1 if all subscriptions are satifised for this member else 0 .
/* Match against member's subscription. */ /* All subscriptions matched */