src/rdkafka_broker.c
void rd_kafka_broker_connect_up (rd_kafka_broker_t *rkb) {

	rkb->rkb_max_inflight = rkb->rkb_rk->rk_conf.max_inflight;
        rkb->rkb_err.err = 0;

	rd_kafka_broker_lock(rkb);
	rd_kafka_broker_set_state(rkb, RD_KAFKA_BROKER_STATE_UP);
	rd_kafka_broker_unlock(rkb);

        /* Request metadata (async):
         * try locally known topics first and if there are none try
         * getting just the broker list. */
        if (rd_kafka_metadata_refresh_known_topics(NULL, rkb, 0/*dont force*/,
                                                   "connected") ==
            RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC)
                rd_kafka_metadata_refresh_brokers(NULL, rkb, "connected");
}
/**
 * @brief Call when connection is ready to transition to fully functional
 *        UP state.
 *
 * @locality Broker thread
 */
@brief Call when connection is ready to transition to fully functional
UP state.

@locality Broker thread
@brief Call when connection is ready to transition to fully functional UP state . @locality Broker thread
void rd_kafka_broker_connect_up (rd_kafka_broker_t *rkb) {
	rkb->rkb_max_inflight = rkb->rkb_rk->rk_conf.max_inflight;
        rkb->rkb_err.err = 0;
	rd_kafka_broker_lock(rkb);
	rd_kafka_broker_set_state(rkb, RD_KAFKA_BROKER_STATE_UP);
	rd_kafka_broker_unlock(rkb);
        if (rd_kafka_metadata_refresh_known_topics(NULL, rkb, 0,
                                                   "connected") ==
            RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC)
                rd_kafka_metadata_refresh_brokers(NULL, rkb, "connected");
}
void rd_kafka_broker_connect_up ( rd_kafka_broker_t * rkb ) { rkb -> rkb_max_inflight = rkb -> rkb_rk -> rk_conf . max_inflight ; rkb -> rkb_err . err = 0 ; rd_kafka_broker_lock ( rkb ) ; rd_kafka_broker_set_state ( rkb , RD_KAFKA_BROKER_STATE_UP ) ; rd_kafka_broker_unlock ( rkb ) ; if ( rd_kafka_metadata_refresh_known_topics ( NULL , rkb , 0 , " " ) == RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC ) rd_kafka_metadata_refresh_brokers ( NULL , rkb , " " ) ; }
@brief Call when connection is ready to transition to fully functional
        UP state.
@brief Call when connection is ready to transition to fully functional UP state .
/* Request metadata (async):
         * try locally known topics first and if there are none try
         * getting just the broker list. */ /*dont force*/