src/rdkafka_broker.c
static void rd_kafka_broker_set_api_versions (rd_kafka_broker_t *rkb,
					      struct rd_kafka_ApiVersion *apis,
					      size_t api_cnt) {

        rd_kafka_broker_lock(rkb);

	if (rkb->rkb_ApiVersions)
		rd_free(rkb->rkb_ApiVersions);


	if (!apis) {
		rd_rkb_dbg(rkb, PROTOCOL | RD_KAFKA_DBG_BROKER, "APIVERSION",
			   "Using (configuration fallback) %s protocol features",
			   rkb->rkb_rk->rk_conf.broker_version_fallback);


		rd_kafka_get_legacy_ApiVersions(rkb->rkb_rk->rk_conf.
						broker_version_fallback,
						&apis, &api_cnt,
						rkb->rkb_rk->rk_conf.
						broker_version_fallback);

		/* Make a copy to store on broker. */
		rd_kafka_ApiVersions_copy(apis, api_cnt, &apis, &api_cnt);
	}

	rkb->rkb_ApiVersions = apis;
	rkb->rkb_ApiVersions_cnt = api_cnt;

	/* Update feature set based on supported broker APIs. */
	rd_kafka_broker_features_set(rkb,
				     rd_kafka_features_check(rkb, apis, api_cnt));

        rd_kafka_broker_unlock(rkb);
}
/**
 * @brief Specify API versions to use for this connection.
 *
 * @param apis is an allocated list of supported partitions.
 *        If NULL the default set will be used based on the
 *        \p broker.version.fallback property.
 * @param api_cnt number of elements in \p apis
 *
 * @remark \p rkb takes ownership of \p apis.
 *
 * @locality Broker thread
 * @locks none
 */
@brief Specify API versions to use for this connection.
@param apis is an allocated list of supported partitions.
If NULL the default set will be used based on the
\p broker.version.fallback property.
@param api_cnt number of elements in \p apis

@remark \p rkb takes ownership of \p apis.

@locality Broker thread
@locks none
@brief Specify API versions to use for this connection . @param apis is an allocated list of supported partitions . If NULL the default set will be used based on the \ p broker . version . fallback property . @param api_cnt number of elements in \ p apis @remark \ p rkb takes ownership of \ p apis . @locality Broker thread @locks none
static void rd_kafka_broker_set_api_versions (rd_kafka_broker_t *rkb,
					      struct rd_kafka_ApiVersion *apis,
					      size_t api_cnt) {
        rd_kafka_broker_lock(rkb);
	if (rkb->rkb_ApiVersions)
		rd_free(rkb->rkb_ApiVersions);
	if (!apis) {
		rd_rkb_dbg(rkb, PROTOCOL | RD_KAFKA_DBG_BROKER, "APIVERSION",
			   "Using (configuration fallback) %s protocol features",
			   rkb->rkb_rk->rk_conf.broker_version_fallback);
		rd_kafka_get_legacy_ApiVersions(rkb->rkb_rk->rk_conf.
						broker_version_fallback,
						&apis, &api_cnt,
						rkb->rkb_rk->rk_conf.
						broker_version_fallback);
		rd_kafka_ApiVersions_copy(apis, api_cnt, &apis, &api_cnt);
	}
	rkb->rkb_ApiVersions = apis;
	rkb->rkb_ApiVersions_cnt = api_cnt;
	rd_kafka_broker_features_set(rkb,
				     rd_kafka_features_check(rkb, apis, api_cnt));
        rd_kafka_broker_unlock(rkb);
}
static void rd_kafka_broker_set_api_versions ( rd_kafka_broker_t * rkb , struct rd_kafka_ApiVersion * apis , size_t api_cnt ) { rd_kafka_broker_lock ( rkb ) ; if ( rkb -> rkb_ApiVersions ) rd_free ( rkb -> rkb_ApiVersions ) ; if ( ! apis ) { rd_rkb_dbg ( rkb , PROTOCOL | RD_KAFKA_DBG_BROKER , " " , " " , rkb -> rkb_rk -> rk_conf . broker_version_fallback ) ; rd_kafka_get_legacy_ApiVersions ( rkb -> rkb_rk -> rk_conf . broker_version_fallback , & apis , & api_cnt , rkb -> rkb_rk -> rk_conf . broker_version_fallback ) ; rd_kafka_ApiVersions_copy ( apis , api_cnt , & apis , & api_cnt ) ; } rkb -> rkb_ApiVersions = apis ; rkb -> rkb_ApiVersions_cnt = api_cnt ; rd_kafka_broker_features_set ( rkb , rd_kafka_features_check ( rkb , apis , api_cnt ) ) ; rd_kafka_broker_unlock ( rkb ) ; }
@brief Specify API versions to use for this connection.
@brief Specify API versions to use for this connection .
/* Make a copy to store on broker. */ /* Update feature set based on supported broker APIs. */