Error producing message Local Value serialization error in Kafka

Question

Hi I am working in Kafka and .Net. I am trying to produce message as below. I have created configuration for schema registry and producer as below. I am using certificates to connect to schema registry

var schemaRegistryConfig = new SchemaRegistryConfig
    {
        Url = "",
        SslKeystoreLocation = Path.Combine(Directory.GetCurrentDirectory(), @"Certs/decodedca.p12"),
        SslKeystorePassword = "",
        EnableSslCertificateVerification = false
    };
 var producerConfig = new ProducerConfig
    {
        BootstrapServers = "3",
        SaslMechanism = SaslMechanism.ScramSha512,
        SaslUsername = "",
        SaslPassword = "",
        SecurityProtocol = SecurityProtocol.SaslSsl,
        SslCaLocation = Path.Combine(Directory.GetCurrentDirectory(), @"Certs/srdecodedca.crt"),
        EnableSslCertificateVerification = false
    };
 var avroSerializerConfig = new AvroSerializerConfig
    {
        // optional Avro serializer properties:
        BufferBytes = 100
    };
 using (var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig))
    using (var producer =
        new ProducerBuilder<string, MLFixtureEmp>(producerConfig)
            .SetValueSerializer(new AvroSerializer<MLFixtureEmp>(schemaRegistry, avroSerializerConfig))
            .Build())
    {
        MLFixtureEmp mLFixtureEmp = new MLFixtureEmp()
        {
            deliveryDate = DateTime.Now,
            ISOCurrencyCode = "INR",
            chartererName = "ss",
            estimateRedeliveryDate = DateTime.Today,
            fixtureId = 4,
            imoNumber = "123",
            managingOwnerName = "dd",
            maximumDate = DateTime.Now,
            minimumDate = DateTime.Now,
            purchaseObligation = "stri",
            rate = 123,
            redeliveryPortName = "dd",
            RedeliveryRanges = "dsd",
            vesselOwnershipType = "dsds"
        };
        producer.ProduceAsync("mytopic", new Message<string, MLFixtureEmp>
        {
            Key = "test",
            Value = mLFixtureEmp
        }).ContinueWith(
            task =>
            {
                if (!task.IsFaulted)
                {
                    Console.WriteLine($"produced to: {task.Result.TopicPartitionOffset}");
                    return;
                }
                Console.WriteLine($"error producing message: {task.Exception.InnerException}");
            });
    }

Whenever I try to push data to topic i get below error message.

error producing message: Confluent.Kafka.ProduceException`2[System.String,MaerskLine.CHAMPS.Dto.MLFixtureEmp]: Local: Value serialization error
 ---> System.InvalidOperationException: AvroSerializer only accepts type parameters of int, bool, double, string, float, long, byte[], instances of ISpecificRecord and subclasses of SpecificFixed.
   at Confluent.SchemaRegistry.Serdes.SpecificSerializerImpl`1..ctor(ISchemaRegistryClient schemaRegistryClient, Boolean autoRegisterSchema, Int32 initialBufferSize)
   at Confluent.SchemaRegistry.Serdes.AvroSerializer`1.SerializeAsync(T value, SerializationContext context)
   at Confluent.Kafka.Producer`2.ProduceAsync(TopicPartition topicPartition, Message`2 message, CancellationToken cancellationToken)
   --- End of inner exception stack trace ---
   at Confluent.Kafka.Producer`2.ProduceAsync(TopicPartition topicPartition, Message`2 message, CancellationToken cancellationToken)

I am not able to figure out the issue. Can someone please help me to identify the issue? Any help would be appreciated. Thanks

Answer 1

As the error message says AvroSerializer only accepts int, bool, double, string, float, long, byte[], instances of ISpecificRecord and subclasses of SpecificFixed.

So, one solution would be: MLFixtureEmp type should implement ISpecificRecord interface.