Kafka 消费消息然后生产到另一个主题

Question

我必须从 Kafka 主题消费，获取消息并做一些 json 清理和过滤工作，然后我需要将新消息生成到另一个 Kafka 主题，我的代码是这样的：

public static YamlMappingNode configs;
        public static void Main(string[] args)
        {
            using (var reader = new StreamReader(Path.Combine(Directory.GetCurrentDirectory(), ".gitlab-ci.yml")))
            {
                var yaml = new YamlStream();
                yaml.Load(reader);

                //find variables 
                configs = (YamlMappingNode)yaml.Documents[0].RootNode;
                configs = (YamlMappingNode)configs.Children.Where(k => k.Key.ToString() == "variables")?.FirstOrDefault().Value;

            }

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

 public static async void Run_ManualAssign(YamlMappingNode configs, CancellationToken cancellationToken)
        {
            var brokerList = configs.Where(k => k.Key.ToString() == "kfk_broker")?.FirstOrDefault().Value.ToString();
            var topics = configs.Where(k => k.Key.ToString() == "input_kfk_topic")?.FirstOrDefault().Value.ToString();
            var config = new ConsumerConfig
            {
                // the group.id property must be specified when creating a consumer, even 
                // if you do not intend to use any consumer group functionality.
                GroupId = new Guid().ToString(),
                BootstrapServers = brokerList,
                // partition offsets can be committed to a group even by consumers not
                // subscribed to the group. in this example, auto commit is disabled
                // to prevent this from occurring.
                EnableAutoCommit = true
            };

            using (var consumer =
                new ConsumerBuilder<Ignore, string>(config)
                    .SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))
                    .Build())
            {
                //consumer.Assign(topics.Select(topic => new TopicPartitionOffset(topic, 0, Offset.Beginning)).ToList());
                consumer.Assign(new TopicPartitionOffset(topics, 0, Offset.End));
                //var producer = new ProducerBuilder<Null, string>(config).Build();
                try
                {
                    while (true)
                    {
                        try
                        {
                            var consumeResult = consumer.Consume(cancellationToken);
                            /// Note: End of partition notification has not been enabled, so
                            /// it is guaranteed that the ConsumeResult instance corresponds
                            /// to a Message, and not a PartitionEOF event.

                            //filter message  
                            var result = ReadMessage(configs, consumeResult.Message.Value);
                            //send to kafka topic
                            await Run_ProducerAsync(configs, result);
                        }
                        catch (ConsumeException e)
                        {

                            Console.WriteLine($"Consume error: {e.Error.Reason}");
                        }
                    }
                }
                catch (OperationCanceledException)
                {
                    Console.WriteLine("Closing consumer.");
                    consumer.Close();
                }
            }
        }
        #endregion

        #region Run_Producer
        public static async Task Run_ProducerAsync(YamlMappingNode configs, string message)
        {
            var brokerList = configs.Where(k => k.Key.ToString() == "kfk_broker")?.FirstOrDefault().Value.ToString();
            var topicName = configs.Where(k => k.Key.ToString() == "target_kafka_topic")?.FirstOrDefault().Value.ToString();
            var config = new ProducerConfig {
                BootstrapServers = brokerList,
            };

            using (var producer = new ProducerBuilder<Null, string>(config).Build())
            {
                try
                {
                    /// Note: Awaiting the asynchronous produce request below prevents flow of execution
                    /// from proceeding until the acknowledgement from the broker is received (at the 
                    /// expense of low throughput).
                    var deliveryReport = await producer.ProduceAsync(topicName, new Message<Null, string> { Value = message });
                    producer.Flush(TimeSpan.FromSeconds(10));
                    Console.WriteLine($"delivered to: {deliveryReport.TopicPartitionOffset}");
                }
                catch (ProduceException<string, string> e)
                {
                    Console.WriteLine($"failed to deliver message: {e.Message} [{e.Error.Code}]");
                }
            }
        }
        #endregion

我是不是做错了什么？执行var deliveryReport = await producer.ProduceAsync(topicName, new Message<Null, string> { Value = message });时程序立即存在，无错误信息，无错误代码。

同时我使用了 Python 并为 Producer 配置了相同的配置，效果很好。

Answer 1

Run_ManualAssign(configs, cts.Token);

对于 Main 函数中的这一行，您是在同步函数中调用异步而不等待。因此程序在此调用开始后立即退出（未完成，因为它是异步的）

你可以有 2 个选择

使用async Main 函数并在此调用前添加await。

https://docs.microsoft.com/en-us/dotnet/csharp/language-reference/proposals/csharp-7.1/async-main
如果你真的想在同步函数中调用异步函数

Run_ManualAssign(configs, ts.Token).ConfigureAwait(false).GetAwaiter().GetResult();

Answer 2

我解决了这个问题，但我不知道为什么。我在这里开了一个issue

Kafka 消费消息然后生产到另一个主题

Kafka consume message and then produce to another topic

c#

apache-kafka

.net-core

kafka-consumer-api

kafka-producer-api