Unable to deserialize the body bytes into object after upgrade the RabbitMQ Client from 5.1.1 to 6.2.4

[LoremipsumSharp]

I am using RabbitMQ.Client for messaging and using dead letter pattern to implement retry mechanism, everything works fine with RabbitMQ.Client with version :5.1.1 .

The pseudo-code is below：

public async Task Handle(BasicDeliverEventArgs basicDeliverEventArgs)
{

    try
    {
        var bodyBytes = basicDeliverEventArgs.Body.ToArray();
        var bodyString = Encoding.UTF8.GetString(bodyBytes);
        // extension method with newtonsoft.json behind
        bodyString.TryDeserializeObject<T>(out var message);
        // do somesting
    }
    catch (Exception ex)
    {
        var properties = basicDeliverEventArgs.BasicProperties;
        if (properties.Headers == null)
        {
            properties.Headers = new Dictionary<string, object>();
        }

        var bodyBytes = basicDeliverEventArgs.Body.ToArray();


        var bodyString = Encoding.UTF8.GetString(bodyBytes);

        // retrive the x-death-count by header 
        var retryTimes = GetRetryTimes(properties.Headers);

        if (retryTimes <= _retryConfiguration.MaxCount)
        {
            using var channel = _conn.CreateModel();
            var delay = _retryConfiguration.GetNextDelay(retryTimes);
            properties.Expiration = delay.TotalMilliseconds.ToString();
            string deadLetterExchangeName = $"ErrorExchange_{_rabbitSubscriptionConfiguration.Queue}";
            string deadLetterQueueName = $"{_rabbitSubscriptionConfiguration.Queue}.Deadletter";
            string deadLetterRoutingKey = $"{_rabbitSubscriptionConfiguration.RoutingKey}.Deadletter";

            channel.ExchangeDeclare(deadLetterExchangeName, "direct", true, true);
            channel.QueueDeclare(deadLetterQueueName, true, false, false, new Dictionary<string, object>()
            {
                { "x-dead-letter-exchange", deadLetterExchangeName },
                { "x-dead-letter-routing-key",_rabbitSubscriptionConfiguration.RoutingKey}
            });

            channel.QueueBind(_rabbitSubscriptionConfiguration.Queue, deadLetterExchangeName, _rabbitSubscriptionConfiguration.RoutingKey);
            channel.QueueBind(deadLetterQueueName, deadLetterExchangeName, deadLetterRoutingKey);

            channel.BasicPublish(deadLetterExchangeName, deadLetterRoutingKey, false, properties, bodyBytes);
        }
        else
        {
            var messageStorage = _serviceProvider.GetService<IMessageStorage>();
            await messageStorage.StoreDeadLetter(bodyString, ex);
        }
    }
}
        

However, when I upgrade the RabbitMQ.Client to 6.2.4,there is some probability that the body bytes of message unable deserialize to origin object when I receiving the message again from dead letter queue.

When the retry ended,I inspect the dead letter in the database,some of them failed to convert to a normal string

By the way,the version of RabbitMQ Server is 3.8.5

[michaelklishin]

I will convert this issue to a GitHub discussion. Currently GitHub will automatically close and lock the issue even though your question will be transferred and responded to elsewhere. This is to let you know that we do not intend to ignore this but this is how the current GitHub conversion mechanism makes it seem for the users :(

[michaelklishin]

The client does not deserialize any body paylods. Can you share two bodies with each version and how they were serialized?

[lukebakken]

@LoremipsumSharp your code doesn't provide quite enough information. We need to know how you set up your Connection, Model and consumer classes to check. I'm assuming your consumer is a subclass of AsyncEventingBasicConsumer at this point.

[LoremipsumSharp]

@lukebakken
I have resolved this issue. Here is the situation, I am using EventingBasicConsumer the communicate with RabbitMQ , if i handle the BasicDeliverEventArgs in this way, there is some probability that i would received the wrong bytes

EventHandler<BasicDeliverEventArgs> onReceived =  async (sender, args) =>
{
    try
    {
        await handler.Handle(args)
    }
    catch (System.Exception e)
    {
       _logger.LogError(e.ToString());
    }
    finally
    {
       _channel.BasicAck(args.DeliveryTag, false);
    }
}

while if i handle BasicDeliverEventArgs in the blocking manner without async/await,every thing goes fine:

EventHandler<BasicDeliverEventArgs> onReceived =   (sender, args) =>
{
    try
    {
         handler.Handle(args).Wait()
    }
    catch (System.Exception e)
    {
        _logger.LogError(e.ToString())
    }
    finally
    {
       _channel.BasicAck(args.DeliveryTag, false);
    }
}

But I was wondering why the RabbitMQ.Client with version 5.1.1 works fine while 6.2.4 did`nt

[lukebakken]

I'm glad you found a workaround. The way the .NET client interacts with memory changed in version 6.x, so that handlers must deal with the body of a message before the handler exits, or copy the bytes. In your case, adding .Wait() ensures the memory will still be around.

I think the permanent fix is to copy the body bytes before calling an async handler.

If you have time to provide a complete example that I can compile and run to see the issue that would be very appreciated! Otherwise I have to spend time trying to figure out how to reproduce this issue rather than fixing it. Have a good weekend!

[lukebakken]

Just so you know, EventingBasicConsumer is NOT intended to be used with async / await. That is why the AsyncEventingBasicConsumer class exists.

I'm coming up with a test to try and reproduce what you reported. As I said in this comment, I would really appreciate some code to be able to see the same issue.

[lukebakken]

Please see #1173

This demonstrates the correct way to do async callbacks.