c# - avro序列化的C# confluent kafka问题

问题描述

我正在使用 docker 在我的测试项目中运行来自https://github.com/confluentinc/cp-all-in-one的 kafka 和其他服务，以及用于 kafka、avro 和 schemaRegistry 的融合 nuget 包。

如果要发送 json 消息，到目前为止我没有问题，但我正在努力发送 avro 序列化消息。

我看到了https://github.com/confluentinc/confluent-kafka-dotnet/tree/master/examples/AvroSpecific示例，我尝试以相同的方式进行操作，但最终出现如下异常：

本地： C:\Users\lu95eb\source\repos\Kafka_playground\Kafka producer\KafkaService.cs:line 126 中 Kafka_producer.KafkaService.d__10.MoveNext()
处 Confluent.Kafka.Producer 2.<ProduceAsync>d__52.MoveNext() at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task) at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) at System.Runtime.CompilerServices.TaskAwaiter1.GetResult() 的值序列化错误

除了内部异常

你调用的对象是空的。
在 Confluent.SchemaRegistry.Serdes.SpecificSerializerImpl 1..ctor(ISchemaRegistryClient schemaRegistryClient, Boolean autoRegisterSchema, Int32 initialBufferSize) at Confluent.SchemaRegistry.Serdes.AvroSerializer1.d__6.MoveNext() 在 System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task) 在 System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) 在 System.Runtime.CompilerServices。 Confluent.Kafka.Producer`2.d__52.MoveNext() 处的 TaskAwaiter.ValidateEnd(Task 任务)

这是我的 SpecificRecord 类

public class UserInfo : ISpecificRecord
{
    public string Name { get; set; }
    public int[] Numbers { get; set; }

    public Schema Schema => Schema.Parse(@"
        {
          ""name"": ""UserInfo"",
          ""type"": ""record"",
          ""namespace"": ""kafka"",
          ""fields"": [
            {
              ""name"": ""Name"",
              ""type"": ""string""
            },
            {
              ""name"": ""Numbers"",
              ""type"": {
                ""type"": ""array"",
                ""items"": ""int""
              }
            }
          ]
        }
        ");

    public object Get(int fieldPos)
    {
        switch (fieldPos)
        {
            case 0: return Name;
            case 1: return Numbers;
            default: throw new AvroRuntimeException($"Bad index {fieldPos} in Get()");
        }
    }

    public void Put(int fieldPos, object fieldValue)
    {
        switch (fieldPos)
        {
            case 0: Name = (string)fieldValue; break;
            case 1: Numbers = (int[])fieldValue; break;
            default: throw new AvroRuntimeException($"Bad index {fieldPos} in Put()");
        }
    }
}

以及用于发送消息的方法

private async Task SendSpecificRecord(UserInfo userInfo)
    {
        using (var schemaRegistry = new CachedSchemaRegistryClient(new SchemaRegistryConfig { Url = _schemaRegistryUrl }))
        using (var producer =
            new ProducerBuilder<string, UserInfo>(new ProducerConfig { BootstrapServers = _brokerUrl })
                .SetKeySerializer(new AvroSerializer<string>(schemaRegistry))
                .SetValueSerializer(new AvroSerializer<UserInfo>(schemaRegistry))
                .Build())
        {

            var message = new Message<string, UserInfo>
            {
                Key = userInfo.Name,
                Value = userInfo
            };


            await producer.ProduceAsync(SpecificTopic, message);
        }
    }

KafkaService.cs：第 126 行是await producer.ProduceAsync(SpecificTopic, message);

就像我在开始时写的那样，我对 schemaRegistry 没有任何问题——我已经注册了模式并且它们对 json 工作正常，我对主题、代理、消费者或其他任何东西都没有问题。

如果有人能指出我做错了什么，我将不胜感激。先感谢您。

标签： c#apache-kafkaavroconfluent-platform

c# - avro序列化的C# confluent kafka问题

问题描述

解决方案

推荐阅读