Home Software Development Kafka Getting Started with Landoop’s Kafka on Docker Locally

Getting Started with Landoop’s Kafka on Docker Locally

July 29, 2018

1646

Reading Time: 5 minutes

Getting Started with Landoop’s Kafka on Docker for Windows

Here’s a quick guide to running Kafka on Windows with Docker. I’ll show you how to pull Landoop’s Kafka image from Docker Hub, run it, and how you can get started with Kafka. We’ll also be building a .NET Core C# console app for this demonstration. I learned most of this from the Kafka Udemy videos mentioned below and examples from the Confluent GitHub. This article is for people who are brand new to Kafka and want to work with .NET Core.

Source Code: Kafka C# Sample

UPDATES: I last updated this on 6/22/2020. I upgraded the code to the latest Confluent Kafka APIs.

Requirements

Make sure you installed and are familiar with Docker for Windows, ConEmu, Visual Studio and C#.

Docker for Windows uses the localhost address of 127.0.0.1 where Docker Toolbox / Docker Machine will use 192.168.99.100. Landoop’s fast-data-dev Github readme currently displays the steps for docker-machine.

Setup Kafka with Docker

After you’ve installed Docker, you will need to pull down the Landoop Kafka (fast-data-dev) image from Docker Hub. The commands below will pull the image.

docker pull landoop/fast-data-dev

Running Landoop’s Kafka Container in Docker

We will start this container in a detached state (“-d”) and then we will access the logs so we can see what’s happening.

docker run -d --name landoopkafka -p 2181:2181 -p 3030:3030 -p 8081:8081 -p 8082:8082 -p 8083:8083 -p 9092:9092 -e ADV_HOST=127.0.0.1 landoop/fast-data-dev

To see the logs of the container and have them persist to the screen pass the docker attribute “-f” for follow. This screen is important for troubleshooting errors. Some common errors you may see are memory issues. The URL for the Kafka UI is also in the logs.

docker logs -f landoopkafka

Landoop’s Kafka UI

Once the Docker image for fast-data-dev is running you will be able to access theLandoop’s Kafka UI. This gives developers the ability to see in real-time what Kafka is doing, how it creates and manages topics. You can visually see configuration and topic data in the UI.

Landoop’s Kafka UI: http://127.0.0.1:3030/

Working with Kafka via Command Line

It will benefit you to learn how Kafka works from a command-line basis. Users can create topics, publish and subscribe; all from the bash terminal. This is a great starting place because you can see exactly how it works under the hood.

Accessing Docker Container

Executing the command below will connect you to the container with bash. Doing so will allow you to interact with Kafka via command line.

docker exec -it landoopkafka bash

Kafka CLI Producer

We’re going to create our first topic, “first_topic” and push a single string of data to it.

If you receive an error message something similar to ” Error: {new_topic={LEADER_NOT_AVAILABLE}”, this is normal because it takes time to create the topic.

kafka-console-producer --topic first_topic --broker-list 127.0.0.1:9092

hello from producer

Kafka CLI Consumer

The consumer will immediately listen for publishes from the producer.

kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic first_topic

However, if you want to pull all of the data from the beginning you can run this command:

kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic first_topic --from-beginning

Confluent Kafka with DotNet Core

The source code for this application is maintained in GitHub. This demo is very simple, we’ll use Kafka as a pub/sub for messages. We’ll serialize a model class, Msg, and publish it to Kafka and allow the consumer to use C# to read that data.

Source Code: Kafka C# Sample
Confluent Kafka: https://github.com/confluentinc/confluent-kafka-dotnet

Message Data Model

This data model will be shared between stored in its own project (Kafka.Data) and shared between the Producer and the Consumer.

Kafka C# Producer

namespace Kafka.Producer
{
    class Program
    {
        static async Task Main(string[] args)
        {
            try
            {
                // banner
                Console.WriteLine("Kafka Producer Sample.");
                Console.WriteLine("Press CTRL+C to exit");

                var config = new ProducerConfig
                {
                    BootstrapServers = "127.0.0.1:9092",
                    ClientId = Dns.GetHostName(),
                };

                var cancelled = false;
                Console.CancelKeyPress += (_, e) => {
                    e.Cancel = true; // prevent the process from terminating.
                    cancelled = true;
                };

                while (!cancelled)
                {
                    // get name
                    Console.WriteLine("\nWhat is your name?");
                    var name = Console.ReadLine();

                    // get message
                    Console.WriteLine("\nWhat is your message?");
                    var message = Console.ReadLine();

                    // set time stamp
                    var msgTimeStamp = DateTime.Now;

                    Msg msg = new Msg();
                    msg.User = name;
                    msg.Message = message;
                    msg.TimeStamp = msgTimeStamp;

                    using (var producer = new ProducerBuilder<Null, string>(config).Build())
                    {
                        // convert the Msg model to json
                        var jsonMsg = JsonConvert.SerializeObject(msg);

                        // push to Kafka
                        await producer.ProduceAsync("topic_messages", new Message<Null, string> { Value = jsonMsg });
                    }

                    // produce message to Kafka
                    Console.WriteLine("Message published to Kafka.");
                    Console.WriteLine("\n");
                }
            }
            catch (System.Exception ex)
            {
                Console.WriteLine(ex.ToString());
            }
        }
    }
}

namespace Kafka.Producer

{

class Program

{

static async Task Main(string[] args)

{

try

{

// banner

Console.WriteLine("Kafka Producer Sample.");

Console.WriteLine("Press CTRL+C to exit");

var config = new ProducerConfig

{

BootstrapServers = "127.0.0.1:9092",

ClientId = Dns.GetHostName(),

};

var cancelled = false;

Console.CancelKeyPress += (_, e) => {

e.Cancel = true; // prevent the process from terminating.

cancelled = true;

};

while (!cancelled)

{

// get name

Console.WriteLine("\nWhat is your name?");

var name = Console.ReadLine();

// get message

Console.WriteLine("\nWhat is your message?");

var message = Console.ReadLine();

// set time stamp

var msgTimeStamp = DateTime.Now;

Msg msg = new Msg();

msg.User = name;

msg.Message = message;

msg.TimeStamp = msgTimeStamp;

using (var producer = new ProducerBuilder<Null, string>(config).Build())

{

// convert the Msg model to json

var jsonMsg = JsonConvert.SerializeObject(msg);

// push to Kafka

await producer.ProduceAsync("topic_messages", new Message<Null, string> { Value = jsonMsg });

}

// produce message to Kafka

Console.WriteLine("Message published to Kafka.");

Console.WriteLine("\n");

}

catch (System.Exception ex)

{

Console.WriteLine(ex.ToString());

}

Kafka C# Consumer

This is a sample of how to read the data from the Kafka stream.

namespace Kafka.Consumer
{
    class Program
    {
        static async Task Main(string[] args)
        {
            try
            {
                var config = new ConsumerConfig
                {
                    BootstrapServers = "127.0.0.1:9092",
                    // Disable auto-committing of offsets.
                    EnableAutoCommit = false,
                    GroupId = "consumergroup1"
                };

                using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build())
                {
                    CancellationTokenSource cancellationToken = new CancellationTokenSource();

                    consumer.Subscribe("topic_messages");

                    bool cancelled = false;

                    Console.CancelKeyPress += (_, e) => {
                        e.Cancel = true;
                        cancelled = true;
                        cancellationToken.Cancel();
                    };

                    while (!cancelled)
                    {
                        var consumeResult = consumer.Consume(cancellationToken.Token);
                        Console.WriteLine($"Consumed message '{consumeResult.Value}' at: '{consumeResult.TopicPartitionOffset}'.");
                    }

                    consumer.Close();
                }
            }
            catch (System.Exception ex)
            {
                Console.WriteLine(ex.ToString());
            }
        }
    }
}

namespace Kafka.Consumer

{

class Program

{

static async Task Main(string[] args)

{

try

{

var config = new ConsumerConfig

{

BootstrapServers = "127.0.0.1:9092",

// Disable auto-committing of offsets.

EnableAutoCommit = false,

GroupId = "consumergroup1"

};

using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build())

{

CancellationTokenSource cancellationToken = new CancellationTokenSource();

consumer.Subscribe("topic_messages");

bool cancelled = false;

Console.CancelKeyPress += (_, e) => {

e.Cancel = true;

cancelled = true;

cancellationToken.Cancel();

};

while (!cancelled)

{

var consumeResult = consumer.Consume(cancellationToken.Token);

Console.WriteLine($"Consumed message '{consumeResult.Value}' at: '{consumeResult.TopicPartitionOffset}'.");

}

consumer.Close();

}

catch (System.Exception ex)

{

Console.WriteLine(ex.ToString());

}

Kafka Tool

Also worth mentioning because it will help you inspect the data in the log of the Kafka Topics is a tool called “Kafka Tool”; available on Windows, Mac, and Linux.

https://www.kafkatool.com/

Using this tool you can browse the Kafka cluster and explore the topics.

Troubleshooting

Some people have issues with getting this up and running because they aren’t famililar with the surrounding technologies like Docker.

Originally I used ConEmu but PowerShell works better. I’d also recommend tricking out your PowerShell with Oh My Posh!

I encountered issues with ConEmu and this helped me…
winpty – if you are running Windows and run into an issue prepend the command with winpty so it reads “winpty docker run -d….”

Closing

We recently started using Kafka at my current job and I thought it would be beneficial to share a quick tutorial of how to use C#, Confluent Kafka and Landoop’s fast-data-dev. I couldn’t find a quick all-in-one tutorial for the basics so I created one.

Confluent Kafka Git Repo

Confluent's Apache Kafka .NET client
https://github.com/confluentinc/confluent-kafka-dotnet
490 forks.
1,584 stars.
281 open issues.
Recent commits:

Merge branch '1.4.x', Matt Howlett
-> v1.4.4 (#1339), GitHub
-> v1.4.4-RC1 (#1327), GitHub
Test well known protobuf schema refs (#1311), GitHub
Fix infinite recursion bug in error_destroy (#1312)* Fix infinite recursion bug in error_destroy* Test that includes error_t marshaling in code path, GitHub

Udemy Videos

If you’d like to learn more about Kafka I highly recommend watching these videos by Stephane Maarek at Udemy.com

https://www.udemy.com/apache-kafka/
https://www.udemy.com/kafka-streams/
https://www.udemy.com/kafka-cluster-setup/
https://www.udemy.com/kafka-connect/
https://www.udemy.com/confluent-schema-registry/
https://www.udemy.com/apache-kafka-security/