Apache Kafka Provider for SlimMessageBus

Please read the Introduction before reading this provider documentation.

Underlying client

The SMB Kafka implementation uses confluent-kafka-dotnet .NET wrapper around the native librdkafka library.

When troubleshooting or fine tuning it is worth reading the librdkafka and confluent-kafka-dotnet docs:

Configuration properties

Producer, consumer and global configuration properties are described here. The configuration on the underlying Kafka client can be adjusted like so:

services.AddSlimMessageBus(mbb =>
{
   // ...
   mbb.WithProviderKafka(cfg =>
   {
      cfg.BrokerList = kafkaBrokers;
      cfg.ProducerConfig = (config) =>
      {
         // adjust the producer config
      };
      cfg.ConsumerConfig = (config) =>
      {
         // adjust the consumer config
      };
   });
});

Minimizing message latency

There is a good description here on improving the latency by applying producer/consumer settings on librdkafka. Here is how you enter the settings using SlimMessageBus:

services.AddSlimMessageBus(mbb =>
{
   mbb.WithProviderKafka(cfg =>
   {
      cfg.BrokerList = kafkaBrokers;

      cfg.ProducerConfig = (config) =>
      {
         config.LingerMs = 5; // 5ms
         config.SocketNagleDisable = true;
      };
      cfg.ConsumerConfig = (config) =>
      {
         config.FetchErrorBackoffMs = 1;
         config.SocketNagleDisable = true;
      };
   });
});

There is also a good discussion around latency in this issue.

More documentation here:

SSL and password authentication

Example on how to configure SSL with SASL authentication (for cloudkarafka.com):

services.AddSlimMessageBus(mbb =>
{
   mbb.WithProviderKafka(cfg =>
   {
      cfg.BrokerList = kafkaBrokers;

      cfg.ProducerConfig = (config) =>
      {
         AddSsl(kafkaUsername, kafkaPassword, config);
      };
      cfg.ConsumerConfig = (config) =>
      {
         AddSsl(kafkaUsername, kafkaPassword, config);
      };
   });
});

private static void AddSsl(string username, string password, ClientConfig c)
{
   c.SecurityProtocol = SecurityProtocol.SaslSsl;
   c.SaslUsername = username;
   c.SaslPassword = password;
   c.SaslMechanism = SaslMechanism.ScramSha256;
   c.SslCaLocation = "cloudkarafka_2020-12.ca";
}

The file cloudkarafka_2020-12.ca has to be set to Copy to Output Directory as Copy always.

Selecting message partition for topic producer

Kafka topics are broken into partitions. The question is how does SMB Kafka choose the partition to assign the message? There are two possible options:

Default partitioner with message key

Currently, confluent-kafka-dotnet does not support custom partitioners (see here). The default partitioner is supported, which works in this way:

SMB Kafka allows to set a provider (selector) that will assign the message key for a given message and topic pair. Here is an example:

// MessageBusBuilder mbb;
mbb
   .Produce<MultiplyRequest>(x =>
   {
      x.DefaultTopic("topic1");
      // Message key could be set for the message
      x.KeyProvider((request, topic) => Encoding.ASCII.GetBytes((request.Left + request.Right).ToString()));
   })
   .WithProviderKafka(cfg => cfg.BrokerList = kafkaBrokers);

The key must be a byte[].

Assigning partition explicitly

SMB Kafka allows to set a provider (selector) that will assign the partition number for a given message and topic pair. Here is an example:

// MessageBusBuilder mbb;
mbb
   .Produce<PingMessage>(x =>
   {
      x.DefaultTopic("topic1");
      // Partition #0 for even counters
      // Partition #1 for odd counters
      x.PartitionProvider((message, topic) => message.Counter % 2);
   })
   .WithProviderKafka(cfg => cfg.BrokerList = kafkaBrokers);

With this approach your provider needs to know the number of partitions for a topic.

Consumer context

The consumer can implement the IConsumerWithContext interface to access the Kafka native message:

public class PingConsumer : IConsumer<PingMessage>, IConsumerWithContext
{
   public IConsumerContext Context { get; set; }

   public Task OnHandle(PingMessage message)
   {
      // SMB Kafka transport specific extension:
      var transportMessage = Context.GetTransportMessage();
      var partition = transportMessage.TopicPartition.Partition;
   }
}

This could be useful to extract the message’s offset or partition.

Message Headers

SMB uses headers to pass additional metadata information with the message. This includes the MessageType (of type string) or in the case of request/response messages the RequestId (of type string), ReplyTo (of type string) and Expires (of type long).

The Kafka message header values are natively binary (byte[]) in the underlying .NET client, as a result SMB needs to serialize the header values. By default the DefaultKafkaHeaderSerializer is used to serialize header values. If you need to specify a different serializer provide a specfic IMessageSerializer implementation (custom or one of the available serialization plugins):

// MessageBusBuilder mbb;
mbb
   .WithProviderKafka(cfg =>
   {
      cfg.BrokerList = kafkaBrokers;
      cfg.HeaderSerializer = new DefaultKafkaHeaderSerializer() // specify a different header values serializer
   });

Since version 2.0.0, uses the DefaultKafkaHeaderSerializer serializer which converts the passed values into string. Prior version 2.0.0, by default the same serializer for the bus was used to also serialize message header values.

Consumers

Offset Commit

In the current Kafka provider implementation, SMB handles the manual commit of topic-partition offsets for the consumer. This configuration is controlled through the following methods on the consumer builder:

Here’s an example:

mbb.Consume<PingMessage>(x =>
{
    x.Topic(topic)
        .WithConsumer<PingConsumer>()
        .KafkaGroup("subscriber")
        .Instances(2)
        .CheckpointEvery(1000)
        .CheckpointAfter(TimeSpan.FromSeconds(600));
});

In future versions, the following features may be added:

Consumer Error Handling

When an error occurs during message processing, SMB attempts to resolve a custom error handler to handle the exception.

The error handler can perform the following actions:

If no custom error handler is provided, the provider logs the exception and moves on to process the next message.

Debugging

Kafka uses a sophisticated protocol for partition assignment:

To better understand what’s happening, you can enable Debug level logging in your library, such as SlimMessageBus.Host.Kafka.KafkaGroupConsumer.

At this logging level, you can track the lifecycle events of the consumer group:

[00:03:06 INF] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Subscribing to topics: 4p5ma6io-test-ping
[00:03:06 INF] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Consumer loop started
[00:03:12 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Assigned partition, Topic: 4p5ma6io-test-ping, Partition: [0]
[00:03:12 INF] SlimMessageBus.Host.Kafka.KafkaPartitionConsumer Creating consumer for Group: subscriber, Topic: 4p5ma6io-test-ping, Partition: [0]
[00:03:12 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Assigned partition, Topic: 4p5ma6io-test-ping, Partition: [1]
[00:03:12 INF] SlimMessageBus.Host.Kafka.KafkaPartitionConsumer Creating consumer for Group: subscriber, Topic: 4p5ma6io-test-ping, Partition: [1]
...
[00:03:15 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Received message with Topic: 4p5ma6io-test-ping, Partition: [1], Offset: 98578, payload size: 57
[00:03:15 INF] SlimMessageBus.Host.Kafka.Test.KafkaMessageBusIt.PingConsumer Got message 073 on topic 4p5ma6io-test-ping.
[00:03:15 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Received message with Topic: 4p5ma6io-test-ping, Partition: [1], Offset: 98579, payload size: 57
[00:03:15 INF] SlimMessageBus.Host.Kafka.Test.KafkaMessageBusIt.PingConsumer Got message 075 on topic 4p5ma6io-test-ping.
[00:03:16 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Reached end of partition, Topic: 4p5ma6io-test-ping, Partition: [0], Offset: 100403
[00:03:16 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Commit Offset, Topic: 4p5ma6io-test-ping, Partition: [0], Offset: 100402
[00:03:16 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Reached end of partition, Topic: 4p5ma6io-test-ping, Partition: [1], Offset: 98580
[00:03:16 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Commit Offset, Topic: 4p5ma6io-test-ping, Partition: [1], Offset: 98579

Deployment

The librdkafka distribution for Windows requires Visual C++ Redistributable for 2013 installed on the server. More information can be found here.