Let’s take a look at running a test instance (as single instance) of Kafka and write a producer and consumer application in C# to interact with it. As is my preference, we’ll use docker to run up our instance of Kafka and in my case this is running on an Ubuntu server.<\/p>\n
Kafka running in Docker<\/strong><\/p>\n We’ll start with the simplest docker compose file we can. So create the file docker-compose.yml and paste the following into it<\/p>\n This is very simple, it’s running a single instance of Kafka (which is only really likely to be something we’d use for testing). Kafka uses Zookeeper (although I believe that dependency may have gone or easy potentially going away), so we have Zookeeper running as well.<\/p>\n In the above file we’re setting the PLAINTEXT_HOST to the machine running the instance of Kafka, obvious this is not ideal so we can change this first to allow the environment to be supplied by either an environment variable of via a .env file. For this example let’s change that line to<\/p>\n just add a .env file in the same location as the docker compose file, and have something like this in it<\/p>\n Now we can run the Kakfa and Zookeeper up using<\/p>\n Remove the -d if you want to watch the log, which I would recommend to at least feel like things are running as expected. Also you can always run docker-compose ps to check that the services are running successfully<\/em><\/p>\n C# Producer<\/strong><\/p>\n We’ll create a console application that will simply send some messages to a topic, it’s our producer. Here’s my Producer.csproj<\/p>\n Whilst we can read configuration for Kafka from an INI file or the like’s but for simplicity we’ll handle these in code. So here’s a very basic sample of a producer (this is heavily based on the Confluent Kafka example)<\/p>\n In the above we’re creating a configuration, with reference to our bootstrap server with a unique client.id. We also need a topic which should be unique and will need to be known by the consumers who want to fetch events for a given topic.<\/p>\n In this example we create a batch of simple string key, string value<\/em> events and the build the producer object. Then we just randomly pick a ticker and assign a trade against it and send that event to Kakfa.<\/p>\n C# Consumer<\/strong><\/p>\n Obviously we’re going to want to fetch these events at some point. We do this via a consumer. Once events are added to Kafka (and depending upon it’s setup\/configuration) these event will “play” to a consumer that attaches to the correct topic. Once the events are received by the consumer they will not be replayed again, unless we explicitly force Kafka to do so.<\/p>\n Again this example is based heavily on the Confluent Kafka C# consumer. Create a Console application and replace the contents of the .csproj with the same csproj listed earlier for the Producer – this is just adding the relevant client package. Here’s the code for our Console based consumer<\/p>\n There’s a little more here than required, just to keep the consumer running and watching for events. In a service we ofcourse wouldn’t need half of this code.<\/p>\n Essentially we create a configurations which tells Kafka that consumer has a group.id (this is mandatory) and where we want the offset to reset to, for playing the events from. In other words, this example will connect to Kafka and only consume events it hasn’t already consumed. It will not replay events from the first to last.<\/p>\n If, and I’ve found it useful in some debugging situations, but it may be required in real world application, we wish to get ALL events, then we change the ConsumerBuilder line to the following<\/p>\n Multiple brokers<\/strong><\/p>\n A single Kafka broker is fine for testing, but Kafa was designed for multiple brokers, here’s a docker compose file that takes out single instance and add’s two more to create three Kafka brokers (I think this is often viewed as the minimal for production, but don’t quote me on that)<\/p>\n We’ve added the following, a KAFKA_BROKER_ID and the KAFKA_ADVERTISED_LISTENERS which references the newly added hostname. Just run up in docker-compose and the previous client code should work happily against this setup.<\/p>\n\r\nversion: '3'\r\nservices:\r\n zookeeper:\r\n image: confluentinc\/cp-zookeeper:latest\r\n environment:\r\n ZOOKEEPER_CLIENT_PORT: 2181\r\n ZOOKEEPER_TICK_TIME: 2000\r\n\r\n kafka:\r\n image: confluentinc\/cp-kafka:latest\r\n depends_on:\r\n - zookeeper\r\n ports:\r\n - 29092:29092\r\n environment:\r\n KAFKA_BROKER_ID: 1\r\n KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181\r\n KAFKA_ADVERTISED_LISTENERS: PLAINTEXT:\/\/kafka:9092,PLAINTEXT_HOST:\/\/192.168.0.1:29092\r\n KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT\r\n KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT\r\n KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1\r\n<\/pre>\n
\r\nKAFKA_ADVERTISED_LISTENERS: PLAINTEXT:\/\/kafka:9092,PLAINTEXT_HOST:\/\/${HOST}:29092\r\n<\/pre>\n\r\nHOST=192.168.0.1\r\n<\/pre>\n
\r\ndocker-compose up -d\r\n<\/pre>\n
\r\n<Project Sdk="Microsoft.NET.Sdk">\r\n\r\n <PropertyGroup>\r\n <OutputType>Exe<\/OutputType>\r\n <TargetFramework>net8.0<\/TargetFramework>\r\n <ImplicitUsings>enable<\/ImplicitUsings>\r\n <Nullable>enable<\/Nullable>\r\n <\/PropertyGroup>\r\n\r\n <ItemGroup>\r\n <PackageReference Include="Confluent.Kafka" Version="2.3.0" \/>\r\n <\/ItemGroup>\r\n\r\n<\/Project>\r\n<\/pre>\n
\r\nusing Confluent.Kafka;\r\n\r\nvar config = new List<KeyValuePair<string, string>>\r\n{\r\n new("bootstrap.servers", "192.168.0.1:19092"),\r\n new("client.id", "my-producer")\r\n};\r\n\r\nconst string topic = "my-topic";\r\n\r\nstring[] tickers = { "AAPL", "GOOGL", "MSFT", "AMZN", "META", "TSLA", "GS" };\r\nstring[] trades = { "Buy 100", "Sell 1000", "Buy 9090", "Sell 45", "Buy 900000", "Sell 123", "Buy 8901" };\r\n\r\nusing var producer = new ProducerBuilder<string, string>(config).Build();\r\n\r\nvar rnd = new Random();\r\n\r\nfor (var i = 0; i < 10; ++i)\r\n{\r\n var ticker = tickers[rnd.Next(tickers.Length)];\r\n var trade = trades[rnd.Next(trades.Length)];\r\n\r\n producer.Produce(topic, new Message<string, string> { Key = ticker, Value = trade },\r\n deliveryReport =>\r\n {\r\n if (deliveryReport.Error.Code != ErrorCode.NoError)\r\n {\r\n Console.WriteLine($"Error sending event: {deliveryReport.Error.Reason}");\r\n }\r\n else\r\n {\r\n Console.WriteLine($"Sent event topic = {topic}: key = {ticker} value = {trade}");\r\n }\r\n });\r\n}\r\n\r\nproducer.Flush(TimeSpan.FromSeconds(10));\r\n<\/pre>\n\r\nusing Confluent.Kafka;\r\n\r\nvar config = new List<KeyValuePair<string, string>>\r\n {\r\n new("bootstrap.servers", "192.168.0.1:19092"),\r\n new("group.id", "my-group"),\r\n new("auto.offset.reset", "earliest")\r\n };\r\n\r\nconst string topic = "my-topic";\r\n\r\nvar cts = new CancellationTokenSource();\r\nConsole.CancelKeyPress += (_, e) =>\r\n{\r\n e.Cancel = true; \/\/ prevent the process from terminating.\r\n cts.Cancel();\r\n};\r\n\r\nusing var consumer = new ConsumerBuilder<string, string>(config).Build();\r\n\r\nconsumer.Subscribe(topic);\r\ntry\r\n{\r\n while (true)\r\n {\r\n var cr = consumer.Consume(cts.Token);\r\n Console.WriteLine($"Consumed event, topic {topic}: key = {cr.Message.Key} value = {cr.Message.Value}");\r\n }\r\n}\r\ncatch (OperationCanceledException)\r\n{\r\n \/\/ Ctrl-C was pressed.\r\n}\r\nfinally\r\n{\r\n consumer.Close();\r\n}\r\n<\/pre>\n\r\nusing var consumer = new ConsumerBuilder<string, string>(config)\r\n .SetPartitionsAssignedHandler((c, partitions) =>\r\n {\r\n \/\/ reset the offsets for this client\r\n var offsets = partitions.Select(tp => new TopicPartitionOffset(tp, Offset.Beginning));\r\n return offsets;\r\n })\r\n .Build();\r\n<\/pre>\n\r\nversion: '3'\r\nservices:\r\n zookeeper:\r\n image: confluentinc\/cp-zookeeper:latest\r\n environment:\r\n ZOOKEEPER_CLIENT_PORT: 2181\r\n ZOOKEEPER_TICK_TIME: 2000\r\n\r\n kafka-broker1:\r\n image: confluentinc\/cp-kafka:latest\r\n hostname: kafka-broker1\r\n depends_on:\r\n - zookeeper\r\n ports:\r\n - 19092:19092\r\n environment:\r\n KAFKA_BROKER_ID: 1\r\n KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181\r\n KAFKA_ADVERTISED_LISTENERS: PLAINTEXT:\/\/kafka-broker1:9092,PLAINTEXT_HOST:\/\/${HOST}:19092\r\n KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT\r\n KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT\r\n KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3\r\n kafka-broker2:\r\n image: confluentinc\/cp-kafka:latest\r\n hostname: kafka-broker2\r\n depends_on:\r\n - zookeeper\r\n ports:\r\n - 29092:29092\r\n environment:\r\n KAFKA_BROKER_ID: 2\r\n KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181\r\n KAFKA_ADVERTISED_LISTENERS: PLAINTEXT:\/\/kafka-broker2:9092,PLAINTEXT_HOST:\/\/${HOST}:29092\r\n KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT\r\n KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT\r\n KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3\r\n kafka-broker3:\r\n image: confluentinc\/cp-kafka:latest\r\n hostname: kafka-broker3\r\n depends_on:\r\n - zookeeper\r\n ports:\r\n - 39092:39092\r\n environment:\r\n KAFKA_BROKER_ID: 3\r\n KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181\r\n KAFKA_ADVERTISED_LISTENERS: PLAINTEXT:\/\/kafka-broker3:9092,PLAINTEXT_HOST:\/\/${HOST}:39092\r\n KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT\r\n KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT\r\n KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3\r\n<\/pre>\n