Recovery does not appear to save consumer arguments

[lukebakken]

Reported here: https://groups.google.com/g/rabbitmq-users/c/rHfpTUFudeM

Code provided to reproduce:

ConnectionFactory factory = new ConnectionFactory();
factory.Uri = new Uri(connectionString);

var rmqConnection = factory.CreateConnection();
IModel _channel =  rmqConnection.CreateModel();

add arguments 
var arguments = new Dictionary<string, object>();
arguments.Add("x-stream-offset", _streamOffset);

_channel.BasicConsume(streamId, false, "", arguments, messageReceiver);

Note: I have not yet reproduced this.

[DmytroMyslyvets]

So, about this issue. As I see it, it would be convenient to have an event that is called before the channels are restored. And the ability to update Consumer arguments. This is what I miss so much..
At the moment, I see only one way is manual recovery after a connection loss.

[Zerpet]

I've managed to reproduce the behaviour described. The library records the consumer arguments when AutorecoveringModel.BasicConsume() is called:

rabbitmq-dotnet-client/projects/RabbitMQ.Client/client/impl/AutorecoveringModel.cs

Lines 232 to 244 in b597225

	public string BasicConsume(
	string queue,
	bool autoAck,
	string consumerTag,
	bool noLocal,
	bool exclusive,
	IDictionary<string, object> arguments,
	IBasicConsumer consumer)
	{
	string result = InnerChannel.BasicConsume(queue, autoAck, consumerTag, noLocal, exclusive, arguments, consumer);
	_connection.RecordConsumer(new RecordedConsumer(this, consumer, queue, autoAck, result, exclusive, arguments));
	return result;
	}

At that time, we keep a copy of the consumer arguments (among other things):

rabbitmq-dotnet-client/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.Recording.cs

Lines 177 to 188 in 6b1a06b

	internal void RecordConsumer(in RecordedConsumer consumer)
	{
	if (!_config.TopologyRecoveryEnabled)
	{
	return;
	}
	lock (_recordedEntitiesLock)
	{
	_recordedConsumers[consumer.ConsumerTag] = consumer;
	}
	}

If the consumer arguments are modified afterwards, the library "misses" those updates.

This makes using Streams over AMQP a bit painful to use, because consumers are expected to track the offset themselves. However, during topology recovery, consumers don't have a mechanism to update the consumer arguments i.e. x-stream-offset value.

Whilst this is not technically a bug, it is a undesirable situation, because users are forced to opt-out from topology recovery in order to use Streams over AMQP effectively.

[Zerpet]

I've been looking at potential options to address this. I'm not convinced that firing an event before recovering a consumer will fix this problem. We could fire an event and pass the consumer arguments (or a reference to the arguments). The first counter-argument that comes to mind is that, we will have to fire an event for each consumer about to recover, wait for the event handler to finish, then proceed with the recovery. This could potentially delay the recovery, if the event handler, for some reason, takes a long time to finish. In addition, what should we do if the event handler throws an exception, or encounters an error? Should we abort the recovery? Should we just move on?

The alternative of disabling topology recovery is not catastrophic, but inconvenient. You can register an event handler for IAutorecoveringConnection.RecoverySucceeded and re-register your consumers (assuming your consumers keep track of their offsets, and assuming your class derives from a consumer class). Something like this, with better error checking and re-try logic.

    public void Subscribe(string queueName)
    {
        _consumerChannel = _connection.CreateModel();
        _consumerChannel.BasicQos(0, 100, false);
        var arguments = new Dictionary<string, object> { { "x-stream-offset", StreamOffset } };
        _consumerTag = _consumerChannel.BasicConsume(queueName, false, "", arguments, this);
    }

// -----

        var myConsumers = new List<MyConsumer>(1);
        myConsumers.Add(new MyConsumer(0L, rmqConnection));
        myConsumers[0].Subscribe(StreamId);

        rmqConnection.RecoverySucceeded += (sender, eventArgs) =>
        {
            foreach (var c in myConsumers)
            {
                c.Subscribe(StreamId);
            }
        };

[michaelklishin]

@Zerpet or connections or channels could "reload" their argument list from somewhere right before they start topology recovery. Not sure where but I assume all user API actions are on the Connection or IModel instances, so this sounds plausible to me.