Kafka.TestBroker 1.0.0

Kafka.TestBroker

An in-process fake Kafka broker for integration tests and local development. No Docker, no Zookeeper, no external infrastructure — just add the package and go.

Installation

dotnet add package Kafka.TestBroker

Quick start

using Kafka.TestFramework;

var settings = new BrokerSettings
{
    Host = "127.0.0.1",
    Port = 0,   // 0 = random free port
    Topics = new List<TopicSettings>
    {
        new TopicSettings { Name = "orders", Partitions = 1 }
    }
};

await using var broker = new TestBroker(settings);
await broker.StartAsync();

Console.WriteLine(broker.BootstrapServers); // e.g. "127.0.0.1:54321"

Pass broker.BootstrapServers to your Confluent.Kafka producer or consumer — they talk to the fake broker exactly as they would to a real one.

Usage in tests

[Test]
public async Task Order_Is_Processed(CancellationToken cancellationToken)
{
    var settings = new BrokerSettings
    {
        Host = "127.0.0.1",
        Port = 0,
        Topics = new List<TopicSettings>
        {
            new TopicSettings { Name = "orders", Partitions = 1 }
        }
    };

    await using var broker = new TestBroker(settings);
    await broker.StartAsync();

    // Produce
    using var producer = new ProducerBuilder<string, string>(new ProducerConfig
    {
        BootstrapServers = broker.BootstrapServers
    }).Build();

    await producer.ProduceAsync("orders", new Message<string, string>
    {
        Key = "order-1",
        Value = """{ "id": 1, "item": "widget" }"""
    }, cancellationToken);
    producer.Flush(cancellationToken);

    // Consume
    using var consumer = new ConsumerBuilder<string, string>(new ConsumerConfig
    {
        BootstrapServers = broker.BootstrapServers,
        GroupId           = "test-group",
        AutoOffsetReset   = AutoOffsetReset.Earliest,
        EnableAutoCommit  = false
    }).Build();

    consumer.Subscribe("orders");

    var result = consumer.Consume(TimeSpan.FromSeconds(5));

    Assert.Equal("order-1", result.Message.Key);
    Assert.Contains("widget", result.Message.Value);
}

BrokerSettings reference

broker.BootstrapServers always returns the actual host:port string the socket is bound to.

Multiple topics and partitions

var settings = new BrokerSettings
{
    Host = "127.0.0.1",
    Port = 0,
    Topics = new List<TopicSettings>
    {
        new TopicSettings { Name = "events",  Partitions = 3 },
        new TopicSettings { Name = "metrics", Partitions = 1 }
    }
};

Error injection

Simulate broker failures to test retry / error handling paths:

// Make partition 0 of "orders" return LEADER_NOT_AVAILABLE (error 5)
broker.SetPartitionError("orders", partition: 0, errorCode: 5);

// ... run your code that should handle the error ...

// Restore normal operation
broker.ClearPartitionError("orders", partition: 0);

Common error codes:

Use in a console app or hosted service

TestBroker has no ASP.NET Core dependency — it can run anywhere:

// Plain console app
var settings = new BrokerSettings
{
    Host = "127.0.0.1",
    Port = 19093,
    Topics = new List<TopicSettings>
    {
        new TopicSettings { Name = "my-topic", Partitions = 1 }
    }
};

await using var broker = new TestBroker(settings);
await broker.StartAsync();

Console.WriteLine($"Broker listening on {broker.BootstrapServers}");
Console.WriteLine("Press Ctrl+C to stop...");
await Task.Delay(Timeout.Infinite);

Or register it in a hosted service:

// Program.cs (ASP.NET Core / Generic Host)
builder.Services.AddSingleton<TestBroker>(sp =>
{
    var settings = sp.GetRequiredService<IConfiguration>()
        .GetSection("BrokerSettings")
        .Get<BrokerSettings>() ?? new BrokerSettings();
    return new TestBroker(settings);
});
builder.Services.AddHostedService<BrokerHostedService>();

What the fake broker supports

Tested with Confluent.Kafka 2.x.

License

MIT