Shuttle.Hopper.Kafka 21.0.2

Shuttle.Hopper.Kafka

Kafka implementation for Shuttle.Hopper.

Installation

dotnet add package Shuttle.Hopper.Kafka

Docker Compose

To get a local Kafka broker up and running using the docker-compose.yml file in the root of the repository:

docker compose up -d

Configuration

The URI structure is kafka://configuration-name/topic-name.

services.AddHopper()
    .UseKafka(builder =>
    {
        builder.Configure("local", options =>
        {
            options.BootstrapServers = "localhost:9092";
            options.ReplicationFactor = 1;
            options.NumPartitions = 1;
            options.MessageSendMaxRetries = 3;
            options.RetryBackoff = TimeSpan.FromSeconds(1);
            options.EnableAutoCommit = false;
            options.EnableAutoOffsetStore = false;
            options.FlushEnqueue = false;
            options.UseCancellationToken = true;
            options.ConsumeTimeout = TimeSpan.FromSeconds(30);
            options.OperationTimeout = TimeSpan.FromSeconds(30);
            options.ConnectionsMaxIdle = TimeSpan.Zero;
            options.Acks = Acks.All;
            options.EnableIdempotence = true;
        });
    });

The documentation for the Confluent.Kafka ConsumerConfig and ProducerConfig can be consulted for specific options. If there are any options that are not exposed via the KafkaOptions class, they can be set by providing the relevant configuration object:

services.AddHopper()
    .UseKafka(builder =>
    {
        builder.Configure("local", options =>
        {
            options.BootstrapServers = "localhost:9092";
            // ... other properties
            options.ConsumerConfig = new ConsumerConfig
            {
                // ... Confluent.Kafka specific options
            };
            options.ProducerConfig = new ProducerConfig
            {
                // ... Confluent.Kafka specific options
            };
        });
    });

Alternatively, you can use the ConsumerBuilder and ProducerBuilder properties to customize the Kafka clients:

services.AddHopper()
    .UseKafka(builder =>
    {
        builder.Configure("local", options =>
        {
            options.BootstrapServers = "localhost:9092";
            options.ConsumerBuilder = new ConsumerBuilder<Ignore, string>(new ConsumerConfig
            {
                // ... custom consumer configuration
            });
            options.ProducerBuilder = new ProducerBuilder<Null, string>(new ProducerConfig
            {
                // ... custom producer configuration
            });
        });
    });

The default JSON settings structure is as follows:

{
  "Shuttle": {
    "Kafka": {
      "local": {
        "BootstrapServers": "localhost:9092",
        "ReplicationFactor": 1,
        "NumPartitions": 1,
        "MessageSendMaxRetries": 3,
        "RetryBackoff": "00:00:01",
        "EnableAutoCommit": false,
        "EnableAutoOffsetStore": false,
        "FlushEnqueue": false,
        "UseCancellationToken": true,
        "ConsumeTimeout": "00:00:30",
        "OperationTimeout": "00:00:30",
        "ConnectionsMaxIdle": "00:00:00",
        "Acks": "All",
        "EnableIdempotence": true
      }
    }
  }
}

Options