Embedded Kafka for integration tests

[tanoshii]

trafficstars

Description

Hi! I want to do integration tests in my project which uses Kafka Consumers and Producers. I have similar tests in Java in which I use an EmbeddedKafka provided by Spring, that allows me to test without having Kafka running previously.

I saw the integration tests in this repository and you use a real running Kafka to execute them. Do you have or have in mind to have something similar to an Embedded or In-Memory Kafka?

Thank you!

How to reproduce

Run integration tests without a running Kafka.

Checklist

Please provide the following information:

• [ ] A complete (i.e. we can run it), minimal program demonstrating the problem. No need to supply a project file.
• [x] Confluent.Kafka nuget version: 1.1.0
• [ ] Apache Kafka version.
• [ ] Client configuration.
• [x] Operating system: Ubuntu
• [ ] Provide logs (with "debug" : "..." as necessary in configuration).
• [ ] Provide broker log excerpts.
• [ ] Critical issue.

[larsbrekken]

FYI we do this with Docker and this image: https://github.com/wurstmeister/kafka-docker

[mhowlett]

We don't have specific plans, but we do plan to put quite a bit more effort into the integration tests. Something that's much more important to this client than others is we'd like a good story for testing on windows, and running Kafka on Windows is problematic (even under docker). Currently, I use the following to bring up services for testing (on all platforms): https://github.com/confluentinc/confluent-kafka-dotnet/pull/936

[larsbrekken]

Out of interest, what issues have you had running Kafka on Windows with Docker? We've had good experience running the image above plus the Zookeeper one with docker-compose, including running it over SSL. While it's not something I'd run in a production setting, it's been pretty stable for our integration tests.

[tanoshii]

We wanted to do integration tests during build time with minimum number of external dependencies, so it can be quickiest as possible. If not, Docker can be our next approach for integration tests.

Our final environment will be on linux, but I agree, not only Kafka but Docker on Windows is also problematic.

[Makashima]

Has there been any further thoughts on implementing a embedded Kafka for integration testing? The ability to run this level of testing at build time without docker dependence would be a huge advantage.

[edenhill]

The client has a built-in mock broker that can be utilized for rudimentary application testing. In its simplest form it is enabled by setting the TestMockNumBrokers configuration property (to e.g. 3), which will replace your configured bootstrap brokers with an internal mock cluster. This mode is typically only suitable for producers since consumers will not have any messages to consume. If you want to test a consumer you will need set up a mock cluster and first produce to it so you can then consume those messages. This is also quite straight forward, all you need is set up a client instance with TestMockNumBrokers=3 and then extract the mock bootstrap servers, but that latter API is currently not exposed in the dotnet client.

There's a rich C API for more advanced use-cases, including error injection, latency control, etc: https://github.com/edenhill/librdkafka/blob/master/src/rdkafka_mock.h

[skovalyova]

Hi, are there any updates since 2019?

[Makashima]

I found adding the setting test.mock.num.brokers=3 was more than enough to create the required mock instance. However the instance always starts on a different port. It would be useful if bootstrap.server was provided when the mock is created to use that static port instead

[edenhill]

I recommend exposing the rd_kafka_mock_cluster_bootstrap() method in public .NET API so that the mock bootstrap.servers can be retrieved.

[mac2000]

Workaround: create admin client first, get cluster metadata, from it grab actual bootstrap servers list, e.g.:

using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Confluent.Kafka;
using Xunit;

namespace UnitTests;

public class Demo
{
    [Fact]
    public async Task Should()
    {
        // Trick: we are going to connect with admin client first so we can get metadata and bootstrap servers from it
        var clientConfig = new ClientConfig();
        clientConfig.Set("bootstrap.servers", "localhost:9200");
        clientConfig.Set("test.mock.num.brokers", "3");
        var adminClient = new AdminClientBuilder(clientConfig).Build();
        var metadata = adminClient.GetMetadata(TimeSpan.FromSeconds(1));
        var bootstrapServers = string.Join(",", metadata.Brokers.Select(b => $"{b.Host}:{b.Port}"));

        // Demo - produce some message to our mock cluster
        var producer = new ProducerBuilder<Null, string>(new ProducerConfig { BootstrapServers = bootstrapServers }).Build();
        var report = await producer.ProduceAsync("ping", new Message<Null, string> { Value = "pong" });
        Assert.Equal(PersistenceStatus.Persisted, report.Status);
        Assert.True(report.Partition >= 0 && report.Offset >= 0);

        // Demo - consume it, be sure to wait a little bit or add some kind of loop with timeout
        var consumer = new ConsumerBuilder<Ignore, string>(new ConsumerConfig { BootstrapServers = bootstrapServers, GroupId = "demo", AutoOffsetReset = AutoOffsetReset.Earliest }).Build();
        consumer.Subscribe("ping");
        var result = consumer.Consume(TimeSpan.FromSeconds(10));
        Assert.Equal("pong", result.Message.Value);
    }
}