src/rdkafka_broker.c
void rd_kafka_broker_connect_done (rd_kafka_broker_t *rkb, const char *errstr) {

	if (errstr) {
		/* Connect failed */
                rd_kafka_broker_fail(rkb,
                                     errno != 0 && rkb->rkb_err.err == errno ?
                                     LOG_DEBUG : LOG_ERR,
                                     RD_KAFKA_RESP_ERR__TRANSPORT,
                                     "%s", errstr);
		return;
	}

	/* Connect succeeded */
	rkb->rkb_connid++;
	rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_PROTOCOL,
		   "CONNECTED", "Connected (#%d)", rkb->rkb_connid);
	rkb->rkb_err.err = 0;
	rkb->rkb_max_inflight = 1; /* Hold back other requests until
				    * ApiVersion, SaslHandshake, etc
				    * are done. */

	rd_kafka_transport_poll_set(rkb->rkb_transport, POLLIN);

	if (rkb->rkb_rk->rk_conf.api_version_request &&
	    rd_interval_immediate(&rkb->rkb_ApiVersion_fail_intvl, 0, 0) > 0) {
		/* Use ApiVersion to query broker for supported API versions. */
		rd_kafka_broker_feature_enable(rkb, RD_KAFKA_FEATURE_APIVERSION);
	}

        if (!(rkb->rkb_features & RD_KAFKA_FEATURE_APIVERSION)) {
                /* Use configured broker.version.fallback to
                 * figure out API versions.
                 * In case broker.version.fallback indicates a version
                 * that supports ApiVersionRequest it will update
                 * rkb_features to have FEATURE_APIVERSION set which will
                 * trigger an ApiVersionRequest below. */
                rd_kafka_broker_set_api_versions(rkb, NULL, 0);
        }

	if (rkb->rkb_features & RD_KAFKA_FEATURE_APIVERSION) {
		/* Query broker for supported API versions.
		 * This may fail with a disconnect on non-supporting brokers
		 * so hold off any other requests until we get a response,
		 * and if the connection is torn down we disable this feature. */
		rd_kafka_broker_lock(rkb);
		rd_kafka_broker_set_state(rkb,RD_KAFKA_BROKER_STATE_APIVERSION_QUERY);
		rd_kafka_broker_unlock(rkb);

		rd_kafka_ApiVersionRequest(
			rkb, RD_KAFKA_NO_REPLYQ,
			rd_kafka_broker_handle_ApiVersion, NULL);
	} else {
		/* Authenticate if necessary */
		rd_kafka_broker_connect_auth(rkb);
	}

}
/**
 * Call when asynchronous connection attempt completes, either succesfully
 * (if errstr is NULL) or fails.
 *
 * Locality: broker thread
 */
Call when asynchronous connection attempt completes, either succesfully
(if errstr is NULL) or fails.

broker thread
Call when asynchronous connection attempt completes either succesfully ( if errstr is NULL ) or fails . broker thread
void rd_kafka_broker_connect_done (rd_kafka_broker_t *rkb, const char *errstr) {
	if (errstr) {
                rd_kafka_broker_fail(rkb,
                                     errno != 0 && rkb->rkb_err.err == errno ?
                                     LOG_DEBUG : LOG_ERR,
                                     RD_KAFKA_RESP_ERR__TRANSPORT,
                                     "%s", errstr);
		return;
	}
	rkb->rkb_connid++;
	rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_PROTOCOL,
		   "CONNECTED", "Connected (#%d)", rkb->rkb_connid);
	rkb->rkb_err.err = 0;
	rkb->rkb_max_inflight = 1; 
	rd_kafka_transport_poll_set(rkb->rkb_transport, POLLIN);
	if (rkb->rkb_rk->rk_conf.api_version_request &&
	    rd_interval_immediate(&rkb->rkb_ApiVersion_fail_intvl, 0, 0) > 0) {
		rd_kafka_broker_feature_enable(rkb, RD_KAFKA_FEATURE_APIVERSION);
	}
        if (!(rkb->rkb_features & RD_KAFKA_FEATURE_APIVERSION)) {
                rd_kafka_broker_set_api_versions(rkb, NULL, 0);
        }
	if (rkb->rkb_features & RD_KAFKA_FEATURE_APIVERSION) {
		rd_kafka_broker_lock(rkb);
		rd_kafka_broker_set_state(rkb,RD_KAFKA_BROKER_STATE_APIVERSION_QUERY);
		rd_kafka_broker_unlock(rkb);
		rd_kafka_ApiVersionRequest(
			rkb, RD_KAFKA_NO_REPLYQ,
			rd_kafka_broker_handle_ApiVersion, NULL);
	} else {
		rd_kafka_broker_connect_auth(rkb);
	}
}
void rd_kafka_broker_connect_done ( rd_kafka_broker_t * rkb , const char * errstr ) { if ( errstr ) { rd_kafka_broker_fail ( rkb , errno != 0 && rkb -> rkb_err . err == errno ? LOG_DEBUG : LOG_ERR , RD_KAFKA_RESP_ERR__TRANSPORT , " " , errstr ) ; return ; } rkb -> rkb_connid ++ ; rd_rkb_dbg ( rkb , BROKER | RD_KAFKA_DBG_PROTOCOL , " " , " " , rkb -> rkb_connid ) ; rkb -> rkb_err . err = 0 ; rkb -> rkb_max_inflight = 1 ; rd_kafka_transport_poll_set ( rkb -> rkb_transport , POLLIN ) ; if ( rkb -> rkb_rk -> rk_conf . api_version_request && rd_interval_immediate ( & rkb -> rkb_ApiVersion_fail_intvl , 0 , 0 ) > 0 ) { rd_kafka_broker_feature_enable ( rkb , RD_KAFKA_FEATURE_APIVERSION ) ; } if ( ! ( rkb -> rkb_features & RD_KAFKA_FEATURE_APIVERSION ) ) { rd_kafka_broker_set_api_versions ( rkb , NULL , 0 ) ; } if ( rkb -> rkb_features & RD_KAFKA_FEATURE_APIVERSION ) { rd_kafka_broker_lock ( rkb ) ; rd_kafka_broker_set_state ( rkb , RD_KAFKA_BROKER_STATE_APIVERSION_QUERY ) ; rd_kafka_broker_unlock ( rkb ) ; rd_kafka_ApiVersionRequest ( rkb , RD_KAFKA_NO_REPLYQ , rd_kafka_broker_handle_ApiVersion , NULL ) ; } else { rd_kafka_broker_connect_auth ( rkb ) ; } }
Call when asynchronous connection attempt completes, either succesfully
 (if errstr is NULL) or fails.
Call when asynchronous connection attempt completes either succesfully ( if errstr is NULL ) or fails .
/* Connect failed */ /* Connect succeeded */ /* Hold back other requests until
				    * ApiVersion, SaslHandshake, etc
				    * are done. */ /* Use ApiVersion to query broker for supported API versions. */ /* Use configured broker.version.fallback to
                 * figure out API versions.
                 * In case broker.version.fallback indicates a version
                 * that supports ApiVersionRequest it will update
                 * rkb_features to have FEATURE_APIVERSION set which will
                 * trigger an ApiVersionRequest below. */ /* Query broker for supported API versions.
		 * This may fail with a disconnect on non-supporting brokers
		 * so hold off any other requests until we get a response,
		 * and if the connection is torn down we disable this feature. */ /* Authenticate if necessary */