c# - KafkaConsumer 在检索消息时永远挂起

Question

我想使用 confluent kafka 库从主题中提取消息，但是 var cr = c.Consume(cts.Token); 在这一行中，应用程序等待无限长的时间并且没有给出任何错误。让你永远等待。

我的源代码如下

var conf = new ConsumerConfig
            {
                BootstrapServers = "url:443",
                SecurityProtocol = Confluent.Kafka.SecurityProtocol.Ssl,
                SslCaLocation = "cert.crt",
                AutoOffsetReset = AutoOffsetReset.Earliest,
            };

            using (var c = new ConsumerBuilder<string, string>(conf).Build())
            {
                c.Subscribe("Test");

                CancellationTokenSource cts = new CancellationTokenSource();
                Console.CancelKeyPress += (_, e) => {
                    e.Cancel = true; // prevent the process from terminating.
                    cts.Cancel();
                };

                try
                {
                    while (true)
                    {
                        try
                        {
                            var cr = c.Consume(cts.Token);
                            Console.WriteLine($"Consumed message '{cr.Value}' at: '{cr.TopicPartitionOffset}'.");
                        }
                        catch (ConsumeException e)
                        {
                            Console.WriteLine($"Error occured: {e.Error.Reason}");
                        }
                    }
                }
                catch (OperationCanceledException)
                {
                    
                    c.Close();
                }