I Just Want to See It Work: Kafka Stream, C# Producer and Consumer

When learning a new technology, sometimes I just want to see it work. It gives me a baseline to extend my ideas, to see what is possible, and to imagine what it can become. While reading through documentation or following tutorials, I am often greeted with numerous options and configurations. After seeing all the dazzling capabilities, I always want to see the core, the fundamental. Didn’t you ever just want to experience a simple working copy? Hence, I just want to see it work. This series aims at minimizing the possibility of having a missing link and encourages you to build your next innovative solution based on what you learned here. What is Kafka Stream? According to , Kafka Streams is a client library for building applications and where the input and output data are stored in Kafka clusters. Kafka microservices, What do we want to achieve? Run Kafka with KRaft in docker. Run a producer on your local machine (not in docker) that publishes messages to a topic. Run a consumer on your local machine (not in docker) that pulls a message whenever a message is published in a topic.] Resources You can find the source code at my GitHub repository Docker image: Bitnami Kafka Run Kafka in Docker Navigate to where the docker-compose.yml is. Run the below command in cmd to start a docker container of Kafka in the background. docker-compose up -d Docker-compose.yml version: "3" services: kafka: image: "bitnami/kafka:3.2.3" hostname: kafka ports: - "9092:9092" environment: - ALLOW_PLAINTEXT_LISTENER=yes - KAFKA_ENABLE_KRAFT=yes - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 It controls whether the Kafka broker allows connections over plaintext listeners. A plaintext listener allows connections over an unencrypted and non-authenticated channel. It means the data exchanged between the client and the Kafka broker is sent in plain text. ALLOW_PLAINTEXT_LISTENER: Enable KRaft mode. KAFKA_ENABLE_KRAFT: The hostname a client should use to connect to the Kafka broker. You need this for both producer and consumer to communicate with Kafka. KAFKA_CFG_ADVERTISED_LISTENERS: Let’s publish a message to Kafka Navigate to where is at. Start a Powershell and run the following command. KafkaStream.Producer.csproj dotnet run When you see the message: “Message delivered to input-topic”, that means you did it! What is in the Producer? We first setup the configuration for the producer. Build a message Produce the message var config = new ProducerConfig { BootstrapServers = "localhost:9092", ClientId = "producer-1" }; using (var producer = new ProducerBuilder (config).Build()) { var message = new Message { Key = null, // Set the key if you want to partition the messages Value = "Hello, Kafka!" }; try { var deliveryResult = await producer .ProduceAsync("input-topic", message); Console.WriteLine($"Message delivered to {deliveryResult.TopicPartitionOffset}"); } catch (ProduceException ex) { Console.WriteLine($"Delivery failed: {ex.Error.Reason}"); } } What is bootstrap-server? It serves as an entry point for a client to connect to a Kafka broker. Let’s setup the consumer Navigate to where KafkaStream.Consumer.csproj is at. Start a Powershell and run the following command. dotnet run The initialization will take a few seconds. Then you should see the following messages. Consuming messages from topic: input-topic Press any key to exit Received message: Hello, Kafka! from partition [0] offset 0 There you go! What is in the Consumer? The main thread sets up the Kafka consumer and starts a background thread to handle message consumption. In the background thread, the code runs in a continuous loop, until a cancellation is requested. A cancellation can be requested by pressing any key while the program is running. const string Topic = "input-topic"; // Configure the consumer var config = new ConsumerConfig { BootstrapServers = "localhost:9092", GroupId = "consumer-group-1", AutoOffsetReset = AutoOffsetReset.Earliest, EnableAutoCommit = false // Disable auto commit to have more control over offsets }; using (var consumer = new ConsumerBuilder (config).Build()) { // Subscribe to the topic consumer.Subscribe(Topic); Console.WriteLine($"Consuming messages from topic: {Topic}"); // Start consuming messages in a background thread var cancellationTokenSource = new CancellationTokenSource(); var cancellationToken = cancellationTokenSource.Token; var consumerThread = new Thread(() => { try { while (true) { try { var consumeResult = consumer.Consume(cancellationToken); Console.WriteLine($"Received message: {consumeResult.Message.Value} from partition {consumeResult.Partition} offset {consumeResult.Offset}"; // Manually commit the offset to mark the message as consumed consumer.Commit(consumeResult); } catch (ConsumeException ex) { Console.WriteLine($"Error occurred: {ex.Error.Reason}"); } } } catch (OperationCanceledException) { // This exception will be thrown when cancellation is requested. Console.WriteLine("Cancellation requested"); } finally { consumer.Close(); } }); // Start the consumer thread consumerThread.Start(); // Wait for a key press to exit Console.WriteLine("Press any key to exit"); Console.ReadKey(); // Request cancellation and wait for the consumer thread to stop cancellationTokenSource.Cancel(); consumerThread.Join(); Console.WriteLine("End Consumer"); } What does offset do? In this context, it tells the consumer where it left off. Gotcha While developing your own producer/consumer, if you get an error message indicating the hostname cannot be resolved, check KAFKA_CFG_ADVERTISED_LISTENERS. If you still run into the same issue, consider updating the host file. Conclusion In this article, we successfully run Kafka in a docker container, produce a message to a topic, and consume a message from the topic. There are limitless applications of using Kafka. Go ahead and play with the project, and build something interesting of your own!