Error on deserialization while using MessagePackSerialization.

[govind2207]

I am working on a POC. I created two console apps. One to publish message and other to consume it. I need to compress the data and serialize it. I decided to use MessagePackSerializer for the same. I am using Kafka as the messagebroker. Below is my producer config

var hostBuilder = Host.CreateDefaultBuilder(args)
.ConfigureLogging(logging =>
{
logging.ClearProviders();
logging.AddConsole();
logging.SetMinimumLevel(LogLevel.Debug); // <== Add this
})
.UseWolverine(opts =>
{
opts.UseKafka("localhost:9092")
.ConfigureProducers(producer =>
{
producer.BootstrapServers = "localhost:9092";
});

    var resolver = MessagePack.Resolvers.CompositeResolver.Create(
        MessagePack.Resolvers.ContractlessStandardResolver.Instance,
        MessagePack.Resolvers.StandardResolver.Instance
    );

    opts.UseMessagePackSerialization(o =>
    {
        o.WithResolver(resolver);
        o.WithCompression(MessagePackCompression.Lz4Block);
        o.WithAllowAssemblyVersionMismatch(true);
    });

    opts.PublishAllMessages()
    .ToKafkaTopic("test-topic");
});

// Await the built host
var host = await hostBuilder.StartAsync();

var publisher = host.Services.GetRequiredService();

// Loop to publish new messages each time a key is pressed
while (true)
{
Console.WriteLine("Press any key to publish a new message...");

// Wait for a key press
Console.ReadKey(intercept: true);


// Create a new message
var message = new KafkaMessages { Name = "Wolverine" };


// Publish the serialized message
await publisher.PublishAsync(message);

Console.WriteLine("Message published: " + message.Name);

}

Below is the consumer config:

var builder = Host.CreateDefaultBuilder(args)
.ConfigureLogging(logging =>
{
logging.ClearProviders();
logging.AddConsole();
logging.SetMinimumLevel(LogLevel.Debug); // <== Add this
})
.UseWolverine(opts =>
{
opts.UseKafka("localhost:9092")
.ConfigureConsumers(consumer =>
{
consumer.BootstrapServers = "localhost:9092";
});

    var resolver = MessagePack.Resolvers.CompositeResolver.Create(
        MessagePack.Resolvers.ContractlessStandardResolver.Instance,
        MessagePack.Resolvers.StandardResolver.Instance
    );

    opts.UseMessagePackSerialization(o =>
    {
        o.WithResolver(resolver);
        o.WithCompression(MessagePackCompression.Lz4Block);
        o.WithAllowAssemblyVersionMismatch(true);
    });

    // Start reading from the latest offset to avoid stale/bad messages
    opts.ListenToKafkaTopic("test-topic");
    
    opts.Discovery.IncludeAssembly(typeof(KafkaMessagesHandler).Assembly);
});

await builder.Build().RunAsync();

Below is the class that I am using

[MessagePackObject(keyAsPropertyName: true)]
public partial class KafkaMessages
{
public string Name { get; set; } = string.Empty;
}

This class I have added in a common library and referred in both the projects.

While consuming I am getting below error

fail: Wolverine.Runtime.WolverineRuntime[108]
Envelope Envelope #08dd8d48-d327-d153-00be-433d89d00000/CorrelationId=c055fd61-3061-47f8-ad9a-95b7ae4ad32a from KafkaProducer to kafka://topic/test-topic was moved to the error queue
MessagePack.MessagePackSerializationException: Failed to deserialize Kafka.Common.KafkaMessages value.
---> MessagePack.MessagePackSerializationException: Unexpected msgpack code 239 (negative fixint) encountered.
at MessagePack.MessagePackReader.ThrowInvalidCode(Byte code)
at MessagePack.MessagePackReader.TryReadMapHeader(Int32& count)
at MessagePack.MessagePackReader.ReadMapHeader()
at MessagePack.Formatters.Kafka_Common_KafkaMessagesFormatter1.Deserialize(MessagePackReader& reader, MessagePackSerializerOptions options)
at MessagePack.MessagePackSerializer.Deserialize[T](MessagePackReader& reader, MessagePackSerializerOptions options)
--- End of inner exception stack trace ---
at MessagePack.MessagePackSerializer.Deserialize[T](MessagePackReader& reader, MessagePackSerializerOptions options)
at lambda_method7(Closure, ReadOnlyMemory1, MessagePackSerializerOptions, CancellationToken) at MessagePack.MessagePackSerializer.Deserialize(Type type, ReadOnlyMemory1 bytes, MessagePackSerializerOptions options, CancellationToken cancellationToken)
at Wolverine.MessagePack.Internal.MessagePackMessageSerializer.ReadFromData(Type messageType, Envelope envelope) in /home/runner/work/wolverine/wolverine/src/Extensions/Wolverine.MessagePack/Internal/MessagePackMessageSerializer.cs:line 36
at Wolverine.Runtime.HandlerPipeline.TryDeserializeEnvelope(Envelope envelope, IContinuation& continuation) in /home/runner/work/wolverine/wolverine/src/Wolverine/Runtime/HandlerPipeline.cs:line 130

I don't understand where I am doing it wrong. Please guide on this.