base.h File Reference

Common functions for interacting with kafk. More...

#include <librdkafka/rdkafka.h>
#include <freeradius-devel/server/cf_parse.h>
#include <freeradius-devel/util/rb.h>

Include dependency graph for base.h:

Go to the source code of this file.

Data Structures
struct	fr_kafka_conf_ctx_t
	uctx attached to each entry in KAFKA_BASE_PRODUCER_CONFIG More...
struct	fr_kafka_conf_t
struct	fr_kafka_topic_conf_t
struct	fr_kafka_topic_s
	Declared topic record - one per topic { <name> { ... } } subsection. More...

Macros
#define	KAFKA_BASE_CONFIG
	Config entries common to producer and consumer clients.
#define	KAFKA_CONSUMER_CONFIG
	Consumer-only delta: consumer-group membership, fetch/queue tuning, declared subscription topics.
#define	KAFKA_PRODUCER_CONFIG
	Producer-only delta: librdkafka producer tuning + declared topics.

Typedefs
typedef struct fr_kafka_topic_s	fr_kafka_topic_t

Functions
void	fr_kafka_free (void)
	Release one reference to librdkafka's global state.
int	fr_kafka_init (void)
	Initialise librdkafka's global state (SSL / SASL / internal ref-count)
int	kafka_config_dflt (CONF_PAIR **out, void *parent, CONF_SECTION *cs, fr_token_t quote, conf_parser_t const *rule)
	Default-generator counterpart to kafka_config_parse - reads the librdkafka default for the property and materialises it as a CONF_PAIR.
int	kafka_config_parse (TALLOC_CTX *ctx, void *out, void *base, CONF_ITEM *ci, conf_parser_t const *rule)
	Generic librdkafka-property parser used by KAFKA_BASE_PRODUCER_CONFIG entries.
int	kafka_config_raw_parse (TALLOC_CTX *ctx, void *out, void *base, CONF_ITEM *ci, conf_parser_t const *rule)
	Untyped passthrough parser used by KAFKA_RAW_CONFIG
fr_kafka_topic_t *	kafka_topic_conf_find (fr_kafka_conf_t const *kc, char const *name)
	Look up a declared topic by name on an fr_kafka_conf_t
int	kafka_topic_config_raw_parse (TALLOC_CTX *ctx, void *out, void *base, CONF_ITEM *ci, conf_parser_t const *rule)
	Topic-level raw passthrough.
int	kafka_topic_subsection_parse (TALLOC_CTX *ctx, void *out, void *base, CONF_ITEM *ci, conf_parser_t const *rule)
	Per-topic subsection hook used by KAFKA_PRODUCER_CONFIG / KAFKA_CONSUMER_CONFIG.

Nested config arrays referenced by <tt>KAFKA_BASE_PRODUCER_CONFIG</tt>
Extern so the macro's FR_CONF_SUBSECTION_GLOBAL entries can name them from any TU. Not part of the stable API - treat as implementation detail of the macro.
conf_parser_t const	kafka_base_consumer_topics_config []
conf_parser_t const	kafka_base_producer_topics_config []
conf_parser_t const	kafka_base_properties_config []
	properties { ... } escape-hatch contents
conf_parser_t const	kafka_base_topic_properties_config []
	Per-topic properties { ... } escape-hatch contents.
conf_parser_t const	kafka_connection_config []
conf_parser_t const	kafka_consumer_group_config []
conf_parser_t const	kafka_metadata_config []
conf_parser_t const	kafka_sasl_config []
conf_parser_t const	kafka_tls_config []
conf_parser_t const	kafka_version_config []

Detailed Description

Common functions for interacting with kafk.

Id

6f774a3ab11adcdd2fca1922bfe86f41ee518a2c

Author

Arran Cudbard-Bell

Copyright

2022 Arran Cudbard-Bell (a.cud.nosp@m.bard.nosp@m.b@fre.nosp@m.erad.nosp@m.ius.o.nosp@m.rg)

Definition in file base.h.

---

Data Structure Documentation

fr_kafka_conf_ctx_t

struct fr_kafka_conf_ctx_t

uctx attached to each entry in KAFKA_BASE_PRODUCER_CONFIG

Public so the config macro's struct literals resolve in caller TUs. You only touch this directly if you're extending the library's producer config with additional librdkafka pass-through properties.

Definition at line 80 of file base.h.

Collaboration diagram for fr_kafka_conf_ctx_t:

Data Fields
bool	empty_default	Don't produce messages saying the default is missing.
fr_table_ptr_sorted_t *	mapping	Mapping table between string constant.
size_t *	mapping_len	Length of the mapping tables.
char const *	property	Kafka configuration property.
size_t	size_scale	Divide/multiply FR_TYPE_SIZE by this amount.
char const *	string_sep	Used for multi-value configuration items. Kafka uses ', ' or ';' seemingly at random.

fr_kafka_conf_t

struct fr_kafka_conf_t

Definition at line 48 of file base.h.

Collaboration diagram for fr_kafka_conf_t:

Data Fields
rd_kafka_conf_t *	conf
fr_rb_tree_t *	topics	Declared topics, keyed by name. Populated during config parsing by the per-topic hook on the topic { } subsection; use kafka_topic_conf_find to query.

fr_kafka_topic_conf_t

struct fr_kafka_topic_conf_t

Definition at line 56 of file base.h.

Data Fields
rd_kafka_topic_conf_t *	rdtc

fr_kafka_topic_s

struct fr_kafka_topic_s

Declared topic record - one per topic { <name> { ... } } subsection.

Built by the library's per-topic parse hook, indexed on the parent fr_kafka_conf_t.topics tree. Callers typically retrieve via kafka_topic_conf_find rather than poking this struct directly.

Definition at line 66 of file base.h.

Collaboration diagram for fr_kafka_topic_s:

Data Fields
fr_kafka_topic_conf_t *	conf	parsed per-topic librdkafka conf
CONF_SECTION *	cs	topic's CONF_SECTION (for call_env lookups of per-topic pairs like value / key)
char const *	name	as it appeared in config
fr_rb_node_t	node

Macro Definition Documentation

KAFKA_BASE_CONFIG

#define KAFKA_BASE_CONFIG

Config entries common to producer and consumer clients.

Broker list, client identity, TLS / SASL, metadata / version / connection tuning, debug / plugin knobs. Usually composed with a role-specific macro (KAFKA_PRODUCER_CONFIG, or a future consumer equivalent).

Definition at line 177 of file base.h.

KAFKA_CONSUMER_CONFIG

#define KAFKA_CONSUMER_CONFIG

Consumer-only delta: consumer-group membership, fetch/queue tuning, declared subscription topics.

Compose with KAFKA_BASE_CONFIG. Same embedding contract as the producer macro (see KAFKA_PRODUCER_CONFIG).

static conf_parser_t const module_config[] = {
    KAFKA_BASE_CONFIG,
    KAFKA_CONSUMER_CONFIG,
    CONF_PARSER_TERMINATOR
};

Definition at line 285 of file base.h.

KAFKA_PRODUCER_CONFIG

#define KAFKA_PRODUCER_CONFIG

Producer-only delta: librdkafka producer tuning + declared topics.

Compose with KAFKA_BASE_CONFIG. Callers must embed fr_kafka_conf_t as the first member of their instance struct so FR_CONF_OFFSET resolves against it.

static conf_parser_t const module_config[] = {
    KAFKA_BASE_CONFIG,
    KAFKA_PRODUCER_CONFIG,
    { FR_CONF_OFFSET("my_thing", rlm_foo_t, my_thing) },
    CONF_PARSER_TERMINATOR
};

Definition at line 228 of file base.h.

Typedef Documentation

fr_kafka_topic_t

typedef struct fr_kafka_topic_s fr_kafka_topic_t

Definition at line 46 of file base.h.

Function Documentation

fr_kafka_free()

void fr_kafka_free

(

void

)

Release one reference to librdkafka's global state.

Call from a module's .unload to pair fr_kafka_init(). The last release is a no-op; librdkafka internally ref-counts its own globals and tears down when the last rd_kafka_t goes.

Release one reference to librdkafka's global state.

librdkafka refcounts its own globals internally; our counter just pairs fr_kafka_init() calls so re-entrant module load/unload in test harnesses does the right thing.

Definition at line 1238 of file base.c.

Here is the caller graph for this function:

fr_kafka_init()

int fr_kafka_init

(

void

)

Initialise librdkafka's global state (SSL / SASL / internal ref-count)

Ref-counted: every call must be paired with fr_kafka_free(). The first call lazily kicks librdkafka's one-time init paths by creating and destroying a throwaway producer; subsequent calls just bump the refcount. Call this from a kafka-using module's .onload so the library's internal globals are set up deterministically at startup rather than racing the first real rd_kafka_new() in a worker thread.

Returns

0 on success, -1 on failure.

Initialise librdkafka's global state (SSL / SASL / internal ref-count)

First call creates and immediately destroys a throwaway producer, which walks all of librdkafka's one-shot init paths (SSL lock callbacks on OpenSSL 1.0.2, SASL global init if compiled in, etc.). Subsequent calls just bump the refcount so multiple kafka-using modules can share the init.

Definition at line 1207 of file base.c.

Here is the call graph for this function:

Here is the caller graph for this function:

kafka_config_dflt()

int kafka_config_dflt	(	CONF_PAIR **	out,
		void *	parent,
		CONF_SECTION *	cs,
		fr_token_t	quote,
		conf_parser_t const *	rule
	)

Default-generator counterpart to kafka_config_parse - reads the librdkafka default for the property and materialises it as a CONF_PAIR.

Default-generator counterpart to kafka_config_parse - reads the librdkafka default for the property and materialises it as a CONF_PAIR.

Parameters

[out]	out	Where to write the pair.
[in]	parent	being populated.
[in]	cs	to allocate the pair in.
[in]	quote	to use when allocing the pair.
[in]	rule	UNUSED.

Returns

• 0 on success.
• -1 on failure.

Definition at line 367 of file base.c.

Here is the call graph for this function:

kafka_config_parse()

int kafka_config_parse	(	TALLOC_CTX *	ctx,
		void *	out,
		void *	base,
		CONF_ITEM *	ci,
		conf_parser_t const *	rule
	)

Generic librdkafka-property parser used by KAFKA_BASE_PRODUCER_CONFIG entries.

Exposed so the macro's FR_CONF_PAIR_GLOBAL entries can reference it from any TU.

kafka_config_raw_parse()

int kafka_config_raw_parse	(	TALLOC_CTX *	ctx,
		void *	out,
		void *	base,
		CONF_ITEM *	ci,
		conf_parser_t const *	rule
	)

Untyped passthrough parser used by KAFKA_RAW_CONFIG

Hands the CONF_PAIR's attribute/value straight to rd_kafka_conf_set. No type dispatch - user is responsible for librdkafka-native units.

kafka_topic_conf_find()

fr_kafka_topic_t * kafka_topic_conf_find	(	fr_kafka_conf_t const *	kc,
		char const *	name
	)

Look up a declared topic by name on an fr_kafka_conf_t

Returns

the fr_kafka_topic_t, or NULL if no topic of that name was declared (or if no topics have been parsed yet).

Definition at line 623 of file base.c.

Here is the call graph for this function:

Here is the caller graph for this function:

kafka_topic_config_raw_parse()

int kafka_topic_config_raw_parse	(	TALLOC_CTX *	ctx,
		void *	out,
		void *	base,
		CONF_ITEM *	ci,
		conf_parser_t const *	rule
	)

Topic-level raw passthrough.

Counterpart to kafka_config_raw_parse for use inside a declared topic's subsection, dispatching to rd_kafka_topic_conf_set.

kafka_topic_subsection_parse()

int kafka_topic_subsection_parse	(	TALLOC_CTX *	ctx,
		void *	out,
		void *	base,
		CONF_ITEM *	ci,
		conf_parser_t const *	rule
	)

Per-topic subsection hook used by KAFKA_PRODUCER_CONFIG / KAFKA_CONSUMER_CONFIG.

Runs the inner rules against each topic { <name> { ... } } block then inserts an fr_kafka_topic_t into the parent fr_kafka_conf_t.topics tree so callers can look topics up by name without re-walking CONF_SECTIONs.

Variable Documentation

kafka_base_consumer_topics_config

conf_parser_t const kafka_base_consumer_topics_config[]

extern

Definition at line 1168 of file base.c.

kafka_base_producer_topics_config

conf_parser_t const kafka_base_producer_topics_config[]

extern

Definition at line 1074 of file base.c.

kafka_base_properties_config

conf_parser_t const kafka_base_properties_config[]

extern

properties { ... } escape-hatch contents

Accepts any key = value pair and hands it straight to rd_kafka_conf_set. See kafka_config_raw_parse.

Definition at line 706 of file base.c.

kafka_base_topic_properties_config

conf_parser_t const kafka_base_topic_properties_config[]

extern

Per-topic properties { ... } escape-hatch contents.

Same idea as kafka_base_properties_config, but dispatches to rd_kafka_topic_conf_set against the enclosing topic's conf.

Definition at line 716 of file base.c.

kafka_connection_config

conf_parser_t const kafka_connection_config[]

extern

Definition at line 858 of file base.c.

kafka_consumer_group_config

conf_parser_t const kafka_consumer_group_config[]

extern

Definition at line 1092 of file base.c.

kafka_metadata_config

conf_parser_t const kafka_metadata_config[]

extern

Definition at line 963 of file base.c.

kafka_sasl_config

conf_parser_t const kafka_sasl_config[]

extern

Definition at line 765 of file base.c.

kafka_tls_config

conf_parser_t const kafka_tls_config[]

extern

Definition at line 799 of file base.c.

kafka_version_config

conf_parser_t const kafka_version_config[]

extern

Definition at line 935 of file base.c.