/*
 *   This program is is free software; you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation; either version 2 of the License, or (at
 *   your option) any later version.
 *
 *   This program is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with this program; if not, write to the Free Software
 *   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
 */

/**
 * $Id: 1c5b31927d4e569cd934a3f2d567d2a70091105c $
 * @file kafka/base.c
 * @brief Kafka global structures
 *
 * @copyright 2022 Arran Cudbard-Bell (a.cudbardb@freeradius.org)
 */

#include <freeradius-devel/kafka/base.h>
#include <freeradius-devel/server/cf_parse.h>
#include <freeradius-devel/server/tmpl.h>
#include <freeradius-devel/util/value.h>
#include <freeradius-devel/util/sbuff.h>
#include <freeradius-devel/util/size.h>

typedef struct {
	rd_kafka_conf_t		*conf;
} fr_kafka_conf_t;

typedef struct {
	rd_kafka_topic_conf_t	*conf;
} fr_kafka_topic_conf_t;

typedef struct {
	fr_table_ptr_sorted_t	*mapping;		//!< Mapping table between string constant.

	size_t			*mapping_len;		//!< Length of the mapping tables

	bool			empty_default;		//!< Don't produce messages saying the default is missing.

	size_t			size_scale;		//!< Divide/multiply FR_TYPE_SIZE by this amount.

	char const		*property;		//!< Kafka configuration property.

	char const		*string_sep;		//!< Used for multi-value configuration items.
							//!< Kafka uses ', ' or ';' seemingly at random.
} fr_kafka_conf_ctx_t;

/** Destroy a kafka configuration handle
 *
 * @param[in] kc	To destroy.
 * @return 0
 */
static int _kafka_conf_free(fr_kafka_conf_t *kc)
{
	rd_kafka_conf_destroy(kc->conf);
	return 0;
}

static inline CC_HINT(always_inline)
fr_kafka_conf_t *kafka_conf_from_cs(CONF_SECTION *cs)
{
	CONF_DATA const	*cd;
	fr_kafka_conf_t	*kc;

	cd = cf_data_find(cs, fr_kafka_conf_t, "conf");
	if (cd) {
		kc = cf_data_value(cd);
	} else {
		MEM(kc = talloc(NULL, fr_kafka_conf_t));
		MEM(kc->conf = rd_kafka_conf_new());
		talloc_set_destructor(kc, _kafka_conf_free);
		cf_data_add(cs, kc, "conf", true);
	}

	return kc;
}

/** Destroy a kafka topic configuration handle
 *
 * @param[in] ktc	To destroy.
 * @return 0
 */
static int _kafka_topic_conf_free(fr_kafka_topic_conf_t *ktc)
{
	rd_kafka_topic_conf_destroy(ktc->conf);
	return 0;
}

static inline CC_HINT(always_inline)
fr_kafka_topic_conf_t *kafka_topic_conf_from_cs(CONF_SECTION *cs)
{
	CONF_DATA const		*cd;
	fr_kafka_topic_conf_t	*ktc;

	cd = cf_data_find(cs, fr_kafka_topic_conf_t, "conf");
	if (cd) {
		ktc = cf_data_value(cd);
	} else {
		MEM(ktc = talloc(NULL, fr_kafka_topic_conf_t));
		MEM(ktc->conf = rd_kafka_topic_conf_new());
		talloc_set_destructor(ktc, _kafka_topic_conf_free);
		cf_data_add(cs, ktc, "conf", true);
	}

	return ktc;
}

/** Perform any conversions necessary to map kafka defaults to our values
 *
 * @param[out] out	Where to write the pair.
 * @param[in] parent	being populated.
 * @param[in] cs	to allocate the pair in.
 * @param[in] value	to convert.
 * @param[in] quote	to use when allocing the pair.
 * @param[in] rule	UNUSED.
 * @return
 *	- 0 on success.
 *      - -1 on failure.
 */
static int kafka_config_dflt_single(CONF_PAIR **out, UNUSED void *parent, CONF_SECTION *cs, char const *value,
				    fr_token_t quote, conf_parser_t const *rule)
{
	char				tmp[sizeof("18446744073709551615b")];
	fr_kafka_conf_ctx_t const	*kctx = rule->uctx;
	fr_type_t			type = rule->type;

	/*
	 *	Apply any mappings available, but default back
	 *      to the raw value if we don't have a match.
	 */
	if (kctx->mapping) {
		fr_table_ptr_sorted_t	*mapping = kctx->mapping;
		size_t			mapping_len = *kctx->mapping_len;

		value = fr_table_str_by_str_value(mapping, value, value);
	}
	/*
	 *	Convert time delta as an integer with ms precision
	 */
	switch (type) {
	case FR_TYPE_TIME_DELTA:
	{
		fr_sbuff_t 	value_elem = FR_SBUFF_IN(tmp, sizeof(tmp));
		fr_time_delta_t	delta;

		if (fr_time_delta_from_str(&delta, value, strlen(value), FR_TIME_RES_MSEC) < 0) {
			cf_log_perr(cs, "Failed parsing default \"%s\"", value);
			return -1;
		}

		fr_time_delta_to_str(&value_elem, delta, FR_TIME_RES_SEC, true);
		value = fr_sbuff_start(&value_elem);
	}
		break;

	case FR_TYPE_SIZE:
	{
		fr_sbuff_t 	value_elem = FR_SBUFF_IN(tmp, sizeof(tmp));
		size_t		size;

		if (fr_size_from_str(&size, &FR_SBUFF_IN(value, strlen(value))) < 0) {
			cf_log_perr(cs, "Failed parsing default \"%s\"", value);
			return -1;
		}

		/*
		 *	Some options are in kbytes *sigh*
		 */
		if (kctx->size_scale) size *= kctx->size_scale;

		/*
		 * 	reprint the size with an appropriate unit
		 */
		if (fr_size_to_str(&value_elem, size) < 0) {
			cf_log_perr(cs, "Failed size reprint");
			return -1;
		}
		value = fr_sbuff_start(&value_elem);
	}
		break;

	default:
		break;
	}

	MEM(*out = cf_pair_alloc(cs, rule->name1, value, T_OP_EQ, T_BARE_WORD, quote));
	cf_pair_mark_parsed(*out);	/* Don't re-parse this */

	return 0;
}

/** Return the default value from the kafka client library
 *
 * @param[out] out	Where to write the pair.
 * @param[in] parent	being populated.
 * @param[in] cs	to allocate the pair in.
 * @param[in] quote	to use when allocing the pair.
 * @param[in] rule	UNUSED.
 * @return
 *	- 0 on success.
 *      - -1 on failure.
 */
static int kafka_config_dflt(CONF_PAIR **out, void *parent, CONF_SECTION *cs, fr_token_t quote, conf_parser_t const *rule)
{
	char				buff[1024];
	size_t				buff_len = sizeof(buff);
	char const			*value;

	fr_kafka_conf_t 		*kc;
	fr_kafka_conf_ctx_t const	*kctx = rule->uctx;
	rd_kafka_conf_res_t 		ret;

	kc = kafka_conf_from_cs(cs);
	fr_assert(kc);

	if ((ret = rd_kafka_conf_get(kc->conf, kctx->property, buff, &buff_len)) != RD_KAFKA_CONF_OK) {
		if (ret == RD_KAFKA_CONF_UNKNOWN) {
			if (kctx->empty_default) return 0;

			cf_log_debug(cs, "No default available for \"%s\" - \"%s\"", rule->name1, kctx->property);
			return 0;	/* Not an error */
		}

		cf_log_err(cs, "Failed retrieving kafka property \"%s\"", kctx->property);
		return -1;
	}
#if 0
	cf_log_debug(cs, "Retrieved dflt \"%s\" for \"%s\" - \"%s\"", buff, rule->name1, kctx->property);
#endif
	value = buff;

	/*
	 *	If it's multi we need to break the string apart on the string separator
	 *	and potentially unescape the separator.
	 */
	if (fr_rule_multi(rule)) {
		fr_sbuff_t 			value_in = FR_SBUFF_IN(value, buff_len);
		char				tmp[256];
		fr_sbuff_t 			value_elem = FR_SBUFF_OUT(tmp, sizeof(tmp));
		fr_sbuff_term_t			tt = FR_SBUFF_TERM(kctx->string_sep);
		fr_sbuff_unescape_rules_t	ue_rules = {
							.name = __FUNCTION__,
							.chr = '\\'
						};
		/*
		 *	Convert escaped separators back
		 */
		ue_rules.subs[(uint8_t)kctx->string_sep[0]] = kctx->string_sep[0];

		while (fr_sbuff_out_unescape_until(&value_elem, &value_in, SIZE_MAX, &tt, &ue_rules) > 0) {
			if (kafka_config_dflt_single(out, parent, cs, fr_sbuff_start(&value_elem), quote, rule) < 0) return -1;

			/*
			 *	Skip past the string separator
			 */
			fr_sbuff_advance(&value_in, strlen(kctx->string_sep));

			/*
			 *	Reset
			 */
			fr_sbuff_set_to_start(&value_elem);
		}
		return 0;
	}

	/*
	 *	Parse a single value
	 */
	if (kafka_config_dflt_single(out, parent, cs, value, quote, rule) < 0) return -1;

	return 0;
}

/** Return the default value for a topic from the kafka client library
 *
 * @param[out] out	Where to write the pair.
 * @param[in] parent	being populated.
 * @param[in] cs	to allocate the pair in.
 * @param[in] quote	to use when allocing the pair.
 * @param[in] rule	UNUSED.
 * @return
 *	- 0 on success.
 *      - -1 on failure.
 */
static int kafka_topic_config_dflt(CONF_PAIR **out, void *parent, CONF_SECTION *cs, fr_token_t quote, conf_parser_t const *rule)
{
	char				buff[1024];
	size_t				buff_len = sizeof(buff);
	char const			*value;

	fr_kafka_topic_conf_t 		*ktc;
	fr_kafka_conf_ctx_t const	*kctx = rule->uctx;
	rd_kafka_conf_res_t 		ret;

	ktc = kafka_topic_conf_from_cs(cs);
	fr_assert(ktc);

	if ((ret = rd_kafka_topic_conf_get(ktc->conf, kctx->property, buff, &buff_len)) != RD_KAFKA_CONF_OK) {
		if (ret == RD_KAFKA_CONF_UNKNOWN) {
			if (kctx->empty_default) return 0;

			cf_log_debug(cs, "No default available for \"%s\" - \"%s\"", rule->name1, kctx->property);
			return 0;	/* Not an error */
		}

		fr_assert(ret == RD_KAFKA_CONF_UNKNOWN);
		cf_log_err(cs, "Failed retrieving kafka property '%s'", kctx->property);
		return -1;
	}
#if 0
	cf_log_debug(cs, "Retrieved dflt \"%s\" for \"%s\" - \"%s\"", buff, rule->name1, kctx->property);
#endif
	value = buff;

	/*
	 *	Parse a single value
	 */
	if (kafka_config_dflt_single(out, parent, cs, value, quote, rule) < 0) return -1;

	return 0;
}

static int kafka_config_parse_single(char const **out, CONF_PAIR *cp, conf_parser_t const *rule)
{
	fr_value_box_t			vb = FR_VALUE_BOX_INITIALISER_NULL(vb);
	fr_kafka_conf_ctx_t const	*kctx = rule->uctx;
	fr_type_t			type = rule->type;
	static _Thread_local char	buff[sizeof("18446744073709551615")];
	static _Thread_local fr_sbuff_t	sbuff;

	/*
	 *	Map string values if possible, and if there's
	 *	no match then just pass the original through.
	 *
	 *      We count this as validation...
	 */
	if (kctx->mapping) {
		fr_table_ptr_sorted_t	*mapping = kctx->mapping;
		size_t			mapping_len = *kctx->mapping_len;

		*out = fr_table_value_by_str(mapping, cf_pair_value(cp), cf_pair_value(cp));
		return 0;
	} else if (fr_type_is_string(type)) {
		*out = cf_pair_value(cp);
		return 0;
	}

	/*
	 *	Parse as a box for basic validation
	 */
	if (cf_pair_to_value_box(NULL, &vb, cp, rule) < 0) return -1;

	/*
	 *	In kafka all the time deltas are in ms
	 *	resolution, so we need to take the parsed value,
	 *	scale it, and print it back to a string.
	 */
	switch (type) {
	case FR_TYPE_TIME_DELTA:
	{
		uint64_t			delta;

		sbuff = FR_SBUFF_IN(buff, sizeof(buff));
		delta = fr_time_delta_to_msec(vb.vb_time_delta);
		if (fr_sbuff_in_sprintf(&sbuff, "%" PRIu64, delta) < 0) {
		error:
			fr_value_box_clear(&vb);
			return -1;
		}
		*out = fr_sbuff_start(&sbuff);
	}
		break;

	case FR_TYPE_SIZE:
	{
		size_t size = vb.vb_size;

		sbuff = FR_SBUFF_IN(buff, sizeof(buff));

		/*
		 *	Most options are in bytes, but some are in kilobytes
		 */
		if (kctx->size_scale) size /= kctx->size_scale;

		/*
		 *	Kafka doesn't want units...
		 */
		if (fr_sbuff_in_sprintf(&sbuff, "%zu", size) < 0) goto error;
		*out = fr_sbuff_start(&sbuff);
	}
		break;

	/*
	 *	Ensure bool is always mapped to the string constants
	 *	"true" or "false".
	 */
	case FR_TYPE_BOOL:
		*out = vb.vb_bool ? "true" : "false";
		break;

	default:
		*out = cf_pair_value(cp);
		break;
	}

	fr_value_box_clear(&vb);

	return 0;
}

/** Translate config items directly to settings in a kafka config struct
 *
 * @param[in] ctx	to allocate fr_kafka_conf_t in.
 * @param[out] out	Unused.
 * @param[in] base	Unused.
 * @param[in] ci	To parse.
 * @param[in] rule	describing how to parse the item.
 * @return
 *	- 0 on success.
 *      - -1 on failure
 */
static int kafka_config_parse(TALLOC_CTX *ctx, UNUSED void *out, UNUSED void *base,
			      CONF_ITEM *ci, conf_parser_t const *rule)
{
	fr_kafka_conf_ctx_t const	*kctx = rule->uctx;
	CONF_ITEM			*parent = cf_parent(ci);
	CONF_SECTION			*cs = cf_item_to_section(parent);
	CONF_PAIR			*cp = cf_item_to_pair(ci);

	fr_kafka_conf_t 		*kc;
	char const			*value;

	kc = kafka_conf_from_cs(cf_item_to_section(parent));

	/*
	 *	Multi rules require us to concat the values together before handing them off
	 */
	if (fr_rule_multi(rule)) {
		unsigned int	i;
		CONF_PAIR	*cp_p;
		size_t		count;
		char const	**array;
		fr_sbuff_t	*agg;
		fr_slen_t	slen;

		FR_SBUFF_TALLOC_THREAD_LOCAL(&agg, 256, SIZE_MAX);

		count = cf_pair_count(cs,  rule->name1);
		if (count <= 1) goto do_single;

		MEM(array = talloc_array(ctx, char const *, count));
		for (cp_p = cp, i = 0;
		     cp_p;
		     cp_p = cf_pair_find_next(cs, cp_p, rule->name1), i++) {
			if (kafka_config_parse_single(&array[i], cp_p, rule) < 0) return -1;
			cf_pair_mark_parsed(cp_p);
		}

		slen = talloc_array_concat(agg, array, kctx->string_sep);
		talloc_free(array);
		if (slen < 0) return -1;

		value = fr_sbuff_start(agg);
	} else {
	do_single:
		if (kafka_config_parse_single(&value, cp, rule) < 0) return -1;
	}

	{
		char errstr[512];

		if (rd_kafka_conf_set(kc->conf, kctx->property,
				      value, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
			cf_log_perr(cp, "%s", errstr);
			return -1;
		}
	}

	return 0;
}


/** Translate config items directly to settings in a kafka topic config struct
 *
 * @param[in] ctx	to allocate fr_kafka_conf_t in.
 * @param[out] out	Unused.
 * @param[in] base	Unused.
 * @param[in] ci	To parse.
 * @param[in] rule	describing how to parse the item.
 * @return
 *	- 0 on success.
 *      - -1 on failure
 */
static int kafka_topic_config_parse(UNUSED TALLOC_CTX *ctx, UNUSED void *out, UNUSED void *base,
				    CONF_ITEM *ci, conf_parser_t const *rule)
{
	fr_kafka_conf_ctx_t const	*kctx = rule->uctx;
	CONF_ITEM			*parent = cf_parent(ci);
	CONF_PAIR			*cp = cf_item_to_pair(ci);

	fr_kafka_topic_conf_t 		*ktc;
	char const			*value;

	ktc = kafka_topic_conf_from_cs(cf_item_to_section(parent));
	if (kafka_config_parse_single(&value, cp, rule) < 0) return -1;

	{
		char errstr[512];

		if (rd_kafka_topic_conf_set(ktc->conf, kctx->property,
					    value, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
			cf_log_perr(cp, "%s", errstr);
			return -1;
		}
	}

	return 0;
}

#if 0
/** Configure a new topic for production or consumption
 *
 */
static int kafka_topic_new(UNUSED TALLOC_CTX *ctx, UNUSED void *out, UNUSED void *base,
			   CONF_ITEM *ci, conf_parser_t const *rule)
{
	fr_kafka_conf_ctx_t const	*kctx = rule->uctx;
	CONF_ITEM			*parent = cf_parent(ci);
	CONF_PAIR			*cp = cf_item_to_pair(ci);

	fr_kafka_topic_conf_t 		*ktc;
	char const			*value;

	ktc = kafka_topic_conf_from_cs(cf_item_to_section(parent));

	rd_kafka_topic_new (rd_kafka_t *rk, const char *topic, rd_kafka_topic_conf_t *conf)
}
#endif

static conf_parser_t const kafka_sasl_oauth_config[] = {
	{ FR_CONF_FUNC("oauthbearer_conf", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.oauthbearer.config", .empty_default = true }},

	{ FR_CONF_FUNC("unsecure_jwt", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "enable.sasl.oauthbearer.unsecure.jwt" }},

	CONF_PARSER_TERMINATOR
};

static conf_parser_t const kafka_sasl_kerberos_config[] = {
	/*
	 *	Service principal
	 */
	{ FR_CONF_FUNC("service_name", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.kerberos.service.name" }},

	/*
	 *	Principal
	 */
	{ FR_CONF_FUNC("principal", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.kerberos.principal" }},

	/*
	 *	knit cmd
	 */
	{ FR_CONF_FUNC("kinit_cmd", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.kerberos.kinit.cmd" }},

	/*
	 *	keytab
	 */
	{ FR_CONF_FUNC("keytab", FR_TYPE_STRING, CONF_FLAG_FILE_INPUT, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.kerberos.kinit.keytab", .empty_default = true }},

	/*
	 *	How long between key refreshes
	 */
	{ FR_CONF_FUNC("refresh_delay", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.kerberos.min.time.before.relogin" }},

	CONF_PARSER_TERMINATOR
};

static conf_parser_t const kafka_sasl_config[] = {
	/*
	 *	SASL mechanism
	 */
	{ FR_CONF_FUNC("mech", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.mechanism" }},

	/*
	 *	Static SASL username
	 */
	{ FR_CONF_FUNC("username", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.username", .empty_default = true }},

	/*
	 *	Static SASL password
	 */
	{ FR_CONF_FUNC("password", FR_TYPE_STRING, CONF_FLAG_SECRET, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.password", .empty_default = true }},

	{ FR_CONF_SUBSECTION_GLOBAL("kerberos", 0, kafka_sasl_kerberos_config) },

	{ FR_CONF_SUBSECTION_GLOBAL("oauth", 0, kafka_sasl_oauth_config) },

	CONF_PARSER_TERMINATOR
};

static fr_table_ptr_sorted_t kafka_check_cert_cn_table[] = {
	{ L("false"),	"none"	},
	{ L("no"),	"none"	},
	{ L("true"),	"https"	},
	{ L("yes"),	"https"	}
};
static size_t kafka_check_cert_cn_table_len = NUM_ELEMENTS(kafka_check_cert_cn_table);

static conf_parser_t const kafka_tls_config[] = {
	/*
	 *	Cipher suite list in OpenSSL's format
	 */
	{ FR_CONF_FUNC("cipher_list", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "ssl.cipher.suites", .empty_default = true }},

	/*
	 *	Curves list in OpenSSL's format
	 */
	{ FR_CONF_FUNC("curve_list", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "ssl.curves.list", .empty_default = true }},

	/*
	 *	Curves list in OpenSSL's format
	 */
	{ FR_CONF_FUNC("sigalg_list", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "ssl.sigalgs.list", .empty_default = true }},

	/*
	 *	Sets the full path to a CA certificate (used to validate
	 *	the certificate the server presents).
	 */
	{ FR_CONF_FUNC("ca_file", FR_TYPE_STRING, CONF_FLAG_FILE_INPUT, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "ssl.ca.location", .empty_default = true }},

	/*
	 *	Location of the CRL file.
	 */
	{ FR_CONF_FUNC("crl_file", FR_TYPE_STRING, CONF_FLAG_FILE_INPUT, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "ssl.crl.location", .empty_default = true }},

	/*
	 *	Sets the path to the public certificate file we present
	 *	to the servers.
	 */
	{ FR_CONF_FUNC("certificate_file", FR_TYPE_STRING, CONF_FLAG_FILE_INPUT, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "ssl.certificate.location", .empty_default = true }},

	/*
	 *	Sets the path to the private key for our public
	 *	certificate.
	 */
	{ FR_CONF_FUNC("private_key_file", FR_TYPE_STRING, CONF_FLAG_FILE_INPUT, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "ssl.key.location", .empty_default = true }},

	/*
	 *	Enable or disable certificate validation
	 */
	{ FR_CONF_FUNC("require_cert", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "enable.ssl.certificate.verification" }},

	{ FR_CONF_FUNC("check_cert_cn", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "ssl.endpoint.identification.algorithm",
	  				  .mapping = kafka_check_cert_cn_table,
	  				  .mapping_len = &kafka_check_cert_cn_table_len }},
	CONF_PARSER_TERMINATOR
};

static conf_parser_t const kafka_connection_config[] = {
	/*
	 *	Socket timeout
	 */
	{ FR_CONF_FUNC("timeout", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "socket.timeout.ms" }},

	/*
	 *	Close broker connections after this period.
	 */
	{ FR_CONF_FUNC("idle_timeout", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "connections.max.idle.ms" }},

	/*
	 *	Maximum requests in flight (per connection).
	 */
	{ FR_CONF_FUNC("max_requests_in_flight", FR_TYPE_UINT64, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "max.in.flight.requests.per.connection" }},

	/*
	 *	Socket send buffer.
	 */
	{ FR_CONF_FUNC("send_buff", FR_TYPE_UINT64, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "socket.send.buffer.bytes" }},

	/*
	 *	Socket recv buffer.
	 */
	{ FR_CONF_FUNC("recv_buff", FR_TYPE_UINT64, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "socket.receive.buffer.bytes" }},

	/*
	 *	If true, send TCP keepalives
	 */
	{ FR_CONF_FUNC("keepalive", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "socket.keepalive.enable" }},

	/*
	 *	If true, disable nagle algorithm
	 */
	{ FR_CONF_FUNC("nodelay", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "socket.nagle.disable" }},

	/*
	 *	How long the DNS resolver cache is valid for
	 */
	{ FR_CONF_FUNC("resolver_cache_ttl", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "broker.address.ttl" }},

	/*
	 *	Should we use A records, AAAA records or either
	 *	when resolving broker addresses
	 */
	{ FR_CONF_FUNC("resolver_addr_family", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "broker.address.family" }},

	/*
	 *	How many failures before we reconnect the connection
	 */
	{ FR_CONF_FUNC("reconnection_failure_count", FR_TYPE_UINT32, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "socket.max.fails" }},

	/*
	 *	Initial time to wait before reconnecting.
	 */
	{ FR_CONF_FUNC("reconnection_delay_initial", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "reconnect.backoff.ms" }},

	/*
	 *	Max time to wait before reconnecting.
	 */
	{ FR_CONF_FUNC("reconnection_delay_max", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "reconnect.backoff.max.ms" }},

	CONF_PARSER_TERMINATOR
};

static conf_parser_t const kafka_version_config[] = {
	/*
	 *	Request the API version from connected brokers
	 */
	{ FR_CONF_FUNC("request", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "api.version.request" }},

	/*
	 *	How long to wait for a version response.
	 */
	{ FR_CONF_FUNC("timeout", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "api.version.request.timeout.ms" }},

	/*
	 *	How long to wait before retrying a version request.
	 */
	{ FR_CONF_FUNC("retry_delay", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "api.version.fallback.ms" }},

	/*
	 *	Default version to use if the version request fails.
	 */
	{ FR_CONF_FUNC("default", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "broker.version.fallback" }},

	CONF_PARSER_TERMINATOR
};

static conf_parser_t const kafka_metadata_config[] = {
	/*
	 *	Interval between attempts to refresh metadata from brokers
	 */
	{ FR_CONF_FUNC("refresh_interval", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "topic.metadata.refresh.interval.ms" }},

	/*
	 *	Interval between attempts to refresh metadata from brokers
	 */
	{ FR_CONF_FUNC("max_age", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "metadata.max.age.ms" }},

	/*
	 *	 Used when a topic loses its leader
	 */
	{ FR_CONF_FUNC("fast_refresh_interval", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "topic.metadata.refresh.fast.interval.ms" }},

	/*
	 *	 Used when a topic loses its leader to prevent spurious metadata changes
	 */
	{ FR_CONF_FUNC("max_propagation", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "topic.metadata.propagation.max.ms" }},

	/*
	 *	Use sparse metadata requests which use less bandwidth maps
	 */
	{ FR_CONF_FUNC("refresh_sparse", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "topic.metadata.refresh.sparse" }},

	/*
	 *	List of topics to ignore
	 */
	{ FR_CONF_FUNC("blacklist", FR_TYPE_STRING, CONF_FLAG_MULTI, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "topic.blacklist", .string_sep = ",", .empty_default = true }},

	CONF_PARSER_TERMINATOR
};

#define BASE_CONFIG \
	{ FR_CONF_FUNC("server", FR_TYPE_STRING, CONF_FLAG_REQUIRED | CONF_FLAG_MULTI, kafka_config_parse, kafka_config_dflt), \
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "metadata.broker.list", .string_sep = "," }}, \
	{ FR_CONF_FUNC("client_id", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt), \
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "client.id" }}, \
	{ FR_CONF_FUNC("rack_id", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt), \
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "client.rack" }}, \
	{ FR_CONF_FUNC("request_max_size", FR_TYPE_SIZE, 0, kafka_config_parse, kafka_config_dflt), \
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "message.max.bytes" }}, \
	{ FR_CONF_FUNC("request_copy_max_size", FR_TYPE_SIZE, 0, kafka_config_parse, kafka_config_dflt), \
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "message.copy.max.bytes" }}, \
	{ FR_CONF_FUNC("response_max_size", FR_TYPE_SIZE, 0, kafka_config_parse, kafka_config_dflt), \
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "receive.message.max.bytes" }}, \
	{ FR_CONF_FUNC("feature", FR_TYPE_STRING, CONF_FLAG_MULTI, kafka_config_parse, kafka_config_dflt), \
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "builtin.features", .string_sep = "," }}, \
	{ FR_CONF_FUNC("debug", FR_TYPE_STRING, CONF_FLAG_MULTI, kafka_config_parse, kafka_config_dflt), \
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "debug", .string_sep = "," }}, \
	{ FR_CONF_FUNC("plugin", FR_TYPE_STRING, CONF_FLAG_MULTI, kafka_config_parse, NULL), \
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "plugin.library.paths", .string_sep = ";" }}, \
	{ FR_CONF_SUBSECTION_GLOBAL("metadata", 0, kafka_metadata_config) }, \
	{ FR_CONF_SUBSECTION_GLOBAL("version", 0, kafka_version_config) }, \
	{ FR_CONF_SUBSECTION_GLOBAL("connection", 0, kafka_connection_config) }, \
	{ FR_CONF_SUBSECTION_GLOBAL("tls", 0, kafka_tls_config) }, \
	{ FR_CONF_SUBSECTION_GLOBAL("sasl", 0, kafka_sasl_config) }

static conf_parser_t const kafka_consumer_group_config[] = {
	/*
	 *	Group consumer is a member of
	 */
	{ FR_CONF_FUNC("id", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "group.id" }},

	/*
	 *	A unique identifier of the consumer instance provided by the end user
	 */
	{ FR_CONF_FUNC("instance_id", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "group.instance.id" }},

	/*
	 *	Range or roundrobin
	 */
	{ FR_CONF_FUNC("partition_assignment_strategy", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "partition.assignment.strategy" }},

	/*
	 *	Client group session and failure detection timeout.
	 */
	{ FR_CONF_FUNC("session_timeout", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "session.timeout.ms" }},

	/*
	 *	Group session keepalive heartbeat interval.
	 */
	{ FR_CONF_FUNC("heartbeat_interval", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "heartbeat.interval.ms" }},

	/*
	 *	How often to query for the current client group coordinator
	 */
	{ FR_CONF_FUNC("coordinator_query_interval", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "coordinator.query.interval.ms" }},


	CONF_PARSER_TERMINATOR
};

static conf_parser_t const kafka_base_consumer_topic_config[] = {
	/*
	 *	How many messages we process at a time
	 *
	 *	High numbers may starve the worker thread
	 */
	{ FR_CONF_FUNC("max_messages_per_cycle", FR_TYPE_UINT32, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "consume.callback.max.messages" }},

	/*
	 *	Action to take when there is no initial offset
	 *	in offset store or the desired offset is out of range.
	 */
	{ FR_CONF_FUNC("auto_offset_reset", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "auto.offset.reset" }},

	CONF_PARSER_TERMINATOR
};

/*
 * Allows topic configurations in the format:
 *
 * topic {
 *   <name> {
 *     request_required_acks = ...
 *   }
 * }
 *
 */
static conf_parser_t const kafka_base_consumer_topics_config[] = {
	{ FR_CONF_SUBSECTION_GLOBAL(CF_IDENT_ANY, CONF_FLAG_MULTI, kafka_base_consumer_topic_config) },

	CONF_PARSER_TERMINATOR
};

conf_parser_t const kafka_base_consumer_config[] = {
	BASE_CONFIG,
	{ FR_CONF_SUBSECTION_GLOBAL("group", 0, kafka_consumer_group_config) },

	/*
	 *	Maximum allowed time between calls to consume messages.
	 */
	{ FR_CONF_FUNC("max_poll_interval", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "max.poll.interval.ms" }},

	/*
	 *	Toggle auto commit
	 */
	{ FR_CONF_FUNC("auto_commit", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "enable_auto.commit" }},

	/*
	 *	Auto commit interval
	 */
	{ FR_CONF_FUNC("auto_commit_interval", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "auto.commit.interval.ms" }},

	/*
	 *	Automatically store offset of last message provided to application.
	 */
	{ FR_CONF_FUNC("auto_offset_store", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "enable.auto.offset.store" }},

	/*
	 *	Minimum number of messages per topic+partition librdkafka tries to
	 *	maintain in the local consumer queue.
	 */
	{ FR_CONF_FUNC("queued_messages_min", FR_TYPE_UINT64, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "queued.min.messages" }},

	/*
	 *	Maximum size of queued pre-fetched messages in the local consumer queue.
	 */
	{ FR_CONF_FUNC("queued_messages_max_size", FR_TYPE_SIZE, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "queued.max.messages.kbytes", .size_scale = 1024 }},

	/*
	 *	 Maximum time the broker may wait to fill the Fetch response.
	 */
	{ FR_CONF_FUNC("fetch_wait_max", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "fetch.wait.max.ms" }},

	/*
	 *	Initial maximum number of bytes per topic+partition to request when
	 *      fetching messages from the broker.
	 */
	{ FR_CONF_FUNC("fetch_message_max_size", FR_TYPE_SIZE, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "fetch.message.max.bytes" }},

	/*
	 *	Initial maximum number of bytes per topic+partition to request when
	 *	fetching messages from the broker.
	 */
	{ FR_CONF_FUNC("fetch_partition_max_size", FR_TYPE_SIZE, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "max.partition.fetch.bytes" }},

	/*
	 *	Maximum amount of data the broker shall return for a Fetch request.
	 */
	{ FR_CONF_FUNC("fetch_max_size", FR_TYPE_SIZE, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "fetch.max.bytes" }},

	/*
	 *	 Minimum number of bytes the broker responds with.
	 */
	{ FR_CONF_FUNC("fetch_min_size", FR_TYPE_SIZE, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "fetch.min.bytes" }},

	/*
	 *	How long to postpone the next fetch request for a topic+partition
	 *	in case of a fetch error.
	 */
	{ FR_CONF_FUNC("fetch_error_backoff", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "fetch.error.backoff.ms" }},

	/*
	 *	Controls how to read messages written transactionally
	 */
	{ FR_CONF_FUNC("isolation_level", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "isolation.level" }},

	/*
	 *	Verify CRC32 of consumed messages, ensuring no on-the-wire or
	 *	on-disk corruption to the messages occurred.
	 */
	{ FR_CONF_FUNC("check_crcs", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "check.crcs" }},

	/*
	 *	Allow automatic topic creation on the broker when subscribing
	 *	to or assigning non-existent topics
	 */
	{ FR_CONF_FUNC("auto_create_topic", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "allow.auto.create.topics" }},

	{ FR_CONF_SUBSECTION_GLOBAL("topic", 0, kafka_base_consumer_topics_config) }, \

	CONF_PARSER_TERMINATOR
};

static conf_parser_t const kafka_base_producer_topic_config[] = {
	/*
	 *	This field indicates the number of acknowledgements the leader
	 *	broker must receive from ISR brokers before responding to the request.
	 */
	{ FR_CONF_FUNC("request_required_acks", FR_TYPE_INT16, 0, kafka_topic_config_parse, kafka_topic_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "request.required.acks" }},

	/*
	 *	medium	The ack timeout of the producer request in milliseconds
	 */
	{ FR_CONF_FUNC("request_timeout", FR_TYPE_TIME_DELTA, 0, kafka_topic_config_parse, kafka_topic_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "request.timeout.ms" }},

	/*
	 *	Local message timeout
	 */
	{ FR_CONF_FUNC("message_timeout", FR_TYPE_TIME_DELTA, 0, kafka_topic_config_parse, kafka_topic_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "message.timeout.ms" }},

	/*
	 *	Partitioning strategy
	 */
	{ FR_CONF_FUNC("partitioner", FR_TYPE_STRING, 0, kafka_topic_config_parse, kafka_topic_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "partitioner" }},

	/*
	 *	compression codec to use for compressing message sets.
	 */
	{ FR_CONF_FUNC("compression_type", FR_TYPE_STRING, 0, kafka_topic_config_parse, kafka_topic_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "compression.type" }},

	/*
	 *	compression level to use
	 */
	{ FR_CONF_FUNC("compression_level", FR_TYPE_INT8, 0, kafka_topic_config_parse, kafka_topic_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "compression.level" }},

	CONF_PARSER_TERMINATOR
};

/*
 * Allows topic configurations in the format:
 *
 * topic {
 *   <name> {
 *     request_required_acks = ...
 *   }
 * }
 *
 */
static conf_parser_t const kafka_base_producer_topics_config[] = {
	{ FR_CONF_SUBSECTION_GLOBAL(CF_IDENT_ANY, 0, kafka_base_producer_topic_config) },

	CONF_PARSER_TERMINATOR
};

conf_parser_t const kafka_base_producer_config[] = {
	BASE_CONFIG,

	/*
	 *	Enables the transactional producer
	 */
	{ FR_CONF_FUNC("transactional_id", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "transactional.id", .empty_default = true }},

	/*
	 *	The maximum amount of time in milliseconds that the transaction
	 *	coordinator will wait for a transaction status update from the
	 *	producer before proactively aborting the ongoing transaction.
	 */
	{ FR_CONF_FUNC("transaction_timeout", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "transaction.timeout.ms" }},

	/*
	 *	When set to true, the producer will ensure that messages are
	 *	successfully produced exactly once and in the original produce
	 *	order.
	 */
	{ FR_CONF_FUNC("idempotence", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "enable.idempotence" }},

	/*
	 *	When set to true, any error that could result in a gap in the
	 *	produced message series when a batch of messages fails.
	 */
	{ FR_CONF_FUNC("gapless_guarantee", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "enable.gapless.guarantee" }},

	/*
	 *	Maximum number of messages allowed on the producer queue.
	 *	This queue is shared by all topics and partitions.
	 */
	{ FR_CONF_FUNC("queue_max_messages", FR_TYPE_UINT32, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "queue.buffering.max.messages" }},

	/*
	 *	Maximum total message size sum allowed on the producer queue.
	 */
	{ FR_CONF_FUNC("queue_max_size", FR_TYPE_SIZE, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "queue.buffering.max.kbytes", .size_scale = 1024 }},

	/*
	 *	How long we wait to aggregate messages
	 */
	{ FR_CONF_FUNC("queue_max_delay", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "queue.buffering.max.ms" }},

	/*
	 *	How many times we resend a message
	 */
	{ FR_CONF_FUNC("message_retry_max", FR_TYPE_UINT32, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "message.send.max.retries" }},

	/*
	 *	The backoff time in milliseconds before retrying a protocol request.
	 */
	{ FR_CONF_FUNC("message_retry_interval", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "retry.backoff.ms" }},

	/*
	 *	The threshold of outstanding not yet transmitted broker requests
	 *      needed to backpressure the producer's message accumulator.
	 */
	{ FR_CONF_FUNC("backpressure_threshold", FR_TYPE_UINT32, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "queue.buffering.backpressure.threshold" }},

	/*
	 *	compression codec to use for compressing message sets.
	 */
	{ FR_CONF_FUNC("compression_type", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "compression.type" }},

	/*
	 *	Maximum size (in bytes) of all messages batched in one MessageSet
	 */
	{ FR_CONF_FUNC("batch_size", FR_TYPE_SIZE, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "batch.size" }},

	/*
	 *	Delay in milliseconds to wait to assign new sticky partitions for each topic
	 */
	{ FR_CONF_FUNC("sticky_partition_delay", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
	  .uctx = &(fr_kafka_conf_ctx_t){ .property = "sticky.partitioning.linger.ms" }},

	{ FR_CONF_SUBSECTION_GLOBAL("topic", CONF_FLAG_MULTI, kafka_base_producer_topics_config) }, \

	CONF_PARSER_TERMINATOR
};
