SimpleKafkaLibrary 2.1.1

SimpleKafkaLibrary 2.1.1

Lightweight Kafka helper for .NET 8 (Confluent.Kafka). Public API matches 2.0 (ConsumeAsync, ConsumeRequestAsync, IMessageProducers, AddKafkaServices) so existing callers keep working.

What’s lighter in 2.1.x

ProduceAsync returns false when the broker is unreachable (within ProduceAsyncTimeoutMs). Callers (WCL/CCL) can fall through to SignalR/gRPC.

Package

<PackageReference Include="SimpleKafkaLibrary" Version="2.1.1" />

Configuration

Bind under a section such as "Kafka". KafkaConfig inherits Confluent ProducerConfig, so standard keys like BootstrapServers work in the same section.

"Kafka": {
  "BootstrapServers": "localhost:9092",
  "UseKafka": true,
  "UseEncryptedData": false,
  "EncryptedKey": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
  "FlushProducerInSeconds": 2,
  "ConsumedInSeconds": 2,
  "RequestReplyMinTimeoutMs": 100,
  "RequestReplyMaxTimeoutMs": 2000,
  "RequestReplyConsumedInMilliseconds": 250,
  "ProduceAsyncTimeoutMs": 1000,
  "ProducerRetryCount": 5,
  "ProducerRetryBackoffMs": 200,
  "MaxPollIntervalMs": 300000,
  "ConsumerQueuedMaxMessagesKbytes": 4096,
  "ConsumerFetchMaxBytes": 1048576,
  "ConsumerMaxPartitionFetchBytes": 1048576
}

EncryptedKey is required only when UseEncryptedData is true. Prefer secrets from configuration / a secret store — do not ship production keys in source.

DI registration

Producer only (host starts its own consumers)

services.AddKafkaServices(configuration.GetSection("Kafka"));

One consumer type (started as IHostedService)

services.AddKafkaServices<Consumer>(configuration.GetSection("Kafka"));

Multiple consumers

services.AddKafkaServices(cfg =>
{
    cfg.Configure(configuration.GetSection("Kafka"));
    cfg.RegisterConsumer<Consumer>();
    cfg.RegisterConsumer<UserConsumer>();
});

Or scan an assembly:

services.AddKafkaServices(cfg =>
{
    cfg.Configure(configuration.GetSection("Kafka"));
    cfg.RegisterConsumer(Assembly.GetExecutingAssembly());
});

Consumers start when the host starts (not during service registration). Broker outages are retried every few seconds and do not stop the process.

Producer usage

public class CreateUserCommandHandler
{
    private readonly IMessageProducers _messageProducers;

    public CreateUserCommandHandler(IMessageProducers messageProducers)
    {
        _messageProducers = messageProducers;
    }

    public async Task Handle(CancellationToken cancellationToken)
    {
        // Fire-and-forget (still uses the shared durable producer)
        await _messageProducers.WriteFireAndForget("User", newUser).ConfigureAwait(false);

        // true = broker accepted; false = Kafka disabled, down, or timed out
        var published = await _messageProducers.ProduceAsync("User", newUser).ConfigureAwait(false);

        // Optional partition key
        await _messageProducers.ProduceWithKeyAsync("User", userId.ToString("N"), newUser)
            .ConfigureAwait(false);
    }
}

Request-reply producer

Pair with ConsumeRequestAsync on the consumer. Wait time is clamped between RequestReplyMinTimeoutMs and RequestReplyMaxTimeoutMs.

var result = await _messageProducers.ProduceWaitForFResponseAsync<CreateUserRequest, CreateUserFeedback>(
    "User",
    request,
    cancellationToken: cancellationToken).ConfigureAwait(false);

if (result.IsSuccessful)
{
    var feedback = result.Data;
}

var fastResult = await _messageProducers.ProduceWaitForFResponseAsync<CreateUserRequest, CreateUserFeedback>(
    "User",
    request,
    responseTimeoutMs: 500,
    cancellationToken: cancellationToken).ConfigureAwait(false);

When Kafka is down, request-reply returns IsSuccessful = false (transaction rolled back). It does not crash the host.

Consumer usage

ConsumerBase requires ILoggerFactory (2.0+). Multiple ConsumeAsync / ConsumeRequestAsync calls on the same instance share one underlying Kafka client per mode.

public class Consumer : ConsumerBase
{
    public Consumer(
        KafkaConfig configuration,
        IMessageAdmin messageAdmin,
        ILoggerFactory loggerFactory)
        : base("my-service-group", configuration, messageAdmin, loggerFactory)
    {
    }

    public override async Task Invoke()
    {
        var task1 = ConsumeAsync<string>("testTopic", value =>
        {
            Console.WriteLine(value);
        });

        var task2 = ConsumeAsync<MyEvent>("testTopic2", evt =>
        {
            // handle evt
        });

        await Task.WhenAll(task1, task2);
        await base.Invoke();
    }
}

Offsets are committed only after the handler succeeds.

Request-reply consumer

public class UserConsumer : ConsumerBase
{
    public UserConsumer(
        KafkaConfig configuration,
        IMessageAdmin messageAdmin,
        ILoggerFactory loggerFactory)
        : base("user-group", configuration, messageAdmin, loggerFactory)
    {
    }

    public override async Task Invoke()
    {
        await ConsumeRequestAsync<CreateUserRequest, CreateUserFeedback>("User", request =>
        {
            if (string.IsNullOrWhiteSpace(request.Email))
            {
                return new CreateUserFeedback
                {
                    Success = false,
                    Message = "Email is required."
                };
            }

            return new CreateUserFeedback
            {
                Success = true,
                Message = "User created successfully.",
                UserId = Guid.NewGuid()
            };
        });

        await base.Invoke();
    }
}

Reply topic is {topic}-reply (created automatically when the broker is available).

Broker-down behavior (UseKafka: true)

Set UseKafka: false to skip all Kafka I/O without changing call sites.

Notes

• Use a distinct consumer groupId per service so each host gets its own copy of topic traffic.
• Prefer gRPC for sync RPC when available; Kafka request-reply is heavier (per-call reply consumer + transactional produce).
• You can still create multiple ConsumerBase subclasses; each instance gets its own shared async consumer and shared request-reply consumer.

License

This project is licensed with the MIT license.