How to consume messages using AvroDeserializer

[shahabganji]

I have a web application and I want to have a consumer there, deserializing messages and storing them to the database.

Here is how I have configured KafkaFlow:

builder.Services.AddKafkaFlowHostedService(kafka => kafka
    .AddCluster(cluster => cluster
        .WithBrokers(["localhost:9092"])
        .WithSchemaRegistry(config => config.Url = "localhost:8081")
        .AddConsumer(consumer => consumer
            .Topic("customer-transformed-topic")
            .WithGroupId("management-backend")
            .WithBufferSize(100)
            .WithAutoOffsetReset(AutoOffsetReset.Latest)
            .AddMiddlewares(middlewares => middlewares
                .AddSchemaRegistryAvroDeserializer()
                .AddTypedHandlers(handler => handler
                    .AddHandler<CustomerTransformedHandler>()
                    .WhenNoHandlerFound(_ => Console.WriteLine("No Handler found")))))
    )
);

and this is my handler:

namespace CustomerManagement.Backend.MessageHandlers;

public class CustomerTransformedHandler : IMessageHandler<Customer>
{
    private readonly ILogger<CustomerTransformedHandler> _logger;

    public CustomerTransformedHandler(ILogger<CustomerTransformedHandler> logger
    {
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
    }

    public async Task Handle(IMessageContext context, Customer message)
    {
        //todo: write to database as well
        
    }
}

and here is the Customer class:

using Avro;
using Avro.Specific;

namespace Transformer.Models;

public class Customer : ISpecificRecord
{
    public Guid Id { get; private set; }

    public string Username { get; set; }
    public string FirstName { get; private set; }
    public string LastName { get; private set; }
    public string Email { get; private set; }

    public Address DefaultDeliveryAddress { get; private set; }
    public Address DefaultBillingAddress { get; private set; }

    public Customer()
    {
    }
    

    public virtual object Get(int fieldPos)
    {
        switch (fieldPos)
        {
            case 0: return DefaultBillingAddress;
            case 1: return DefaultDeliveryAddress;
            case 2: return Email;
            case 3: return FirstName;
            case 4: return Id;
            case 5: return LastName;
            case 6: return Username;
            default: throw new AvroRuntimeException("Bad index " + fieldPos + " in Get()");
        }

        ;
    }

    public virtual void Put(int fieldPos, object fieldValue)
    {
        switch (fieldPos)
        {
            case 0:
                DefaultBillingAddress = (Address)fieldValue;
                break;
            case 1:
                DefaultDeliveryAddress = (Address)fieldValue;
                break;
            case 2:
                Email = (string)fieldValue;
                break;
            case 3:
                FirstName = (string)fieldValue;
                break;
            case 4:
                Id = (Guid)fieldValue;
                break;
            case 5:
                LastName = (string)fieldValue;
                break;
            case 6:
                Username = (string)fieldValue;
                break;
            default: throw new AvroRuntimeException("Bad index " + fieldPos + " in Put()");
        }

        ;
    }

    public Schema Schema => Schema.Parse(
        @"{""type"":""record"",""name"":""Customer"",""namespace"":""Transformer.Models"",""fields"":[{""name"":""DefaultBillingAddress"",""type"":{""type"":""record"",""name"":""Address"",""namespace"":""Transformer.Models"",""fields"":[{""name"":""City"",""type"":""string""},{""name"":""Lien1"",""type"":""string""},{""name"":""Zipcode"",""type"":""string""}]}},{""name"":""DefaultDeliveryAddress"",""type"":""Address""},{""name"":""Email"",""type"":""string""},{""name"":""FirstName"",""type"":""string""},{""name"":""Id"",""type"":{""type"":""string"",""logicalType"":""uuid""}},{""name"":""LastName"",""type"":""string""},{""name"":""Username"",""type"":""string""}]}");
}

and the Address class:

using Avro;
using Avro.Specific;

namespace Transformer.Models;

public record Address : ISpecificRecord
{
    public string Lien1 { get; private set; }
    public string Zipcode { get; private set; }
    public string City { get; private set; }

    public Address(string lien1, string zipcode, string city)
    {
        Lien1 = lien1;
        Zipcode = zipcode;
        City = city;
    }

    public virtual object Get(int fieldPos)
    {
        switch (fieldPos)
        {
            case 0: return City;
            case 1: return Lien1;
            case 2: return Zipcode;
            default: throw new AvroRuntimeException("Bad index " + fieldPos + " in Get()");
        }

        ;
    }

    public virtual void Put(int fieldPos, object fieldValue)
    {
        switch (fieldPos)
        {
            case 0:
                City = (string)fieldValue;
                break;
            case 1:
                Lien1 = (string)fieldValue;
                break;
            case 2:
                Zipcode = (string)fieldValue;
                break;
            default: throw new AvroRuntimeException("Bad index " + fieldPos + " in Put()");
        }

        ;
    }

    public Schema Schema => Schema.Parse(
        "{\"type\":\"record\",\"name\":\"Address\",\"namespace\":\"Transformer.Models\",\"fields\":[{\"na" +
        "me\":\"City\",\"type\":\"string\"},{\"name\":\"Lien1\",\"type\":\"string\"},{\"name\":\"Zipcode\",\"" +
        "type\":\"string\"}]}");
}

I checked out the sample for the schema registry, and I could not point out what is wrong with my code, also nothing is printed on the console when running the application, it is a Web API application.

Any help is highly appreciated!

[chezuma]

@shahabganji, did you resolve this? :)

[shahabganji]

@chezuma Yes, there were two tools for generating classes and schemas, and it seemed one of them is not compatible. I used the one from confluent and this has been solved. However, I had to use the actual SDK, to figure out the underlying error, somehow Kafkaflow was hiding the actual error in my scenario!