thirdparty/librdkafka-0.11.1/src/rdkafka_feature.c - nifi-minifi-cpp - Git at Google

 /*
  * librdkafka - Apache Kafka C library
  *
  * Copyright (c) 2016, Magnus Edenhill
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions are met:
  *
  * 1. Redistributions of source code must retain the above copyright notice,
  *    this list of conditions and the following disclaimer.
  * 2. Redistributions in binary form must reproduce the above copyright notice,
  *    this list of conditions and the following disclaimer in the documentation
  *    and/or other materials provided with the distribution.
  *
  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  * POSSIBILITY OF SUCH DAMAGE.
  */


 #include "rdkafka_int.h"
 #include "rdkafka_feature.h"

 #include <stdlib.h>

 static const char *rd_kafka_feature_names[] = {
 	"MsgVer1",
 	"ApiVersion",
 	"BrokerBalancedConsumer",
 	"ThrottleTime",
 	"Sasl",
 	"SaslHandshake",
 	"BrokerGroupCoordinator",
 	"LZ4",
         "OffsetTime",
         "MsgVer2",
 	NULL
 };


 static const struct rd_kafka_feature_map {
 	/* RD_KAFKA_FEATURE_... */
 	int feature;

 	/* Depends on the following ApiVersions overlapping with
 	 * what the broker supports: */
 	struct rd_kafka_ApiVersion depends[RD_KAFKAP__NUM];

 } rd_kafka_feature_map[] = {
 	/**
 	 * @brief List of features and the ApiVersions they depend on.
 	 *
 	 * The dependency list consists of the ApiKey followed by this
 	 * client's supported minimum and maximum API versions.
 	 * As long as this list and its versions overlaps with the
 	 * broker supported API versions the feature will be enabled.
 	 */
 	{

 		/* @brief >=0.10.0: Message.MagicByte version 1:
 		 * Relative offsets (KIP-31) and message timestamps (KIP-32). */
 		.feature = RD_KAFKA_FEATURE_MSGVER1,
 		.depends = {
 			{ RD_KAFKAP_Produce, 2, 2 },
 			{ RD_KAFKAP_Fetch, 2, 2 },
 			{ -1 },
 		},
 	},
         {
                 /* @brief >=0.11.0: Message.MagicByte version 2 */
                 .feature = RD_KAFKA_FEATURE_MSGVER2,
                 .depends = {
                         { RD_KAFKAP_Produce, 3, 3 },
                         { RD_KAFKAP_Fetch, 4, 4 },
                         { -1 },
                 },
         },
 	{

 		/* @brief >=0.10.0: ApiVersionQuery support.
 		 * @remark This is a bit of chicken-and-egg problem but needs to be
 		 *         set by feature_check() to avoid the feature being cleared
 		 *         even when broker supports it. */
 		.feature = RD_KAFKA_FEATURE_APIVERSION,
 		.depends = {
 			{ RD_KAFKAP_ApiVersion, 0, 0 },
 			{ -1 },
 		},
 	},
 	{
 		/* @brief >=0.8.2.0: Broker-based Group coordinator */
 		.feature = RD_KAFKA_FEATURE_BROKER_GROUP_COORD,
 		.depends = {
 			{ RD_KAFKAP_GroupCoordinator, 0, 0 },
 			{ -1 },
 		},
 	},
 	{
 		/* @brief >=0.9.0: Broker-based balanced consumer groups. */
 		.feature = RD_KAFKA_FEATURE_BROKER_BALANCED_CONSUMER,
 		.depends = {
 			{ RD_KAFKAP_GroupCoordinator, 0, 0 },
 			{ RD_KAFKAP_OffsetCommit, 1, 2 },
 			{ RD_KAFKAP_OffsetFetch, 1, 1 },
 			{ RD_KAFKAP_JoinGroup, 0, 0 },
 			{ RD_KAFKAP_SyncGroup, 0, 0 },
 			{ RD_KAFKAP_Heartbeat, 0, 0 },
 			{ RD_KAFKAP_LeaveGroup, 0, 0 },
 			{ -1 },
 		},
 	},
 	{
 		/* @brief >=0.9.0: ThrottleTime */
 		.feature = RD_KAFKA_FEATURE_THROTTLETIME,
 		.depends = {
 			{ RD_KAFKAP_Produce, 1, 2 },
 			{ RD_KAFKAP_Fetch, 1, 2 },
 			{ -1 },
 		},

         },
         {
                 /* @brief >=0.9.0: SASL (GSSAPI) authentication.
                  * Since SASL is not using the Kafka protocol
                  * we must use something else to map us to the
                  * proper broker version support:
                  * JoinGroup was released along with SASL in 0.9.0. */
                 .feature = RD_KAFKA_FEATURE_SASL_GSSAPI,
                 .depends = {
                         { RD_KAFKAP_JoinGroup, 0, 0 },
                         { -1 },
                 },
         },
         {
                 /* @brief >=0.10.0: SASL mechanism handshake (KIP-43)
                  *                  to automatically support other mechanisms
                  *                  than GSSAPI, such as PLAIN. */
                 .feature = RD_KAFKA_FEATURE_SASL_HANDSHAKE,
                 .depends = {
                         { RD_KAFKAP_SaslHandshake, 0, 0 },
                         { -1 },
                 },
         },
         {
                 /* @brief >=0.8.2: LZ4 compression.
                  * Since LZ4 initially did not rely on a specific API
                  * type or version (it does in >=0.10.0)
                  * we must use something else to map us to the
                  * proper broker version support:
                  * GrooupCoordinator was released in 0.8.2 */
                 .feature = RD_KAFKA_FEATURE_LZ4,
                 .depends = {
                         { RD_KAFKAP_GroupCoordinator, 0, 0 },
                         { -1 },
                 },
         },
         {
                 /* @brief >=0.10.1.0: Offset v1 (KIP-79)
                  * Time-based offset requests */
                 .feature = RD_KAFKA_FEATURE_OFFSET_TIME,
                 .depends = {
                         { RD_KAFKAP_Offset, 1, 1 },
                         { -1 },
                 }
         },
         { .feature = 0 }, /* sentinel */
 };


 /**
  * @brief In absence of KIP-35 support in earlier broker versions we provide hardcoded
  *        lists that corresponds to older broker versions.
  */

 /* >= 0.10.0.0: dummy for all future versions that support ApiVersionRequest */
 static struct rd_kafka_ApiVersion rd_kafka_ApiVersion_Queryable[] = {
 	{ RD_KAFKAP_ApiVersion, 0, 0 }
 };


 /* =~ 0.9.0 */
 static struct rd_kafka_ApiVersion rd_kafka_ApiVersion_0_9_0[] = {
 	{ RD_KAFKAP_Produce, 0, 1 },
 	{ RD_KAFKAP_Fetch, 0, 1 },
 	{ RD_KAFKAP_Offset, 0, 0 },
 	{ RD_KAFKAP_Metadata, 0, 0 },
 	{ RD_KAFKAP_OffsetCommit, 0, 2 },
 	{ RD_KAFKAP_OffsetFetch, 0, 1 },
 	{ RD_KAFKAP_GroupCoordinator, 0, 0 },
 	{ RD_KAFKAP_JoinGroup, 0, 0 },
 	{ RD_KAFKAP_Heartbeat, 0, 0 },
 	{ RD_KAFKAP_LeaveGroup, 0, 0 },
 	{ RD_KAFKAP_SyncGroup, 0, 0 },
 	{ RD_KAFKAP_DescribeGroups, 0, 0 },
 	{ RD_KAFKAP_ListGroups, 0, 0 }
 };

 /* =~ 0.8.2 */
 static struct rd_kafka_ApiVersion rd_kafka_ApiVersion_0_8_2[] = {
 	{ RD_KAFKAP_Produce, 0, 0 },
 	{ RD_KAFKAP_Fetch, 0, 0 },
 	{ RD_KAFKAP_Offset, 0, 0 },
 	{ RD_KAFKAP_Metadata, 0, 0 },
 	{ RD_KAFKAP_OffsetCommit, 0, 1 },
 	{ RD_KAFKAP_OffsetFetch, 0, 1 },
 	{ RD_KAFKAP_GroupCoordinator, 0, 0 }
 };

 /* =~ 0.8.1 */
 static struct rd_kafka_ApiVersion rd_kafka_ApiVersion_0_8_1[] = {
 	{ RD_KAFKAP_Produce, 0, 0 },
 	{ RD_KAFKAP_Fetch, 0, 0 },
 	{ RD_KAFKAP_Offset, 0, 0 },
 	{ RD_KAFKAP_Metadata, 0, 0 },
 	{ RD_KAFKAP_OffsetCommit, 0, 1 },
 	{ RD_KAFKAP_OffsetFetch, 0, 0 }
 };

 /* =~ 0.8.0 */
 static struct rd_kafka_ApiVersion rd_kafka_ApiVersion_0_8_0[] = {
 	{ RD_KAFKAP_Produce, 0, 0 },
 	{ RD_KAFKAP_Fetch, 0, 0 },
 	{ RD_KAFKAP_Offset, 0, 0 },
 	{ RD_KAFKAP_Metadata, 0, 0 }
 };


 /**
  * @brief Returns the ApiVersion list for legacy broker versions that do not
  *        support the ApiVersionQuery request. E.g., brokers <0.10.0.
  *
  * @param broker_version Broker version to match (longest prefix matching).
  * @param use_default If no match is found return the default APIs (but return 0).
  *
  * @returns 1 if \p broker_version was recognized: \p *apisp will point to
  *          the ApiVersion list and *api_cntp will be set to its element count.
  *          0 if \p broker_version was not recognized: \p *apisp remains unchanged.
  *
  */
 int rd_kafka_get_legacy_ApiVersions (const char *broker_version,
 				     struct rd_kafka_ApiVersion **apisp,
 				     size_t *api_cntp, const char *fallback) {
 	static const struct {
 		const char *pfx;
 		struct rd_kafka_ApiVersion *apis;
 		size_t api_cnt;
 	} vermap[] = {
 #define _VERMAP(PFX,APIS) { PFX, APIS, RD_ARRAYSIZE(APIS) }
 		_VERMAP("0.9.0", rd_kafka_ApiVersion_0_9_0),
 		_VERMAP("0.8.2", rd_kafka_ApiVersion_0_8_2),
 		_VERMAP("0.8.1", rd_kafka_ApiVersion_0_8_1),
 		_VERMAP("0.8.0", rd_kafka_ApiVersion_0_8_0),
 		{ "0.7.", NULL }, /* Unsupported */
 		{ "0.6.", NULL }, /* Unsupported */
 		_VERMAP("", rd_kafka_ApiVersion_Queryable),
 		{ NULL }
 	};
 	int i;
 	int fallback_i = -1;
         int ret = 0;

         *apisp = NULL;
         *api_cntp = 0;

 	for (i = 0 ; vermap[i].pfx ; i++) {
 		if (!strncmp(vermap[i].pfx, broker_version, strlen(vermap[i].pfx))) {
 			if (!vermap[i].apis)
 				return 0;
 			*apisp = vermap[i].apis;
 			*api_cntp = vermap[i].api_cnt;
                         ret = 1;
                         break;
 		} else if (fallback && !strcmp(vermap[i].pfx, fallback))
 			fallback_i = i;
 	}

 	if (!*apisp && fallback) {
 		rd_kafka_assert(NULL, fallback_i != -1);
 		*apisp    = vermap[fallback_i].apis;
 		*api_cntp = vermap[fallback_i].api_cnt;
 	}

         return ret;
 }


 /**
  * @returns 1 if the provided broker version (probably)
  *          supports api.version.request.
  */
 int rd_kafka_ApiVersion_is_queryable (const char *broker_version) {
 	struct rd_kafka_ApiVersion *apis;
 	size_t api_cnt;


 	if (!rd_kafka_get_legacy_ApiVersions(broker_version,
 					     &apis, &api_cnt, 0))
 		return 0;

 	return apis == rd_kafka_ApiVersion_Queryable;
 }


 /**
  * @brief Check if match's versions overlaps with \p apis.
  *
  * @returns 1 if true, else 0.
  * @remark \p apis must be sorted using rd_kafka_ApiVersion_key_cmp()
  */
 static RD_INLINE int
 rd_kafka_ApiVersion_check (const struct rd_kafka_ApiVersion *apis, size_t api_cnt,
 			   const struct rd_kafka_ApiVersion *match) {
 	const struct rd_kafka_ApiVersion *api;

 	api = bsearch(match, apis, api_cnt, sizeof(*apis),
 		      rd_kafka_ApiVersion_key_cmp);
 	if (unlikely(!api))
 		return 0;

 	return match->MinVer <= api->MaxVer && api->MinVer <= match->MaxVer;
 }


 /**
  * @brief Compare broker's supported API versions to our feature request map
  *        and enable/disable features accordingly.
  *
  * @param broker_apis Broker's supported APIs. If NULL the
  *        \p broker.version.fallback configuration property will specify a
  *        default legacy version to use.
  * @param broker_api_cnt Number of elements in \p broker_apis
  *
  * @returns the supported features (bitmask) to enable.
  */
 int rd_kafka_features_check (rd_kafka_broker_t *rkb,
 			     struct rd_kafka_ApiVersion *broker_apis,
 			     size_t broker_api_cnt) {
 	int features = 0;
 	int i;

 	/* Scan through features. */
 	for (i = 0 ; rd_kafka_feature_map[i].feature != 0 ; i++) {
 		const struct rd_kafka_ApiVersion *match;
 		int fails = 0;

 		/* For each feature check that all its API dependencies
 		 * can be fullfilled. */

 		for (match = &rd_kafka_feature_map[i].depends[0] ;
 		     match->ApiKey != -1 ; match++) {
 			int r;

 			r = rd_kafka_ApiVersion_check(broker_apis, broker_api_cnt,
 						      match);

 			rd_rkb_dbg(rkb, FEATURE, "APIVERSION",
 				   " Feature %s: %s (%hd..%hd) "
 				   "%ssupported by broker",
 				   rd_kafka_features2str(rd_kafka_feature_map[i].
 							feature),
 				   rd_kafka_ApiKey2str(match->ApiKey),
 				   match->MinVer, match->MaxVer,
 				   r ? "" : "NOT ");

 			fails += !r;
 		}

 		rd_rkb_dbg(rkb, FEATURE, "APIVERSION",
 			   "%s feature %s",
 			   fails ? "Disabling" : "Enabling",
 			   rd_kafka_features2str(rd_kafka_feature_map[i].feature));


 		if (!fails)
 			features |= rd_kafka_feature_map[i].feature;
 	}

 	return features;
 }


 /**
  * @brief Make an allocated and sorted copy of \p src.
  */
 void
 rd_kafka_ApiVersions_copy (const struct rd_kafka_ApiVersion *src,
                            size_t src_cnt,
                            struct rd_kafka_ApiVersion **dstp,
                            size_t *dst_cntp) {
         *dstp = rd_memdup(src, sizeof(*src) * src_cnt);
         *dst_cntp = src_cnt;
         qsort(*dstp, *dst_cntp, sizeof(**dstp), rd_kafka_ApiVersion_key_cmp);
 }


 /**
  * @returns a human-readable feature flag string.
  */
 const char *rd_kafka_features2str (int features) {
 	static RD_TLS char ret[4][128];
 	size_t of = 0;
 	static RD_TLS int reti = 0;
 	int i;

 	reti = (reti + 1) % 4;

 	*ret[reti] = '\0';
 	for (i = 0 ; rd_kafka_feature_names[i] ; i++) {
 		int r;
 		if (!(features & (1 << i)))
 			continue;

 		r = rd_snprintf(ret[reti]+of, sizeof(ret[reti])-of, "%s%s",
 				of == 0 ? "" : ",",
 				rd_kafka_feature_names[i]);
 		if ((size_t)r > sizeof(ret[reti])-of) {
 			/* Out of space */
 			memcpy(&ret[reti][sizeof(ret[reti])-3], "..", 3);
 			break;
 		}

 		of += r;
 	}

 	return ret[reti];
 }
	/*
	* librdkafka - Apache Kafka C library
	*
	* Copyright (c) 2016, Magnus Edenhill
	* All rights reserved.
	*
	* Redistribution and use in source and binary forms, with or without
	* modification, are permitted provided that the following conditions are met:
	*
	* 1. Redistributions of source code must retain the above copyright notice,
	* this list of conditions and the following disclaimer.
	* 2. Redistributions in binary form must reproduce the above copyright notice,
	* this list of conditions and the following disclaimer in the documentation
	* and/or other materials provided with the distribution.
	*
	* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
	* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
	* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
	* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
	* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
	* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
	* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
	* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
	* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
	* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
	* POSSIBILITY OF SUCH DAMAGE.
	*/


	#include "rdkafka_int.h"
	#include "rdkafka_feature.h"

	#include <stdlib.h>

	static const char *rd_kafka_feature_names[] = {
	"MsgVer1",
	"ApiVersion",
	"BrokerBalancedConsumer",
	"ThrottleTime",
	"Sasl",
	"SaslHandshake",
	"BrokerGroupCoordinator",
	"LZ4",
	"OffsetTime",
	"MsgVer2",
	NULL
	};


	static const struct rd_kafka_feature_map {
	/* RD_KAFKA_FEATURE_... */
	int feature;

	/* Depends on the following ApiVersions overlapping with
	* what the broker supports: */
	struct rd_kafka_ApiVersion depends[RD_KAFKAP__NUM];

	} rd_kafka_feature_map[] = {
	/**
	* @brief List of features and the ApiVersions they depend on.
	*
	* The dependency list consists of the ApiKey followed by this
	* client's supported minimum and maximum API versions.
	* As long as this list and its versions overlaps with the
	* broker supported API versions the feature will be enabled.
	*/
	{

	/* @brief >=0.10.0: Message.MagicByte version 1:
	* Relative offsets (KIP-31) and message timestamps (KIP-32). */
	.feature = RD_KAFKA_FEATURE_MSGVER1,
	.depends = {
	{ RD_KAFKAP_Produce, 2, 2 },
	{ RD_KAFKAP_Fetch, 2, 2 },
	{ -1 },
	},
	},
	{
	/* @brief >=0.11.0: Message.MagicByte version 2 */
	.feature = RD_KAFKA_FEATURE_MSGVER2,
	.depends = {
	{ RD_KAFKAP_Produce, 3, 3 },
	{ RD_KAFKAP_Fetch, 4, 4 },
	{ -1 },
	},
	},
	{

	/* @brief >=0.10.0: ApiVersionQuery support.
	* @remark This is a bit of chicken-and-egg problem but needs to be
	* set by feature_check() to avoid the feature being cleared
	* even when broker supports it. */
	.feature = RD_KAFKA_FEATURE_APIVERSION,
	.depends = {
	{ RD_KAFKAP_ApiVersion, 0, 0 },
	{ -1 },
	},
	},
	{
	/* @brief >=0.8.2.0: Broker-based Group coordinator */
	.feature = RD_KAFKA_FEATURE_BROKER_GROUP_COORD,
	.depends = {
	{ RD_KAFKAP_GroupCoordinator, 0, 0 },
	{ -1 },
	},
	},
	{
	/* @brief >=0.9.0: Broker-based balanced consumer groups. */
	.feature = RD_KAFKA_FEATURE_BROKER_BALANCED_CONSUMER,
	.depends = {
	{ RD_KAFKAP_GroupCoordinator, 0, 0 },
	{ RD_KAFKAP_OffsetCommit, 1, 2 },
	{ RD_KAFKAP_OffsetFetch, 1, 1 },
	{ RD_KAFKAP_JoinGroup, 0, 0 },
	{ RD_KAFKAP_SyncGroup, 0, 0 },
	{ RD_KAFKAP_Heartbeat, 0, 0 },
	{ RD_KAFKAP_LeaveGroup, 0, 0 },
	{ -1 },
	},
	},
	{
	/* @brief >=0.9.0: ThrottleTime */
	.feature = RD_KAFKA_FEATURE_THROTTLETIME,
	.depends = {
	{ RD_KAFKAP_Produce, 1, 2 },
	{ RD_KAFKAP_Fetch, 1, 2 },
	{ -1 },
	},

	},
	{
	/* @brief >=0.9.0: SASL (GSSAPI) authentication.
	* Since SASL is not using the Kafka protocol
	* we must use something else to map us to the
	* proper broker version support:
	* JoinGroup was released along with SASL in 0.9.0. */
	.feature = RD_KAFKA_FEATURE_SASL_GSSAPI,
	.depends = {
	{ RD_KAFKAP_JoinGroup, 0, 0 },
	{ -1 },
	},
	},
	{
	/* @brief >=0.10.0: SASL mechanism handshake (KIP-43)
	* to automatically support other mechanisms
	* than GSSAPI, such as PLAIN. */
	.feature = RD_KAFKA_FEATURE_SASL_HANDSHAKE,
	.depends = {
	{ RD_KAFKAP_SaslHandshake, 0, 0 },
	{ -1 },
	},
	},
	{
	/* @brief >=0.8.2: LZ4 compression.
	* Since LZ4 initially did not rely on a specific API
	* type or version (it does in >=0.10.0)
	* we must use something else to map us to the
	* proper broker version support:
	* GrooupCoordinator was released in 0.8.2 */
	.feature = RD_KAFKA_FEATURE_LZ4,
	.depends = {
	{ RD_KAFKAP_GroupCoordinator, 0, 0 },
	{ -1 },
	},
	},
	{
	/* @brief >=0.10.1.0: Offset v1 (KIP-79)
	* Time-based offset requests */
	.feature = RD_KAFKA_FEATURE_OFFSET_TIME,
	.depends = {
	{ RD_KAFKAP_Offset, 1, 1 },
	{ -1 },
	}
	},
	{ .feature = 0 }, /* sentinel */
	};



	/**
	* @brief In absence of KIP-35 support in earlier broker versions we provide hardcoded
	* lists that corresponds to older broker versions.
	*/

	/* >= 0.10.0.0: dummy for all future versions that support ApiVersionRequest */
	static struct rd_kafka_ApiVersion rd_kafka_ApiVersion_Queryable[] = {
	{ RD_KAFKAP_ApiVersion, 0, 0 }
	};


	/* =~ 0.9.0 */
	static struct rd_kafka_ApiVersion rd_kafka_ApiVersion_0_9_0[] = {
	{ RD_KAFKAP_Produce, 0, 1 },
	{ RD_KAFKAP_Fetch, 0, 1 },
	{ RD_KAFKAP_Offset, 0, 0 },
	{ RD_KAFKAP_Metadata, 0, 0 },
	{ RD_KAFKAP_OffsetCommit, 0, 2 },
	{ RD_KAFKAP_OffsetFetch, 0, 1 },
	{ RD_KAFKAP_GroupCoordinator, 0, 0 },
	{ RD_KAFKAP_JoinGroup, 0, 0 },
	{ RD_KAFKAP_Heartbeat, 0, 0 },
	{ RD_KAFKAP_LeaveGroup, 0, 0 },
	{ RD_KAFKAP_SyncGroup, 0, 0 },
	{ RD_KAFKAP_DescribeGroups, 0, 0 },
	{ RD_KAFKAP_ListGroups, 0, 0 }
	};

	/* =~ 0.8.2 */
	static struct rd_kafka_ApiVersion rd_kafka_ApiVersion_0_8_2[] = {
	{ RD_KAFKAP_Produce, 0, 0 },
	{ RD_KAFKAP_Fetch, 0, 0 },
	{ RD_KAFKAP_Offset, 0, 0 },
	{ RD_KAFKAP_Metadata, 0, 0 },
	{ RD_KAFKAP_OffsetCommit, 0, 1 },
	{ RD_KAFKAP_OffsetFetch, 0, 1 },
	{ RD_KAFKAP_GroupCoordinator, 0, 0 }
	};

	/* =~ 0.8.1 */
	static struct rd_kafka_ApiVersion rd_kafka_ApiVersion_0_8_1[] = {
	{ RD_KAFKAP_Produce, 0, 0 },
	{ RD_KAFKAP_Fetch, 0, 0 },
	{ RD_KAFKAP_Offset, 0, 0 },
	{ RD_KAFKAP_Metadata, 0, 0 },
	{ RD_KAFKAP_OffsetCommit, 0, 1 },
	{ RD_KAFKAP_OffsetFetch, 0, 0 }
	};

	/* =~ 0.8.0 */
	static struct rd_kafka_ApiVersion rd_kafka_ApiVersion_0_8_0[] = {
	{ RD_KAFKAP_Produce, 0, 0 },
	{ RD_KAFKAP_Fetch, 0, 0 },
	{ RD_KAFKAP_Offset, 0, 0 },
	{ RD_KAFKAP_Metadata, 0, 0 }
	};


	/**
	* @brief Returns the ApiVersion list for legacy broker versions that do not
	* support the ApiVersionQuery request. E.g., brokers <0.10.0.
	*
	* @param broker_version Broker version to match (longest prefix matching).
	* @param use_default If no match is found return the default APIs (but return 0).
	*
	* @returns 1 if \p broker_version was recognized: \p *apisp will point to
	* the ApiVersion list and *api_cntp will be set to its element count.
	* 0 if \p broker_version was not recognized: \p *apisp remains unchanged.
	*
	*/
	int rd_kafka_get_legacy_ApiVersions (const char *broker_version,
	struct rd_kafka_ApiVersion **apisp,
	size_t api_cntp, const char fallback) {
	static const struct {
	const char *pfx;
	struct rd_kafka_ApiVersion *apis;
	size_t api_cnt;
	} vermap[] = {
	#define _VERMAP(PFX,APIS) { PFX, APIS, RD_ARRAYSIZE(APIS) }
	_VERMAP("0.9.0", rd_kafka_ApiVersion_0_9_0),
	_VERMAP("0.8.2", rd_kafka_ApiVersion_0_8_2),
	_VERMAP("0.8.1", rd_kafka_ApiVersion_0_8_1),
	_VERMAP("0.8.0", rd_kafka_ApiVersion_0_8_0),
	{ "0.7.", NULL }, /* Unsupported */
	{ "0.6.", NULL }, /* Unsupported */
	_VERMAP("", rd_kafka_ApiVersion_Queryable),
	{ NULL }
	};
	int i;
	int fallback_i = -1;
	int ret = 0;

	*apisp = NULL;
	*api_cntp = 0;

	for (i = 0 ; vermap[i].pfx ; i++) {
	if (!strncmp(vermap[i].pfx, broker_version, strlen(vermap[i].pfx))) {
	if (!vermap[i].apis)
	return 0;
	*apisp = vermap[i].apis;
	*api_cntp = vermap[i].api_cnt;
	ret = 1;
	break;
	} else if (fallback && !strcmp(vermap[i].pfx, fallback))
	fallback_i = i;
	}

	if (!*apisp && fallback) {
	rd_kafka_assert(NULL, fallback_i != -1);
	*apisp = vermap[fallback_i].apis;
	*api_cntp = vermap[fallback_i].api_cnt;
	}

	return ret;
	}


	/**
	* @returns 1 if the provided broker version (probably)
	* supports api.version.request.
	*/
	int rd_kafka_ApiVersion_is_queryable (const char *broker_version) {
	struct rd_kafka_ApiVersion *apis;
	size_t api_cnt;


	if (!rd_kafka_get_legacy_ApiVersions(broker_version,
	&apis, &api_cnt, 0))
	return 0;

	return apis == rd_kafka_ApiVersion_Queryable;
	}





	/**
	* @brief Check if match's versions overlaps with \p apis.
	*
	* @returns 1 if true, else 0.
	* @remark \p apis must be sorted using rd_kafka_ApiVersion_key_cmp()
	*/
	static RD_INLINE int
	rd_kafka_ApiVersion_check (const struct rd_kafka_ApiVersion *apis, size_t api_cnt,
	const struct rd_kafka_ApiVersion *match) {
	const struct rd_kafka_ApiVersion *api;

	api = bsearch(match, apis, api_cnt, sizeof(*apis),
	rd_kafka_ApiVersion_key_cmp);
	if (unlikely(!api))
	return 0;

	return match->MinVer <= api->MaxVer && api->MinVer <= match->MaxVer;
	}


	/**
	* @brief Compare broker's supported API versions to our feature request map
	* and enable/disable features accordingly.
	*
	* @param broker_apis Broker's supported APIs. If NULL the
	* \p broker.version.fallback configuration property will specify a
	* default legacy version to use.
	* @param broker_api_cnt Number of elements in \p broker_apis
	*
	* @returns the supported features (bitmask) to enable.
	*/
	int rd_kafka_features_check (rd_kafka_broker_t *rkb,
	struct rd_kafka_ApiVersion *broker_apis,
	size_t broker_api_cnt) {
	int features = 0;
	int i;

	/* Scan through features. */
	for (i = 0 ; rd_kafka_feature_map[i].feature != 0 ; i++) {
	const struct rd_kafka_ApiVersion *match;
	int fails = 0;

	/* For each feature check that all its API dependencies
	* can be fullfilled. */

	for (match = &rd_kafka_feature_map[i].depends[0] ;
	match->ApiKey != -1 ; match++) {
	int r;

	r = rd_kafka_ApiVersion_check(broker_apis, broker_api_cnt,
	match);

	rd_rkb_dbg(rkb, FEATURE, "APIVERSION",
	" Feature %s: %s (%hd..%hd) "
	"%ssupported by broker",
	rd_kafka_features2str(rd_kafka_feature_map[i].
	feature),
	rd_kafka_ApiKey2str(match->ApiKey),
	match->MinVer, match->MaxVer,
	r ? "" : "NOT ");

	fails += !r;
	}

	rd_rkb_dbg(rkb, FEATURE, "APIVERSION",
	"%s feature %s",
	fails ? "Disabling" : "Enabling",
	rd_kafka_features2str(rd_kafka_feature_map[i].feature));


	if (!fails)
	features \|= rd_kafka_feature_map[i].feature;
	}

	return features;
	}



	/**
	* @brief Make an allocated and sorted copy of \p src.
	*/
	void
	rd_kafka_ApiVersions_copy (const struct rd_kafka_ApiVersion *src,
	size_t src_cnt,
	struct rd_kafka_ApiVersion **dstp,
	size_t *dst_cntp) {
	dstp = rd_memdup(src, sizeof(src) * src_cnt);
	*dst_cntp = src_cnt;
	qsort(dstp, dst_cntp, sizeof(**dstp), rd_kafka_ApiVersion_key_cmp);
	}






	/**
	* @returns a human-readable feature flag string.
	*/
	const char *rd_kafka_features2str (int features) {
	static RD_TLS char ret[4][128];
	size_t of = 0;
	static RD_TLS int reti = 0;
	int i;

	reti = (reti + 1) % 4;

	*ret[reti] = '\0';
	for (i = 0 ; rd_kafka_feature_names[i] ; i++) {
	int r;
	if (!(features & (1 << i)))
	continue;

	r = rd_snprintf(ret[reti]+of, sizeof(ret[reti])-of, "%s%s",
	of == 0 ? "" : ",",
	rd_kafka_feature_names[i]);
	if ((size_t)r > sizeof(ret[reti])-of) {
	/* Out of space */
	memcpy(&ret[reti][sizeof(ret[reti])-3], "..", 3);
	break;
	}

	of += r;
	}

	return ret[reti];
	}