Kafka Configuration

Reference · Applies to Brighter V10

Kafka General

Kafka is OSS message-oriented-middleware and is well documented. Brighter handles the details of sending to or receiving from Kafka. You may find it useful to understand the building blocks of the protocol. Brighter's Kafka support is implemented on top of the Confluent .NET client, and you might find the documentation for the .NET client helpful when debugging, but you should not have to interact with it directly to use Brighter (although we expose many of its configuration options).

Kafka has two main roles:

• Producer: A producer sends events to a Topic on a Kafka broker.

• Consumer: A consumer reads events from a Topic on a Kafka broker.

Topics are append-only streams of events. Multiple producers can write to a topic, and multiple consumers can read from one. A consumer uses an offset into the stream to indicate the event it wants to read. Kafka does not delete an event from the stream when it is ack'd by the consumer; instead a consumer increments its offset once an item has been read so that it can avoid processing the same event twice. See Offset Management for more on how Brighter manages consumer offsets. As a result the lifetime of events on a stream is instead a configuration setting for the stream.

As a consumer manages an offset to record events that is has read, you cannot scale an application that wishes to consume a topic by increasing the number of consumers--they don't share an offset--without partitioning the topic. If you supply a partition key, a partition uses consistent hashing to slice a topic into a number of streams; otherwise it will use round-robin. See this documentation for more. See Kafka Hash Partitioning for how to control the hashing algorithm that Brighter uses to map a partition key to a partition. Each partition is only read by a single consumer within the application. All of the consumers for an application should share the same group id, called a consumer group in Kafka. As each consumer tracks the offset for the partitions it is reading, it is possible to have multiple consumers read and process the same topic.

A consumer may read from multiple partitions, but only one consumer may read from a partition at one time in a given consumer group. Kafka will assign partitions across the pool of consumers for the consumer group. When the pool changes, a rebalance occurs, which may mean that a consumer changes the partition that it is assigned within the consumer group. Brighter favors sticky assignment of partitions to avoid unnecessary churn of partitions.

In addition to the Producer API and Consumer API Kafka streams have features such as the Streams API and the Connect API. We do not use either of these from Brighter.

Kafka Connection

The Connection to Kafka is provided by a KafkaMessagingGatewayConfiguration which allows you to configure the following:

• BootStrapServers: A bootstrap server is a well-known broker through which we discover the servers in the Kafka cluster that we can connect to. You should supply a comma-separated list of host and port pairs. These are the addresses of the Kafka brokers in the "bootstrap" Kafka cluster.

• Debug: A comma-separated list of debug contexts to enable. Producer: broker, topic, msg. Consumer: consumer, cgrp, topic, fetch.

• Name: An identifier to use for the client.

• SaslMechanisms: If any, what is the protocol used for authenticated connection to the Kafka broker: plain, scram-sha-256, scram-sha-256, gssapi (kerberos), oauthbearer

• SaslKerberosPrincipal: If using kerberos, what is the connection name.

• SaslUsername: SASL username for use with PLAIN and SASL-SCRAM

• SaslPassword: SASL password for use with PLAIN and SASL-SCRAM

• SecurityProtocol: How are messages between client and server encrypted, if at all: plaintext, ssl, saslplaintext, saslssl

• SslCaLocation: Where is the CA certificate located (see here for guidance).

• SslKeystoreLocation: Path to the client's keystore

• SslKeystorePassword: Password for the client's keystore

Kafka Connection Options

The type is KafkaMessagingGatewayConfiguration, which is the name every example on this page uses. It takes its options as properties, so the option is the property you set.

Option	Type	Default	Description
BootStrapServers	string[]	null	Host and port pairs for the brokers the client discovers the cluster through.
Debug	string?	null	A comma-separated list of librdkafka debug contexts to enable.
Name	string?	null	Identifies the client to the broker.
SaslMechanisms	SaslMechanism?	null	The SASL mechanism used to authenticate to the broker.
SaslKerberosPrincipal	string?	null	The Kerberos principal used when the mechanism is GSSAPI.
SaslUsername	string?	null	The SASL username used with PLAIN and SCRAM.
SaslPassword	string?	null	The SASL password used with PLAIN and SCRAM.
SecurityProtocol	SecurityProtocol?	null	How traffic between client and broker is encrypted.
SslCaLocation	string?	null	Path to the CA certificate that signs the broker's certificate.
SslKeystoreLocation	string?	null	Path to the client's keystore.
SslKeystorePassword	string?	null	Password for the client's keystore.

Two spellings are easy to get wrong: BootStrapServers carries a capital S in the middle, and the Kerberos option is SaslKerberosPrincipal.

The following code connects to a local Kafka instance (for development):

	services.AddBrighter(...)
	.AddProducers((configure) =>
	{
	    configure.ProducerRegistry = new KafkaProducerRegistryFactory(
		new KafkaMessagingGatewayConfiguration()
		{
		 	Name = "paramore.brighter.greetingsender",
			BootStrapServers = new[] {"localhost:9092"}
		},
		...//publication, see below
		)
	    .Create();
	})
	...

The following code connects to a remote Kafka instance. The settings here will depend on how your production broker is configured for access. We show getting secrets from environment variables for simplicity, again you will need to adjust this for your approach to secrets management:

	services.AddBrighter(...)
	.AddProducers((configure) =>
        {
            configure.ProducerRegistry = new KafkaProducerRegistryFactory(
		new KafkaMessagingGatewayConfiguration()
		{
			Name = "paramore.brighter.greetingsender",
			BootStrapServers = new[] { Environment.GetEnvironmentVariable("BOOSTRAP_SERVER")},
			SecurityProtocol = Paramore.Brighter.MessagingGateway.Kafka.SecurityProtocol.SaslSsl,
			SaslMechanisms = Paramore.Brighter.MessagingGateway.Kafka.SaslMechanism.Plain,
			SaslUsername = Environment.GetEnvironmentVariable("SASL_USERNAME"),
			SaslPassword = Environment.GetEnvironmentVariable("SASL_PASSWORD"),
			SslCaLocation = RuntimeInformation.IsOSPlatform(OSPlatform.OSX) ? "/usr/local/etc/openssl@1.1/cert.pem" : null;
		},
		...//publication, see below
		)
	    .Create();
	})
	...

Kafka Publication

For more on a Publication see the material on an Add Producers in Command Processor Configuration Reference.

We allow you to configure properties for both Brighter and the Confluent .NET client. Because there are many properties on the Confluent .NET Client we also configure a callback to let you inspect and modify the configuration that we will pass to the client if you so desire. This can be used to add properties we do not support or adjust how we set them.

• Replication: how many ISR nodes must receive the record before the producer can consider the write successful. Default is Acks.All.

• BatchNumberMessages: Maximum number of messages batched in one MessageSet. Default is 10000.

• EnableIdempotence: Messages are produced once only. Will adjust the following if not set: max.in.flight.requests.per.connection=5 (must be less than or equal to 5), retries=INT32_MAX (must be greater than 0), acks=all, queuing.strategy=fifo. Default is true.

• LingerMs: Maximum time, in milliseconds, for buffering data on the producer queue. Default is 5.

• MessageSendMaxRetries: How many times to retry sending a failing MessageSet. Note: retrying may cause reordering, set the max in flight to 1 if you need ordering by when sent. Default is 3.

• MessageTimeoutMs: Local message timeout. This value is only enforced locally and limits the time a produced message waits for successful delivery. A time of 0 is infinite. Default is 5000.

• MaxInFlightRequestsPerConnection: Maximum number of in-flight requests the client will send. We default this to 1, so as to allow retries to not de-order the stream.

• NumPartitions: How many partitions for this topic. We default to 1.

• Partitioner: How do we map a partition key to a partition? Defaults to Partitioner.ConsistentRandom, but we recommend Partitioner.Murmur2Random for a more even distribution of messages across partitions. See Kafka Hash Partitioning below for the supported values and the differences between them.

• QueueBufferingMaxMessages: Maximum number of messages allowed on the producer queue. Defaults to 100000.

• QueueBufferingMaxKbytes: Maximum total message size sum allowed on the producer queue. Defaults to 1048576 bytes (so for 10 messages about 104Kb per message).

• ReplicationFactor: What is the replication factor? How many nodes is the topic copied to on the broker? Defaults to 1.

• RetryBackoff: The backoff time before retrying a message send. Defaults to 100.

• RequestTimeoutMs: The ack timeout of the producer request. This value is only enforced by the broker and relies on Replication being != AcksEnum.None. Defaults to 500.

• TopicFindTimeoutMs: How long to wait when asking for topic metadata. Defaults to 5000.

• TransactionalId: The unique identifier for this producer, used with transactions

Kafka Publication Options

KafkaPublication takes its options as properties and adds these seventeen to the base publication options, which it inherits.

Option	Type	Default	Description
Replication	Acks	All	In-sync replicas that must acknowledge a write before it is complete.
BatchNumberMessages	int	10000	Messages batched into one MessageSet.
EnableIdempotence	bool	true	Whether the producer guarantees each message is written once.
LingerMs	int	5	Milliseconds data is buffered on the producer queue before it is sent.
MessageSendMaxRetries	int	3	Times a failing MessageSet is retried.
MessageTimeoutMs	int	5000	Milliseconds a produced message waits locally for successful delivery; 0 is infinite.
MaxInFlightRequestsPerConnection	int	1	Requests the client sends before waiting for a response.
NumPartitions	int	1	Partitions given to the topic when Brighter creates it.
Partitioner	Partitioner	ConsistentRandom	Maps a partition key to a partition.
QueueBufferingMaxMessages	int	100000	Messages allowed on the producer queue.
QueueBufferingMaxKbytes	int	1048576	Total size in kilobytes allowed on the producer queue.
ReplicationFactor	short	1	Broker nodes the topic is copied to when Brighter creates it.
RetryBackoff	int	100	Milliseconds before a failed send is retried.
RequestTimeoutMs	int	500	Milliseconds the broker waits to acknowledge a producer request.
TopicFindTimeoutMs	int	5000	Milliseconds a request for topic metadata waits.
TransactionalId	string?	null	Identifies this producer across restarts when using transactions.
MessageHeaderBuilder	IKafkaMessageHeaderBuilder	a KafkaDefaultMessageHeaderBuilder	Maps a Brighter message's headers onto Kafka headers.

MaxInFlightRequestsPerConnection is 1 rather than the Confluent client's own 5, so that a retry cannot re-order the stream; raising it trades ordering for throughput. MessageHeaderBuilder is the extension point for the mapping between a Brighter message's headers and Kafka's.

The following example shows how a Publication might be configured:

	services.AddBrighter(...)
	.AddProducers((configure) =>
        {
            configure.ProducerRegistry = new KafkaProducerRegistryFactory(
		...,//connection see above
		new KafkaPublication[] {new KafkaPublication()
                {
                    Topic = new RoutingKey("MyTopicName"),
                    NumPartitions = 3,
                    ReplicationFactor = 3,
                    MessageTimeoutMs = 1000,
                    RequestTimeoutMs = 1000,
                    MakeChannels = OnMissingChannel.Create 
                }}
		).Create();
	})

Kafka Hash Partitioning

A Kafka topic is split into partitions, and the producer decides which partition each message is written to. The algorithm that makes this decision is the partitioner, which Brighter exposes through the Partitioner property on a Publication. Brighter's Partitioner enum maps directly onto the Confluent .NET client's partitioner setting (from librdkafka).

How the partitioner behaves depends on whether the message has a partition key. You set the partition key on the message header in your message mapper:

public Message MapToMessage(GreetingEvent request)
{
	var header = new MessageHeader(request.Id, "greeting.event", MessageType.MT_EVENT)
	{
		//Messages with the same partition key go to the same partition
		PartitionKey = request.CustomerId.ToString()
	};
	...
}

When a partition key is present, the partitioner hashes the key and selects a partition deterministically: all messages with the same key are written to the same partition, which preserves their order relative to one another (and means they are handled by the same consumer in a consumer group). When no key is set, the behavior depends on the partitioner variant, as described below.

Supported Partitioners

Brighter supports the following partitioners:

Partitioner	Keyed messages	Unkeyed messages	Notes
Random	Random partition	Random partition	Ignores the partition key entirely, so there are no per-key ordering guarantees.
Consistent	CRC32 hash of the key	Always the same (single) partition	librdkafka's legacy consistent partitioner. CRC32 can cluster keys, risking uneven distribution.
ConsistentRandom	CRC32 hash of the key	Random partition	Brighter's default. CRC32 can cluster keys, risking uneven distribution.
Murmur2	Murmur2 hash of the key	Always the same (single) partition	Good key distribution, but the single partition for unkeyed messages can become a hot spot.
Murmur2Random	Murmur2 hash of the key	Random partition	The most even distribution for both keyed and unkeyed messages. Recommended.

The difference between the Consistent* family and the Murmur2* family is the hash function used to map a key to a partition: CRC32 versus Murmur2. Because the hash functions differ, the same key maps to a different partition under each family. The difference between each base variant and its Random counterpart is what happens to messages without a partition key: the base variants (Consistent, Murmur2) always write unkeyed messages to the same single partition, whereas the Random variants (ConsistentRandom, Murmur2Random) spread them randomly across partitions. A single partition for unkeyed messages can become a bottleneck and a hot spot, so the Random variants are generally preferable.

Why We Recommend Murmur2Random

We recommend setting Partitioner to Partitioner.Murmur2Random because of how it distributes messages across partitions:

1. More even distribution of keyed messages: Murmur2 generally spreads keys more uniformly across partitions than the CRC32 hash used by Consistent and ConsistentRandom. This matters because uneven distribution creates "hot" partitions: a few partitions receive a disproportionate share of the messages, so the consumers assigned to those partitions become a bottleneck while the remaining consumers sit underused. Because only one consumer in a group may read a partition at a time, a hot partition caps your effective throughput and grows consumer lag no matter how many consumers you add.

2. No single partition for unkeyed messages: where Murmur2 (and Consistent) send every message without a partition key to the same partition—concentrating all of that load on one partition, and therefore on one consumer—Murmur2Random spreads unkeyed messages randomly across all partitions.

	new KafkaPublication[]
	{
		new KafkaPublication()
		{
			Topic = new RoutingKey("MyTopicName"),
			NumPartitions = 3,
			Partitioner = Partitioner.Murmur2Random,
			MakeChannels = OnMissingChannel.Create
		}
	}

Note that changing the partitioner on an existing topic changes where keys land: messages with the same key may be written to different partitions before and after the change, which can break per-key ordering during the transition. Plan such a change for a deployment window where this is acceptable, or apply it when you create a new topic.

Publication Configuration Callback

The Confluent .NET client has a range of configuration options. Some of those can be controlled through the publication. But, to allow you the full range of configuration options for the Confluent client, including new options that may appear, we provide a callback on the KafkaProducerRegistryFactory. The registry exposes a method, SetConfigHook(Action hook). The method takes a delegate (you can pass a lambda). Your delegate will be called with the proposed ProducerConfig (taking into account the Publication settings). You can adjust additional parameters at this point.

You can use it as follows:

	var publication = new KafkaPublication()
	{
		Topic = new RoutingKey("MyTopicName"),
		NumPartitions = 3,
		ReplicationFactor = 3,
		MessageTimeoutMs = 1000,
		RequestTimeoutMs = 1000,
		MakeChannels = OnMissingChannel.Create 
	};
	publication.SetConfigHook(config => config.EnableGaplessGuarantee = true)

	services.AddBrighter(...)
	.AddProducers((configure) =>
        {
            configure.ProducerRegistry = new KafkaProducerRegistryFactory(
		...,//connection see above
		new KafkaPublication() {publication}
	    ).Create();
	}) 

Kafka Topic Auto Create

Brighter uses the Kafka AdminClient for topic creation. For this to work as expected you should set the server property of auto.create.topics.enable to false; otherwise the topic will be auto-created with the values defined by your server for new topics, such as the number of partitions. This error can be insidious because your code will still work against this topic, but without inspection you will not observe that its properties do not match those requested.

If you want to specify the topic through Brighter, or through your own IaaS code, we recommend always setting this setting to false; we recommend only setting it to true if you tell Brighter to assume that the infrastructure exists, as it will then be created on the first write.

Kafka Subscription

For more on a Subscription see the material on configuring the Dispatcher in Basic Configuration.

We support a number of Kafka specific Subscription options:

• CommitBatchSize: We commit processed work (marked as acked or rejected) when a batch size worth of work has been completed (see below). Defaults to 10.

• ConfigHook: Allows you to modify the Kafka client configuration before a consumer is created. Used to set properties that Brighter does not expose. See Configuration Callback below.

• GroupId: Only one consumer in a group can read from a partition at any one time; this preserves ordering. We do not default this value, and expect you to set it.

• GroupProtocol: Selects the Kafka consumer group protocol: ClassicGroupProtocol (default) or ConsumerGroupProtocol (KIP-848). See Consumer Group Protocol (KIP-848) below. Defaults to null, which uses the classic protocol.

• IsolationLevel: Default to read only committed messages, change if you want to read uncommitted messages. May cause duplicates. Defaults to ReadCommitted.

• MaxPollInterval: How often the consumer needs to poll for new messages to be considered alive, polling greater than this interval triggers a re-balance. Defaults to 300000ms (5 minutes).

• NumPartitions: How many partitions does the topic have? Used for topic creation, if required. Defaults to 1.

• OffsetDefault: What do we do if there is no offset stored for this consumer. Defaults to AutoOffsetReset.Earliest - Begin reading the stream from the start. Options include AutoOffsetReset.Latest - Start from now i.e. only consume messages after we start and AutoOffsetReset.Error - which considers it an error if no reset is found

• PartitionAssignmentStrategy: How do partitions get assigned to consumers in the group? Defaults to RoundRobin for even distribution. See Partition Assignment Strategy below. Deprecated: set PartitionAssignmentStrategy on a ClassicGroupProtocol via GroupProtocol instead.

• ReadCommittedOffsetsTimeOut: How long before attempting to read back committed offsets (mainly used in debugging) is an error. Defaults to 5000ms.

• ReplicationFactor: What is the replication factor? How many nodes is the topic copied to on the broker? Defaults to 1. Used for topic creation if required.

• SessionTimeout: If Kafka does not receive a heartbeat from the consumer within this time window, trigger a re-balance. Defaults to 10000ms (10 seconds). Deprecated: set SessionTimeout on a ClassicGroupProtocol via GroupProtocol instead.

• SweepUncommittedOffsetsInterval: The interval at which we sweep, looking for offsets that have not been flushed (see below). Defaults to 30000ms (30 seconds).

Kafka Subscription Options

KafkaSubscription takes its options as constructor arguments, so the option is the parameter you type. The seventeen it shares with Subscription behave the same way here; the other thirteen are Kafka's own, and this is the widest subscription table in the documentation.

Option	Type	Default	Description
subscriptionName	SubscriptionName	none	Names the subscription for diagnostics; read back as Name.
channelName	ChannelName	none	Names the channel this subscription reads.
routingKey	RoutingKey	none	The Kafka topic the consumer subscribes to.
requestType	Type?	none	The request type messages on this topic are translated into.
getRequestType	Func<Message, Type>?	derives the type from requestType	Determines the request type from the message rather than from the topic.
groupId	string?	null	The consumer group this consumer joins.
bufferSize	int	1	Messages read from the topic at once and held in the channel.
noOfPerformers	int	1	Threads reading this topic, each with its own message pump.
timeOut	TimeSpan?	300 ms	How long a read waits before treating the topic as empty.
requeueCount	int	-1	Times a message is requeued before it is treated as a poison pill; -1 is unlimited.
requeueDelay	TimeSpan?	0 ms	How long delivery of a requeued message is delayed.
unacceptableMessageLimit	int	0	Unacceptable messages before the channel stops; 0 disables the limit.
unacceptableMessageLimitWindow	TimeSpan?	null	The window the unacceptable-message count resets at the end of.
offsetDefault	AutoOffsetReset	Earliest	Where the consumer starts when the group has no stored offset.
commitBatchSize	long	10	Completed messages whose offsets are committed in one batch.
sessionTimeout	TimeSpan?	10000 ms	How long Kafka waits for a heartbeat before rebalancing the group.
maxPollInterval	TimeSpan?	300000 ms	How long Kafka waits for a poll before rebalancing the group.
sweepUncommittedOffsetsInterval	TimeSpan?	30000 ms	How often offsets that never reached a batch are swept and committed.
isolationLevel	IsolationLevel	ReadCommitted	Whether uncommitted messages are read.
messagePumpType	MessagePumpType	none	Selects the Reactor or Proactor concurrency model.
numOfPartitions	int	1	Partitions given to the topic when Brighter creates it; read back as NumPartitions.
replicationFactor	short	1	Broker nodes the topic is copied to when Brighter creates it.
channelFactory	IAmAChannelFactory?	null	Creates the channel; falls back to DefaultChannelFactory when null.
makeChannels	OnMissingChannel	Create	Whether Brighter creates the topic, validates it, or assumes it.
emptyChannelDelay	TimeSpan?	500 ms	How long the pump pauses after a read that found no message.
channelFailureDelay	TimeSpan?	1000 ms	How long the pump pauses after a channel failure.
partitionAssignmentStrategy	PartitionAssignmentStrategy	RoundRobin	How partitions are assigned across the consumer group.
configHook	Action<ConsumerConfig>?	null	Modifies the Confluent ConsumerConfig before the consumer is created.
deadLetterRoutingKey	RoutingKey?	null	The topic messages are dead-lettered to.
invalidMessageRoutingKey	RoutingKey?	null	The topic unacceptable messages are routed to.

makeChannels is Create, so Brighter creates a topic it cannot find, using numOfPartitions and replicationFactor; set it to Validate where the topic is declared by your infrastructure. sessionTimeout and partitionAssignmentStrategy are deprecated in favour of the same settings on a ClassicGroupProtocol, as the bullets above record.

Four options are set as properties after construction rather than as constructor arguments, so they are not on the table above: GroupProtocol, ReadCommittedOffsetsTimeOut (5000 ms), TopicFindTimeout (5000 ms) and TimeProvider. The bullets above cover the first two.

The generic form KafkaSubscription<T>, which every example below uses, takes the same options and supplies four defaults the table cannot: requestType is T, subscriptionName, channelName and routingKey are T's full name, and messagePumpType is Reactor — Kafka is the one transport here whose generic subscription defaults to the Reactor rather than the Proactor.

The following example shows how a subscription might be configured:

// ...
 private static void ConfigureBrighter(HostBuilderContext hostContext, IServiceCollection services)
{
	var subscriptions = new KafkaSubscription[]
	{
		new KafkaSubscription<GreetingEvent>(
			subscriptionName: new SubscriptionName("paramore.example.greeting"),
			channelName: new ChannelName("greeting.event"),
			routingKey: new RoutingKey("greeting.event"),
			groupId: Environment.GetEnvironmentVariable("KAFKA_GROUPID"),
			timeOut: TimeSpan.FromMilliseconds(100),
			commitBatchSize: 5,
			sweepUncommittedOffsetsInterval: TimeSpan.FromMilliseconds(3000),
			messagePumpType: MessagePumpType.Reactor
		)
	};

	//create the gateway
	var consumerFactory = new KafkaMessageConsumerFactory(
	new KafkaMessagingGatewayConfiguration {...} // see connection information above
	);

	services.AddConsumers(options =>
	{
		options.Subscriptions = subscriptions;
		options.DefaultChannelFactory = new ChannelFactory(consumerFactory);
	}).AutoFromAssemblies();


	services.AddHostedService<ServiceActivatorHostedService>();
}

Subscription Configuration Callback

Similar to producers, the Confluent .NET client for consumers has a range of configuration options. Some of those can be controlled through the subscription. But, to allow you the full range of configuration options for the Confluent client, including new options that may appear, we provide a configHook parameter on KafkaSubscription.

The configHook takes a delegate (you can pass a lambda). Your delegate will be called with the proposed ConsumerConfig (taking into account the Subscription settings). You can adjust additional parameters at this point.

You can use it as follows:

// ...
var subscription = new KafkaSubscription<GreetingEvent>(
	subscriptionName: new SubscriptionName("paramore.example.greeting"),
	channelName: new ChannelName("greeting.event"),
	routingKey: new RoutingKey("greeting.event"),
	groupId: Environment.GetEnvironmentVariable("KAFKA_GROUPID"),
	timeOut: TimeSpan.FromMilliseconds(100),
	commitBatchSize: 5,
	sweepUncommittedOffsetsInterval: TimeSpan.FromMilliseconds(3000),
	messagePumpType: MessagePumpType.Reactor,
	configHook: config =>
	{
		// Customize the Confluent Consumer configuration
		config.FetchMinBytes = 1024;
		config.FetchWaitMaxMs = 500;
		config.StatisticsIntervalMs = 60000; // Enable statistics
	}
);

var subscriptions = new KafkaSubscription[] { subscription };

//create the gateway
var consumerFactory = new KafkaMessageConsumerFactory(
	new KafkaMessagingGatewayConfiguration {...} // see connection information above
);

services.AddConsumers(options =>
{
	options.Subscriptions = subscriptions;
	options.DefaultChannelFactory = new ChannelFactory(consumerFactory);
}).AutoFromAssemblies();

Common Configuration Callback Use Cases

The configHook is useful for:

1. Fine-tuning fetch behavior:

configHook: config =>
{
	config.FetchMinBytes = 1024;        // Minimum data per fetch request
	config.FetchWaitMaxMs = 100;        // Max time to wait for fetch.min.bytes
	config.FetchMaxBytes = 52428800;    // Maximum data for all partitions
}

2. Enabling consumer statistics:

configHook: config =>
{
	config.StatisticsIntervalMs = 30000; // Emit stats every 30 seconds
}

3. Customizing security settings:

configHook: config =>
{
	config.EnableSslCertificateVerification = true;
	config.SslEndpointIdentificationAlgorithm = SslEndpointIdentificationAlgorithm.Https;
}

4. Adjusting partition assignment:

configHook: config =>
{
	// Note: PartitionAssignmentStrategy should typically be set via the subscription parameter
	// But you can override it here if needed
	config.SessionTimeoutMs = 45000;  // Increase for slow rebalances
}

5. Debugging and monitoring:

configHook: config =>
{
	config.Debug = "consumer,cgrp,topic,fetch"; // Enable debug logging
	config.StatisticsIntervalMs = 10000;        // Frequent statistics for debugging
}

Partition Assignment Strategy

Kafka distributes partitions across consumers in a consumer group using a partition assignment strategy. Brighter provides control over this strategy through the partitionAssignmentStrategy parameter on the subscription, or the PartitionAssignmentStrategy property on a ClassicGroupProtocol (see Consumer Group Protocol (KIP-848)).

Deprecated: The partitionAssignmentStrategy parameter and the PartitionAssignmentStrategy property on KafkaSubscription are obsolete and will be removed in a future version. Configure the strategy on a ClassicGroupProtocol assigned to the subscription's GroupProtocol property instead. Note that partition assignment strategies only apply to the classic protocol; with KIP-848 partition assignment is broker-driven.

Available Strategies:

• RoundRobin (default): Distributes partitions evenly across consumers in a round-robin fashion. This is the most balanced distribution but may cause all partitions to be reassigned during rebalances.

• Range: Assigns contiguous partition ranges to consumers. Useful for co-locating related partitions on the same consumer.

• CooperativeSticky: Not supported with Brighter's manual offset commits (will throw an exception).

Why RoundRobin is the Default:

Brighter uses RoundRobin as the default strategy because:

1. It provides the most even distribution of partitions across consumers

2. It works reliably with Brighter's manual offset commit strategy

3. It's straightforward to reason about for most use cases

Using Range Strategy:

The Range strategy is useful when you want to co-locate partitions on the same consumer:

var subscription = new KafkaSubscription<GreetingEvent>(
	subscriptionName: new SubscriptionName("paramore.example.greeting"),
	channelName: new ChannelName("greeting.event"),
	routingKey: new RoutingKey("greeting.event"),
	groupId: Environment.GetEnvironmentVariable("KAFKA_GROUPID")
)
{
	GroupProtocol = new ClassicGroupProtocol
	{
		PartitionAssignmentStrategy = PartitionAssignmentStrategy.Range
	}
};

CooperativeSticky Limitation:

The CooperativeSticky strategy is not supported when using manual offset commits (which Brighter requires for its at-least-once delivery guarantees). Attempting to use it will throw an ArgumentOutOfRangeException:

// This will throw an exception:
var subscription = new KafkaSubscription<GreetingEvent>(
	// ...
)
{
	GroupProtocol = new ClassicGroupProtocol
	{
		PartitionAssignmentStrategy = PartitionAssignmentStrategy.CooperativeSticky // ❌ Not supported
	}
};

This is due to a known issue in librdkafka where CooperativeSticky doesn't work correctly with manual offset management.

Consumer Group Protocol (KIP-848)

KIP-848 is Kafka's next-generation consumer group protocol: group membership and partition assignment become broker-driven, which significantly reduces rebalance times and removes the "stop-the-world" rebalances of the classic protocol. It is generally available from Apache Kafka 4.0 (which also requires KRaft mode, with no ZooKeeper).

Brighter supports both protocols through the GroupProtocol property on KafkaSubscription, which takes an IGroupProtocol implementation:

• ClassicGroupProtocol (default): Kafka's original consumer group protocol, where partition assignment is computed by the clients. Carries SessionTimeout, HeartbeatInterval, and PartitionAssignmentStrategy.

• ConsumerGroupProtocol: The KIP-848 consumer protocol (group.protocol=consumer), where partition assignment is computed by the broker. Carries GroupRemoteAssignor (the broker-side assignor name) and GroupInstanceId (static membership).

Using the classic protocol explicitly:

var subscription = new KafkaSubscription<GreetingEvent>(
	subscriptionName: new SubscriptionName("paramore.example.greeting"),
	channelName: new ChannelName("greeting.event"),
	routingKey: new RoutingKey("greeting.event"),
	groupId: Environment.GetEnvironmentVariable("KAFKA_GROUPID")
)
{
	GroupProtocol = new ClassicGroupProtocol
	{
		SessionTimeout = TimeSpan.FromSeconds(30),
		HeartbeatInterval = TimeSpan.FromSeconds(10),
		PartitionAssignmentStrategy = PartitionAssignmentStrategy.RoundRobin
	}
};

When you don't supply a GroupProtocol, Brighter uses a ClassicGroupProtocol and back-fills any null properties from the consumer's constructor parameters — SessionTimeout (default 10 seconds) and PartitionAssignmentStrategy (default RoundRobin) — preserving the existing behavior. If you do supply a ClassicGroupProtocol, any values you have set are preserved; only null properties are back-filled.

Warning: The back-fill mutates the ClassicGroupProtocol instance in place. Do not share one instance across subscriptions — with a shared instance, the first consumer's back-filled values would silently apply to the second subscription as well. The instance is also applied to every consumer created from the subscription (one per performer).

Using the KIP-848 consumer protocol:

var subscription = new KafkaSubscription<GreetingEvent>(
	subscriptionName: new SubscriptionName("paramore.example.greeting"),
	channelName: new ChannelName("greeting.event"),
	routingKey: new RoutingKey("greeting.event"),
	groupId: Environment.GetEnvironmentVariable("KAFKA_GROUPID")
)
{
	GroupProtocol = new ConsumerGroupProtocol()
};

ConsumerGroupProtocol deliberately does not set session.timeout.ms, heartbeat.interval.ms, or partition.assignment.strategy: librdkafka rejects those properties when group.protocol=consumer. Heartbeats, session management, and partition assignment are broker-driven under KIP-848.

Broker requirement: The KIP-848 consumer protocol requires a broker that supports it (Apache Kafka 4.0 or later). Using ConsumerGroupProtocol against an older broker will fail.

Static membership:

ConsumerGroupProtocol.GroupInstanceId enables Kafka static membership (group.instance.id). Static membership requires a unique id per consumer instance. Because the subscription's GroupProtocol instance is applied to every consumer created from that subscription (one per performer), all of those consumers would register the same static id and collide on group join. Only set GroupInstanceId when the subscription has a single performer; to use static membership with multiple performers, assign a unique group.instance.id per consumer via the configHook instead:

configHook: config => config.GroupInstanceId = $"greeter-{Guid.NewGuid()}"

Infrastructure validation caveat:

With the consumer protocol, group membership is broker-driven and completes asynchronously: subscribing to a non-existent topic does not fail synchronously at consumer creation. Infrastructure validation via OnMissingChannel.Validate therefore has weaker guarantees than with the classic protocol and cannot reliably detect missing topics. Prefer OnMissingChannel.Create, or provision topics out-of-band (for example, via IaC).

Deprecated subscription properties:

KafkaSubscription.SessionTimeout and KafkaSubscription.PartitionAssignmentStrategy are now marked obsolete and will be removed in a future version. Their values are still honored as back-fill defaults for the classic protocol, but new code should configure these values on a ClassicGroupProtocol instead.

Offset Management

It is important to understand how Brighter manages the offset of any partitions assigned to your consumer.

• Brighter manages committing offsets to Kafka. This means we set the Confluent client's auto store and auto commit properties to false.

• The CommitBatchSize setting on the Subscription determines the size of your buffer. A smaller buffer is less efficient, but if your consumer crashes any offsets pending commit in the buffer will be lost, and you will be represented with those records when you next read from the partition. We default this value to 10.

• We do not add an offset commit to the buffer until you Ack the request. The message pump will Ack for you once you exit your handler (via return or throwing an exception).

• Flushing the commit buffer happens on a separate thread. We only run one flush at a time, and we flush a CommitBatchSize number of items from the buffer.

  • A busy consumer may not flush on every increment of the CommitBatchSize, as it may need to wait for the last flush to finish.

  • We won't flush again until we cross the next multiple of the CommitBatchSize. For example if the CommitBatchSize is 10, and the handler is busy so that by the time the buffer flushes there are 13 pending commits in the buffer, the buffer would only flush 10, and 3 would remain in the buffer; we would not flush the next 10 until the buffer hit 20.

  • If your CommitBatchSize is too low for the throughput, you might find that you miss a flush interval, because you are already flushing.

  • If you miss a flush on a busy consumer, your buffer will begin to back up. If this continues, you will not catch up with subsequent flushes, which only flush the CommitBatchSize each time. This would lead to you continually being "backed up".

  • For this reason you must set a CommitBatchSize that keeps pace with the throughput of your consumer. Use a larger CommitBatchSize for higher throughput consumers, smaller for lower.

• We sweep uncommitted offsets at an interval. This triggers a flush if no flush has run since the last flush plus the Subscription's SweepUncommittedOffsetsIntervalMs.

  • A sweep will not run if a flush is currently running (and will in turn block a flush).

  • A sweep flushes a CommitBatchSize worth of commits.

  • It is intended for low-throughput consumers where commits might otherwise languish waiting for a batch-size increment.

  • It is not intended to flush a buffer that backs up because the CommitBatchSize is too low, and won't function for that. Fix the CommitBatchSize instead.

• On a re-balance where we stop processing a partition on an individual consumer, we flush the remaining offsets for the revoked partitions.

  • We configure the consumer to use sticky assignment strategy to avoid unnecessary re-assignments (see the Confluent documentation).

• On a consumer shutdown we flush the buffer to commit all offsets.

Working with Schema Registry

If you want to use tools within the Kafka ecosystem such as Kafka Connect or KQSL you will almost certainly need to use Confluent Schema Registry to provide the schema of your message.

You will need to pull in the following package:

• Confluent.SchemaRegistry

and a package for the serialization of your choice. Here we are using JSON, so we use

• Confluent.SchemaRegistry.Serdes.Json

When working with Brighter, to use Confluent Schema Registry you will need to take a dependency on ISchemaRegistry in the constructor of your message mapper. To fulfill this constructor, in your application setup you will need to register an instance of schema registry. You should configure the schema registry config url to be the url of you schema registry. (Here we just use localhost for a development instance running in docker as an example).

var schemaRegistryConfig = new SchemaRegistryConfig { Url = "http://localhost:8081"};
var cachedSchemaRegistryClient = new CachedSchemaRegistryClient(schemaRegistryConfig);
services.AddSingleton<ISchemaRegistryClient>(cachedSchemaRegistryClient);

Once you can satisfy the dependency, you will want to use the serializer from the Serdes package to serialize the body of your message, instead of System.Text.Json. Note that 'under-the-hood' the Serdes serializer uses Json.NET and NJsonSchema, so you may need to mark up your code with attributes from these packages to create the schema you want and serialize a valid message to it. (Note that, at this time, the Serdes package does not support System.Text.Json so you will need to take a dependency on Json.NET if you want to use the schema registry).

It is worth noting the following aspects of the code sample below:

• We need to set up a SerializationContext and tell Serdes that we are serializing the message body using their serializer

• We provide two helpers, though you can pass your own settings if you prefer:

  • ConfluentJsonSerializationConfig.SerdesJsonSerializerConfig() offers default settings for JSON serialization (many of these are passed through to Json.NET).

  • ConfluentJsonSerializationConfig.NJsonSchemaGeneratorSettings() offers default settings for JSON Schema generation (such as using camelCase).

public class GreetingEventMessageMapper : IAmAMessageMapper<GreetingEvent>
{
	private readonly ISchemaRegistryClient _schemaRegistryClient;
	private readonly string _partitionKey = "KafkaTestQueueExample_Partition_One";
	private SerializationContext _serializationContext;
	private const string Topic = "greeting.event";

	public GreetingEventMessageMapper(ISchemaRegistryClient schemaRegistryClient)
	{
		_schemaRegistryClient = schemaRegistryClient;
		//We care about ensuring that we serialize the body using the Confluent tooling, as it registers and validates schema
		_serializationContext = new SerializationContext(MessageComponentType.Value, Topic);
	}

	public Message MapToMessage(GreetingEvent request)
	{
		var header = new MessageHeader(messageId: request.Id, topic: Topic, messageType: MessageType.MT_EVENT);
		//This uses the Confluent JSON serializer, which wraps Newtonsoft but also performs schema registration and validation
		var serializer = new JsonSerializer<GreetingEvent>(_schemaRegistryClient, ConfluentJsonSerializationConfig.SerdesJsonSerializerConfig(), ConfluentJsonSerializationConfig.NJsonSchemaGeneratorSettings()).AsSyncOverAsync();
		var s = serializer.Serialize(request, _serializationContext);
		var body = new MessageBody(s, "JSON");
		header.PartitionKey = _partitionKey;

		var message = new Message(header, body);
		return message;
	}

	public GreetingEvent MapToRequest(Message message)
	{
		var deserializer = new JsonDeserializer<GreetingEvent>().AsSyncOverAsync();
		//This uses the Confluent JSON serializer, which wraps Newtonsoft but also performs schema registration and validation
		var greetingCommand = deserializer.Deserialize(message.Body.Bytes, message.Body.Bytes is null, _serializationContext);
		
		return greetingCommand;
	}
}

Requeue with Delay (Non-Blocking Retry)

We don't currently support requeue with delay for Kafka. This is also known as non-blocking retry. With a stream if your app cannot process a record but it might be able to process the record after a delay (for example the DB is temporarily unavailable) then the options are:

• Blocking Retry - keep retrying the processing of this record

• Load Shedding - ack the record to commit the offset, skipping this record

• Non-Blocking Retry - move the record to a new store or queue, skipping the original, append after a delay

Brighter supports the first two of these options.

• Blocking Retry - use a Polly policy via the UsePolicy attribute

• Load Shedding - allow the handler to complete, or throw an exception. This will cause the handler to commit the offset.

Note that Blocking Retry means you will apply backpressure as the blocking retry means you will pause consumption until the record can be processed.

A non-blocking retry typically creates a copy of the current record, and appends it to the stream so that it can be processed later:

• Publish the message to be requeued to a new stream or store with a timestamp

• Ack the existing message so at to commit the offset

• Poll that stream or store and publish anything whose timestamp + delay means it is now due

(You may need multiple tables or streams to support different delay lengths)

Until Brighter supports this for you, implementation of non-blocking consumers is left to the user.