Creating a Topic in Kafka on startup for a test

[KazysNoobiys]

I am using Aspire for local development and I would like to have the Kafka topics created immediately on startup, is there any way to do this?

[krisnashaypp]

If there's no aspire way you can probably use the Confluent AdminClientBuilder

        var config = new AdminClientConfig
        {
            BootstrapServers = KafkaConnectionString
        };

        using var adminClient = new AdminClientBuilder(config).Build();
        try
        {
            await adminClient.CreateTopicsAsync(new TopicSpecification[]
            {
                new TopicSpecification { Name = "my-topic-1", NumPartitions = 1, ReplicationFactor = 1 },
                new TopicSpecification { Name = "my-topic-2", NumPartitions = 1, ReplicationFactor = 1 }
            });
        }
        catch (CreateTopicsException e)
        {
            Console.WriteLine($"An error occurred creating topic: {e.Message}");
            throw;
        }

[davidfowl]

You can run code using the ResourceReadyEvent, this will fire after the kafka container is healthy:

https://learn.microsoft.com/en-us/dotnet/aspire/app-host/eventing#resource-eventing

Then as @krisnashaypp said, you can use the kafka client apis to connect and write code to seed topics. We're working on making this slightly better in 9.1 but the pattern is the same.

Note, this only works in development. Here's an example of doing it with mongodb:

var db = builder.AddMongoDB("mongo")
    .WithMongoExpress(c => c.WithHostPort(3022))
    .AddDatabase("db");

builder.Eventing.Subscribe<ResourceReadyEvent>(db.Resource, async (@event, ct) =>
{
    // Seed the database with some data
    var cs = await db.Resource.ConnectionStringExpression.GetValueAsync(ct);
    using var client = new MongoClient(cs);

    const string collectionName = "entries";

    var myDb = client.GetDatabase("db");
    await myDb.CreateCollectionAsync(collectionName, cancellationToken: ct);

    for (int i = 0; i < 10; i++)
    {
        await myDb.GetCollection<Entry>(collectionName).InsertOneAsync(new Entry(), cancellationToken: ct);
    }
});

[sunerievers]

Thanks - it worked flawlessly. FWIW, here is my complete code snippet in AppHost:

builder.Eventing.Subscribe<ResourceReadyEvent>(kafka.Resource, async (@event, ct) =>
{
    var cs = await kafka.Resource.ConnectionStringExpression.GetValueAsync(ct);

    var config = new AdminClientConfig
    {
        BootstrapServers = cs
    };

    using var adminClient = new AdminClientBuilder(config).Build();
    try
    {
        await adminClient.CreateTopicsAsync(new TopicSpecification[]
        {
                new TopicSpecification { Name = "my-topic-1", NumPartitions = 1, ReplicationFactor = 1 },
                new TopicSpecification { Name = "my-topic-2", NumPartitions = 1, ReplicationFactor = 1 }
        });
    }
    catch (CreateTopicsException e)
    {
        Console.WriteLine($"An error occurred creating topic: {e.Message}");
        throw;
    }
});