How to programatically create topic and send message to Kafka using C# dot net client

Question

I'm new to kafka and I want to try create topic and send message to kafka from my .net application. I'm using kafka.net dll and created topic successfully with this code:

        Uri uri = new Uri("http://localhost:9092");

        string topic = "testkafka";

        string payload = "test msg";

        var sendMsg = new Thread(() =>

        {

            KafkaNet.Protocol.Message msg = new KafkaNet.Protocol.Message(payload);

            var options = new KafkaOptions(uri);

            var router = new KafkaNet.BrokerRouter(options);

            var client = new Producer(router);

            client.SendMessageAsync(topic, new List<KafkaNet.Protocol.Message> { msg }).Wait();

        });

        sendMsg.Start();

but I can't see any message on:

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testkafka --from-beginning

anyone can help me with example? Thanks.

Answer 1

For both actions, you can make use of the cofluent-kafka-dotnet client.

---

In order to programatically create a topic:

static async Task CreateTopicAsync(string bootstrapServers, string topicName) {

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

---

In order to produce a message:

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

class Program
{
    public static async Task Main(string[] args)
    {
        var config = new ProducerConfig { BootstrapServers = "localhost:9092" };

        // If serializers are not specified, default serializers from
        // `Confluent.Kafka.Serializers` will be automatically used where
        // available. Note: by default strings are encoded as UTF8.
        using (var p = new ProducerBuilder<Null, string>(config).Build())
        {
            try
            {
                var dr = await p.ProduceAsync("myTopicName", new Message<Null, string> { Value="test" });
                Console.WriteLine($"Delivered '{dr.Value}' to '{dr.TopicPartitionOffset}'");
            }
            catch (ProduceException<Null, string> e)
            {
                Console.WriteLine($"Delivery failed: {e.Error.Reason}");
            }
        }
    }
}