plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_assignor.c
int rd_kafka_assignors_init (rd_kafka_t *rk, char *errstr, size_t errstr_size) {
	char *wanted;
	char *s;

        rd_list_init(&rk->rk_conf.partition_assignors, 2,
                     (void *)rd_kafka_assignor_destroy);

	rd_strdupa(&wanted, rk->rk_conf.partition_assignment_strategy);

	s = wanted;
	while (*s) {
		rd_kafka_assignor_t *rkas = NULL;
		char *t;

		/* Left trim */
		while (*s == ' ' || *s == ',')
			s++;

		if ((t = strchr(s, ','))) {
			*t = '\0';
			t++;
		} else {
			t = s + strlen(s);
		}

		/* Right trim */
		rtrim(s);

		/* Match builtin consumer assignors */
		if (!strcmp(s, "range"))
			rd_kafka_assignor_add(
				rk, &rkas, "consumer", "range",
				rd_kafka_range_assignor_assign_cb,
				NULL);
		else if (!strcmp(s, "roundrobin"))
			rd_kafka_assignor_add(
				rk, &rkas, "consumer", "roundrobin",
				rd_kafka_roundrobin_assignor_assign_cb,
				NULL);
		else {
			rd_snprintf(errstr, errstr_size,
				    "Unsupported partition.assignment.strategy:"
				    " %s", s);
			return -1;
		}

		if (rkas) {
			if (!rkas->rkas_enabled) {
				rkas->rkas_enabled = 1;
				rk->rk_conf.enabled_assignor_cnt++;
			}
		}

		s = t;
	}

	return 0;
}
/**
 * Initialize assignor list based on configuration.
 */
Initialize assignor list based on configuration.
Initialize assignor list based on configuration .
int rd_kafka_assignors_init (rd_kafka_t *rk, char *errstr, size_t errstr_size) {
	char *wanted;
	char *s;
        rd_list_init(&rk->rk_conf.partition_assignors, 2,
                     (void *)rd_kafka_assignor_destroy);
	rd_strdupa(&wanted, rk->rk_conf.partition_assignment_strategy);
	s = wanted;
	while (*s) {
		rd_kafka_assignor_t *rkas = NULL;
		char *t;
		while (*s == ' ' || *s == ',')
			s++;
		if ((t = strchr(s, ','))) {
			*t = '\0';
			t++;
		} else {
			t = s + strlen(s);
		}
		rtrim(s);
		if (!strcmp(s, "range"))
			rd_kafka_assignor_add(
				rk, &rkas, "consumer", "range",
				rd_kafka_range_assignor_assign_cb,
				NULL);
		else if (!strcmp(s, "roundrobin"))
			rd_kafka_assignor_add(
				rk, &rkas, "consumer", "roundrobin",
				rd_kafka_roundrobin_assignor_assign_cb,
				NULL);
		else {
			rd_snprintf(errstr, errstr_size,
				    "Unsupported partition.assignment.strategy:"
				    " %s", s);
			return -1;
		}
		if (rkas) {
			if (!rkas->rkas_enabled) {
				rkas->rkas_enabled = 1;
				rk->rk_conf.enabled_assignor_cnt++;
			}
		}
		s = t;
	}
	return 0;
}
int rd_kafka_assignors_init ( rd_kafka_t * rk , char * errstr , size_t errstr_size ) { char * wanted ; char * s ; rd_list_init ( & rk -> rk_conf . partition_assignors , 2 , ( void * ) rd_kafka_assignor_destroy ) ; rd_strdupa ( & wanted , rk -> rk_conf . partition_assignment_strategy ) ; s = wanted ; while ( * s ) { rd_kafka_assignor_t * rkas = NULL ; char * t ; while ( * s == ' ' || * s == ' ' ) s ++ ; if ( ( t = strchr ( s , ' ' ) ) ) { * t = ' \0 ' ; t ++ ; } else { t = s + strlen ( s ) ; } rtrim ( s ) ; if ( ! strcmp ( s , " " ) ) rd_kafka_assignor_add ( rk , & rkas , " " , " " , rd_kafka_range_assignor_assign_cb , NULL ) ; else if ( ! strcmp ( s , " " ) ) rd_kafka_assignor_add ( rk , & rkas , " " , " " , rd_kafka_roundrobin_assignor_assign_cb , NULL ) ; else { rd_snprintf ( errstr , errstr_size , " " " " , s ) ; return -1 ; } if ( rkas ) { if ( ! rkas -> rkas_enabled ) { rkas -> rkas_enabled = 1 ; rk -> rk_conf . enabled_assignor_cnt ++ ; } } s = t ; } return 0 ; }
Initialize assignor list based on configuration.
Initialize assignor list based on configuration .
/* Left trim */ /* Right trim */ /* Match builtin consumer assignors */