plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_cgrp.c
void rd_kafka_cgrp_coord_query (rd_kafka_cgrp_t *rkcg,
				const char *reason) {
	rd_kafka_broker_t *rkb;
        rd_kafka_resp_err_t err;

        rkb = rd_kafka_broker_any_usable(rkcg->rkcg_rk,
                                         RD_POLL_NOWAIT,
                                         RD_DO_LOCK,
                                         RD_KAFKA_FEATURE_BROKER_GROUP_COORD,
                                         "coordinator query");

	if (!rkb) {
		/* Reset the interval because there were no brokers. When a
		 * broker becomes available, we want to query it immediately. */
		rd_interval_reset(&rkcg->rkcg_coord_query_intvl);
		rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "CGRPQUERY",
			     "Group \"%.*s\": "
			     "no broker available for coordinator query: %s",
			     RD_KAFKAP_STR_PR(rkcg->rkcg_group_id), reason);
		return;
	}

        rd_rkb_dbg(rkb, CGRP, "CGRPQUERY",
                   "Group \"%.*s\": querying for coordinator: %s",
                   RD_KAFKAP_STR_PR(rkcg->rkcg_group_id), reason);

        err = rd_kafka_FindCoordinatorRequest(
                rkb, RD_KAFKA_COORD_GROUP, rkcg->rkcg_group_id->str,
                RD_KAFKA_REPLYQ(rkcg->rkcg_ops, 0),
                rd_kafka_cgrp_handle_FindCoordinator, rkcg);

        if (err) {
                rd_rkb_dbg(rkb, CGRP, "CGRPQUERY",
                           "Group \"%.*s\": "
                           "unable to send coordinator query: %s",
                           RD_KAFKAP_STR_PR(rkcg->rkcg_group_id),
                           rd_kafka_err2str(err));
                rd_kafka_broker_destroy(rkb);
                return;
        }

        if (rkcg->rkcg_state == RD_KAFKA_CGRP_STATE_QUERY_COORD)
                rd_kafka_cgrp_set_state(rkcg, RD_KAFKA_CGRP_STATE_WAIT_COORD);

	rd_kafka_broker_destroy(rkb);

        /* Back off the next intervalled query since we just sent one. */
        rd_interval_reset_to_now(&rkcg->rkcg_coord_query_intvl, 0);
}
/**
 * Query for coordinator.
 * Ask any broker in state UP
 *
 * Locality: main thread
 */
Query for coordinator.
Ask any broker in state UP

main thread
Query for coordinator . Ask any broker in state UP main thread
void rd_kafka_cgrp_coord_query (rd_kafka_cgrp_t *rkcg,
				const char *reason) {
	rd_kafka_broker_t *rkb;
        rd_kafka_resp_err_t err;
        rkb = rd_kafka_broker_any_usable(rkcg->rkcg_rk,
                                         RD_POLL_NOWAIT,
                                         RD_DO_LOCK,
                                         RD_KAFKA_FEATURE_BROKER_GROUP_COORD,
                                         "coordinator query");
	if (!rkb) {
		rd_interval_reset(&rkcg->rkcg_coord_query_intvl);
		rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "CGRPQUERY",
			     "Group \"%.*s\": "
			     "no broker available for coordinator query: %s",
			     RD_KAFKAP_STR_PR(rkcg->rkcg_group_id), reason);
		return;
	}
        rd_rkb_dbg(rkb, CGRP, "CGRPQUERY",
                   "Group \"%.*s\": querying for coordinator: %s",
                   RD_KAFKAP_STR_PR(rkcg->rkcg_group_id), reason);
        err = rd_kafka_FindCoordinatorRequest(
                rkb, RD_KAFKA_COORD_GROUP, rkcg->rkcg_group_id->str,
                RD_KAFKA_REPLYQ(rkcg->rkcg_ops, 0),
                rd_kafka_cgrp_handle_FindCoordinator, rkcg);
        if (err) {
                rd_rkb_dbg(rkb, CGRP, "CGRPQUERY",
                           "Group \"%.*s\": "
                           "unable to send coordinator query: %s",
                           RD_KAFKAP_STR_PR(rkcg->rkcg_group_id),
                           rd_kafka_err2str(err));
                rd_kafka_broker_destroy(rkb);
                return;
        }
        if (rkcg->rkcg_state == RD_KAFKA_CGRP_STATE_QUERY_COORD)
                rd_kafka_cgrp_set_state(rkcg, RD_KAFKA_CGRP_STATE_WAIT_COORD);
	rd_kafka_broker_destroy(rkb);
        rd_interval_reset_to_now(&rkcg->rkcg_coord_query_intvl, 0);
}
void rd_kafka_cgrp_coord_query ( rd_kafka_cgrp_t * rkcg , const char * reason ) { rd_kafka_broker_t * rkb ; rd_kafka_resp_err_t err ; rkb = rd_kafka_broker_any_usable ( rkcg -> rkcg_rk , RD_POLL_NOWAIT , RD_DO_LOCK , RD_KAFKA_FEATURE_BROKER_GROUP_COORD , " " ) ; if ( ! rkb ) { rd_interval_reset ( & rkcg -> rkcg_coord_query_intvl ) ; rd_kafka_dbg ( rkcg -> rkcg_rk , CGRP , " " , " \" \" " " " , RD_KAFKAP_STR_PR ( rkcg -> rkcg_group_id ) , reason ) ; return ; } rd_rkb_dbg ( rkb , CGRP , " " , " \" \" " , RD_KAFKAP_STR_PR ( rkcg -> rkcg_group_id ) , reason ) ; err = rd_kafka_FindCoordinatorRequest ( rkb , RD_KAFKA_COORD_GROUP , rkcg -> rkcg_group_id -> str , RD_KAFKA_REPLYQ ( rkcg -> rkcg_ops , 0 ) , rd_kafka_cgrp_handle_FindCoordinator , rkcg ) ; if ( err ) { rd_rkb_dbg ( rkb , CGRP , " " , " \" \" " " " , RD_KAFKAP_STR_PR ( rkcg -> rkcg_group_id ) , rd_kafka_err2str ( err ) ) ; rd_kafka_broker_destroy ( rkb ) ; return ; } if ( rkcg -> rkcg_state == RD_KAFKA_CGRP_STATE_QUERY_COORD ) rd_kafka_cgrp_set_state ( rkcg , RD_KAFKA_CGRP_STATE_WAIT_COORD ) ; rd_kafka_broker_destroy ( rkb ) ; rd_interval_reset_to_now ( & rkcg -> rkcg_coord_query_intvl , 0 ) ; }
Query for coordinator.
Query for coordinator .
/* Reset the interval because there were no brokers. When a
		 * broker becomes available, we want to query it immediately. */ /* Back off the next intervalled query since we just sent one. */