Kafka Consumer не принимает сообщения. C#
Скачал с официального сайта Apache kafka. Так же установил Java и прописал все пути в среде пользователя. Запускаю сначала zookeper из каталога скаченного файла с оф. сайта bin\windows\zookeeper-server-start.bat config\zookeeper.properties.
Дальше запускаю с этого же каталога уже сервер кафки
bin\windows\kafka-server-start.bat config\server.properties

Как я понял все вроде запустилось иначе мой код вообще бы давал ексепшен при старте, но Producer создает канал и начинает отправлять сообщения. Сообщения идут, а вывода у получателя нету.
Producer
class Program
{
public static async Task Main()
{
var config = new ProducerConfig
{
BootstrapServers = "localhost:9092"
};
// Create a producer that can be used to send messages to kafka that have no key and a value of type string
using var p = new ProducerBuilder<Null, string>(config).Build();
var i = 0;
while (true)
{
// Construct the message to send (generic type must match what was used above when creating the producer)
var message = new Message<Null, string>
{
Value = $"Message #{++i}"
};
// Send the message to our test topic in Kafka
var dr = await p.ProduceAsync("test", message);
Console.WriteLine($"Produced message '{dr.Value}' to topic {dr.Topic}, partition {dr.Partition}, offset {dr.Offset}");
Thread.Sleep(5000);
}
}
}
и Consumer
class Program
{
static void Main()
{
var conf = new ConsumerConfig
{
GroupId = "test-consumer-group",
BootstrapServers = "localhost:9092",
AutoOffsetReset = AutoOffsetReset.Earliest
};
using var c = new ConsumerBuilder<Ignore, string>(conf).Build();
c.Subscribe("test");
// Because Consume is a blocking call, we want to capture Ctrl+C and use a cancellation token to get out of our while loop and close the consumer gracefully.
var cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) => {
e.Cancel = true;
cts.Cancel();
};
try
{
while (true)
{
// Consume a message from the test topic. Pass in a cancellation token so we can break out of our loop when Ctrl+C is pressed
var cr = c.Consume(cts.Token);
Console.WriteLine($"Consumed message '{cr.Value}' from topic {cr.Topic}, partition {cr.Partition}, offset {cr.Offset}");
// Do something interesting with the message you consumed
}
}
catch (OperationCanceledException)
{
}
finally
{
c.Close();
}
}
}
Как я не пытался работать с Consumer не хочет он выводить сообщения. В итоге получаю вот это. Самое интересно, если я закрою приложение и снова запущу, то нумерация отправленных сообщения сохраниться, значит зукипер и кафка работают нормально. В чем может быть проблема?
