How to create a Kafka Topic using Confluent.Kafka .Net Client

Question

It seems like most popular .net client for Kafka (https://github.com/confluentinc/confluent-kafka-dotnet) is missing methods to setup and create Topics. When calling Producer.ProduceAsync() the topic is created automatically but I can't find a way to setup partitions, retention policy and other settings.

I tried to find any examples online but all I found just use defaults.

Maybe there is another .net client that I can use instead?

Josephina

"ProduceAsync" only creates a topic automatically if auto topic creation is enabled at the broker (which in most cases, it shouldn't be) – Diatessaron

Answer 1

It is now available in latest release of Confluent.Kafka .Net client library.

See: https://github.com/confluentinc/confluent-kafka-dotnet/blob/b7b04fed82762c67c2841d7481eae59dee3e4e20/examples/AdminClient/Program.cs

        using (var adminClient = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrapServers }).Build())
        {
            try
            {
                await adminClient.CreateTopicsAsync(new TopicSpecification[] { 
                    new TopicSpecification { Name = topicName, ReplicationFactor = 1, NumPartitions = 1 } });
            }
            catch (CreateTopicsException e)
            {
                Console.WriteLine($"An error occured creating topic {e.Results[0].Topic}: {e.Results[0].Error.Reason}");
            }
        }

Reliance

Answer 2

Confluent yet not provide any APIs to create topic from dot net client, however there is a workaround for the same.

1. Set auto.create.topics.enable = true in kafka configuration

2. use var brokerMetadata = producer.GetMetadata(false, topicName); to query available topics in existing brokers, if specified topic is not available then kafka will create a topic with specified name.

    private static bool CreateTopicIfNotExist(Producer producer, string topicName)
    {
        bool isTopicExist = producer.GetMetadata().Topics.Any(t => t.Topic == topicName);
        if (!isTopicExist)
        {
            //Creates topic if it is not exist; Only in case of auto.create.topics.enable = true is set into kafka configuration
            var topicMetadata = producer.GetMetadata(false, topicName).Topics.FirstOrDefault();
            if (topicMetadata != null && (topicMetadata.Error.Code != ErrorCode.UnknownTopicOrPart || topicMetadata.Error.Code == ErrorCode.Local_UnknownTopic))
                isTopicExist = true;
        }
        return isTopicExist;
    }

Thus you can use this work around, I know this is dirty solution but it seems there is no other way as of now.

Holguin

The problem that I am facing is still exists: the topic will be created with all default settings. Simple call to Producer.ProduceAsync() also creates a topic if not exists – Josephina

Answer needs to be updated with the correct api, or clarify this will use default topic settings – Diatessaron

Answer 3

Confluent.Kafka.AdminClient is available in version 1.0.0-experimental-2 but doesn't allow creating topics etc.

Its built on librdkafka which doesn't have APIs for this yet.

So for now you have to configure this on the broker using e.g. bin\windows\kafka-topics.sh --create ...

Paramagnetic