Kafka: Configuration

General

To add platform Kafka to DI container, you should call AddPlatformKafka extension method on an IServiceCollection instance. This method accepts a delegate, that have to perform some configuration actions on a builder. This builder allows you to specify configuration, add consumers and producers.

collection.AddPlatformKafka(builder => builder
    .ConfigureOptions(configuration.GetSection("Presentation:Kafka")))

Configuration schema

{
  "Host": string,
  "SecurityProtocol": string enum,
  "SslCaPem": string,
  "SaslMechanism": string enum,
  "SaslUsername": string,
  "SaslPassword": string
}

• Host
  Bootstrap servers of Kafka cluster.
• SecurityProtocol
  Enum that specifies a security protocol used for connection to Kafka cluster. Allowed values: Plaintext (default), Ssl, SaslPlaintext, SaslSsl). This value determines whether some other configuration values are required or not.
• SslCaPem
  String representation of CA certificate. Required if SecurityProtocol is set to Ssl or SaslSsl.
• SaslMechanism
  Enum that specifies Sasl mechanism used for connection to Kafka cluster. Allowed values: Gssapi, Plain, ScramSha256, ScramSha512, OAuthBearer. Used if SecurityProtocol is set to SaslPlaintext or SaslSsl, but not required (default value = null).
• SaslUsername
  Username used to authorise when connecting to Kafka cluster. Required if SecurityProtocol is set to SaslPlaintext or SaslSsl.
• SaslPassword
  Password used to authorise when connecting to Kafka cluster. Required if SecurityProtocol is set to SaslPlaintext or SaslSsl.

Configuration example

{
  "Presentation": {
    "Kafka": {
      "Host": "localhost:8001",
      "SecurityProtocol": "SaslSsl",
      "SslCaPem": "...",
      "SaslMechanism": "ScramSha512",
      "SaslUsername": "username",
      "SaslPassword": "password"
    }
  }
}

Consumer

To register a consumer, you have to use same builder, provided in AddPlatformKafka extension method.

Call method AddConsumer in builder's call chain, after ConfigureOptions call. It accepts a delegate, used to configure consumer's builder.

You must configure message's key and value type, configuration, key and value deserialisers and message handler. All these methods can only be called in a strict order, which is shown in a snipped below. Method HandleWith<> accepts a class, implementing IKafkaConsumerHandler<,> as it's type parameter.

string group = Assembly.GetExecutingAssembly().GetName().Name ?? string.Empty;

collection.AddPlatformKafka(builder => builder
    .ConfigureOptions(...)
    .AddConsumer(b => b
        .WithKey<MessageKey>()
        .WithValue<MessageValue>()
        .WithConfiguration(configuration.GetSection("Presentation:Kafka:Consumers:MessageName"), c => c.WithGroup(group))
        .DeserializeKeyWithNewtonsoft()
        .DeserializeValueWithNewtonsoft()
        .HandleWith<MessageHandler>()));

You can configure custom deserialiser types (that have to implement IDeserializer<> interface) by using .DeserializeKeyWith<> and .DeserializeKeyWith<T>(IDeserializer<T>) (same for values). Library provides an extensions for using protobuf and Newtonsoft.Json deserialisers out of the box (.DeserializeKeyWithProto, .DeserializeKeyWithNewtonsoft)

WithConfiguration method can accept a delegate as a second parameter for additional configuration, it allows to configure consumer group and instance id for consumer via WithGroup and WithInstanceId methods respectively.

Configuration schema

{
    "IsDisabled": bool,
    "Topic": string,
    "Group": string,
    "InstanceId": string,
    "ParallelismDegree": int,
    "BufferSize": int,
    "BufferWaitLimit": string timespan,
    "ReadLatest": bool
}

• IsDisabled
  Specifies whether consumer is disabled. As consumer options are used as IOptionsMonitor<>, it will react in real time to this options being changed.
• Topic
  Topic name to be consumer. Required. Changing this option during application execution will lead to undefined behaviour.
• Group
  Consumer group name used when connecting to Kafka cluster. Not required, default is empty string.
• InstanceId
  Id of instance in specified consumer group. Not required, default is empty string.
• ParallelismDegree
  Number of threads that will handle consumed messages. Not required, default is 1, must be >= 1.
• BufferSize
  Consumer handler receives messages in batches, this parameter is used to configure maximum batch size. Not required, default is 1, must be >= 1.
• BufferWaitLimit
  To avoid consumer starvation when batching, you can configure maximum time waiting for a single batch accumulation. If the wait time for a singe batch exceeds this threshold, the batch will be yielded with items that were already collected. Not required when BufferSize = 1, otherwise must be > TimeSpan.Zero.
• ReadLatest
  Configures an offset selection strategy for consumer's first subscription to the topic. When true the offset would default to the latest message in the topic, when false the offset will be configured to the first message in the topic. Not required, default = false.

Configuration example

{
  "Presentation": {
    "Kafka": {
      "Consumers": {
        "MessageName": {
          "IsDisabled": false,
          "Topic": "my_topic",
          "ParallelismDegree": 2,
          "BufferSize": 100,
          "BufferWaitLimit": "00:00:01.500",
          "ReadLatest": false
        },
      },
    }
  }
}

In this example BufferWaitLimit is set to 1 second and 500 milliseconds.

Inbox

To use inbox with Kafka consumer, you have to end your configuration with HandleInboxWith<> call, instead of HandleWith<> and pass a type implementing IKafkaInboxHandler<,> to it's generic parameter.

collection.AddPlatformKafka(builder => builder
    .ConfigureOptions(...)
    .AddConsumer(b => b
        .WithKey<MessageKey>()
        .WithValue<MessageValue>()
        .WithConfiguration(...)
        .DeserializeKeyWithNewtonsoft()
        .DeserializeValueWithNewtonsoft()
        .HandleInboxWith<InboxMessageHandler>()));

Producer

Outbox