C# 无法使用关于 Kafka 主题的消息？

Question

我一直在查看 Confluent.Kafka 客户端（https://github.com/confluentinc/confluent-kafka-dotnet/）的几个示例，虽然我可以成功地让生产者将消息推送到 Kafka，但我我无法向消费者拉回任何消息。

通过 UI，我可以看到主题已创建并且消息正在进入该主题（目前有 10 个分区和 3 条消息），但我的消费者总是报告“分区结束”，而没有任何消息的消费（3 个保持在主题和“OnMessage”永远不会触发）。

然而，消费者肯定是在访问主题，并且可以在其中一个分区上看到 3 条消息：

分区结束：dotnet-test-topic [6] @3

它只是不消耗消息并触发 OnMessage()。有任何想法吗？

var conf = new Dictionary<string, object> 

{ 

    { "group.id", Guid.NewGuid().ToString() },

    { "bootstrap.servers", "mykafkacluster:9094" },

    { "sasl.mechanisms", "SCRAM-SHA-256" },

    { "security.protocol", "SASL_SSL" },

    { "sasl.username", "myuser" },

    { "sasl.password", "mypass" }

};

using (var producer = new Producer<string, string>(conf, new StringSerializer(Encoding.UTF8), new StringSerializer(Encoding.UTF8)))

{

    producer.ProduceAsync("dotnet-test-topic", "some key", "some value")

            .ContinueWith(result => 

            {

                var msg = result.Result;

                if (msg.Error.Code != ErrorCode.NoError)

                {

                    Console.WriteLine($"failed to deliver message: {msg.Error.Reason}");

                }

                else 

                {

                    Console.WriteLine($"delivered to: {result.Result.TopicPartitionOffset}");

                }

            });

    producer.Flush(TimeSpan.FromSeconds(10));

}

using (var consumer = new Consumer<string, string>(conf, new StringDeserializer(Encoding.UTF8), new StringDeserializer(Encoding.UTF8)))

{ 

    consumer.Subscribe("dotnet-test-topic");

    consumer.OnConsumeError += (_, err)

        => Console.WriteLine($"consume error: {err.Error.Reason}");

    consumer.OnMessage += (_, msg)

        => Console.WriteLine($"consumed: {msg.Value}");

    consumer.OnPartitionEOF += (_, tpo)

        => Console.WriteLine($"end of partition: {tpo}");

    while (true)

    {

        consumer.Poll(TimeSpan.FromMilliseconds(100));

    }  

}

Answer 1

如果不提供以下配置，则 OnMessage 事件似乎不会触发：{&nbsp;"auto.offset.reset",&nbsp;"smallest"&nbsp;}添加此内容后，我能够阅读有关该主题的消息。