System.AggregateException while using Stream Client

[plewam]

Describe the bug

While calling the IProducer.Send() method I spradically encounter a System.AggregateException which terminated my whole application. I am even unable to catch this exception with a try-catch clause.

Maybe this is soemthing for you to look at. I am using version 1.3 (latest) of the Stream library.

CoreCLR Version: 6.0.1523.11507 .NET Version: 6.0.15 Description: The process was terminated due to an unhandled exception. Exception Info: System.AggregateException: One or more errors occurred. (Operation is not valid due to the current state of the object.) ---> System.InvalidOperationException: Operation is not valid due to the current state of the object. at System.Threading.Tasks.Sources.ManualResetValueTaskSourceCore1.SignalCompletion() at System.Threading.Tasks.Sources.ManualResetValueTaskSourceCore1.SetException(Exception error) at RabbitMQ.Stream.Client.ManualResetValueTaskSource1.SetException(Exception error) in /_/RabbitMQ.Stream.Client/Client.cs:line 705 at RabbitMQ.Stream.Client.Client.<>c__542.<Request>b__54_0(Object valueTaskSource) in /_/RabbitMQ.Stream.Client/Client.cs:line 354 at System.Threading.CancellationTokenSource.Invoke(Delegate d, Object state, CancellationTokenSource source) at System.Threading.CancellationTokenSource.CallbackNode.<>c.<ExecuteCallback>b__9_0(Object s) at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state) --- End of stack trace from previous location --- at System.Threading.CancellationTokenSource.CallbackNode.ExecuteCallback() at System.Threading.CancellationTokenSource.ExecuteCallbackHandlers(Boolean throwOnFirstException) --- End of inner exception stack trace --- at System.Threading.CancellationTokenSource.ExecuteCallbackHandlers(Boolean throwOnFirstException) at System.Threading.CancellationTokenSource.NotifyCancellation(Boolean throwOnFirstException) at System.Threading.CancellationTokenSource.TimerCallback(Object state) at System.Threading.TimerQueueTimer.Fire(Boolean isThreadPool) at System.Threading.TimerQueue.FireNextTimers()

Reproduction steps

For me it happens every couple of hours during continuous message sending

Expected behavior

Ideally either the Send or Confim method of the Producer would return a negative publish result

Additional context

No response

[Gsantomaggio]

Can you provide more info? Like a snippet code. If you are using Tasks .. etc
It seems that the thread pool is full.

Just reporting the stack trace does not help too much

[plewam]

Hi, no problem. May need a little bit of refactoring, but this is how I publish messages. Not sure whether this helps you in any way. My publish method will also create the Producer in case it does not exist. This is done synchronous.
Please let me know if you need more info.

public Task PublishMessage(List<IDataAgentEntity> messages, Type type)
        {
            return Task.Run(new Action(async () =>
            {
                try
                {
                    //--------------------------------------------------------------------
                    // Check if there are any messages to send at all
                    //--------------------------------------------------------------------
                    if (messages == null || messages.Count == 0)
                    {
                        Logger.Log(LogLevel.Warning, $"No message to send found when publishing stream messages");
                        return;
                    }

                    //--------------------------------------------------------------------
                    // Check if stream is open at all
                    //--------------------------------------------------------------------
                    if (!OpenOrVerifyConnection())
                    {
                        Logger.Log(LogLevel.Error, $"Stream is not open during push of type <{type.Name}>");
                        return;
                    }

                    //--------------------------------------------------------------------
                    // Try to get attribute from type
                    //--------------------------------------------------------------------
                    StreamAttribute? streamAttribute = (StreamAttribute?)Attribute.GetCustomAttribute(type, typeof(StreamAttribute));

                    //--------------------------------------------------------------------
                    // Check if attribute is null
                    //--------------------------------------------------------------------
                    if (streamAttribute == null)
                    {
                        Logger.Log(LogLevel.Error, $"Type <{type.Name}> has no StreamAttribute");
                        return;
                    }

                    //--------------------------------------------------------------------
                    // Check if given stream exists
                    //--------------------------------------------------------------------
                    if (Streams.Contains(streamAttribute.StreamName) == false)
                    {
                        if (Stream.StreamExists(streamAttribute.StreamName).Result == false)
                        {
                            Logger.Log(LogLevel.Warning, $"Stream with name <{streamAttribute.StreamName}> does not exist. Create it");
                            Stream.CreateStream(new StreamSpec(streamAttribute.StreamName)
                            {
                                MaxLengthBytes = StreamLengthBytes //10 GBytes
                            }).Wait();

                            Streams.Add(streamAttribute.StreamName);
                        }
                    }

                    //--------------------------------------------------------------------
                    // Check if we got a producer for that stream already
                    //--------------------------------------------------------------------
                    if (!Producers.ContainsKey(streamAttribute.StreamName))
                    {
                        Logger.Log(LogLevel.Information, $"No producer created for stream <{streamAttribute.StreamName}>");
                        IProducer p = Stream.CreateRawProducer(
                        new RawProducerConfig(streamAttribute.StreamName)
                        {
                            Reference = Guid.NewGuid().ToString(),
                            ConfirmHandler = ConfirmMessage,
                            BatchSize = 1
                        }, Logger).Result;
                        Producers.Add(streamAttribute.StreamName, p);
                        Logger.Log(LogLevel.Information, $"Producer created and added for stream <{streamAttribute.StreamName}>");
                    }

                    IProducer producer = Producers[streamAttribute.StreamName];

                    List<Message> messagesToSend = new List<Message>();
                    foreach (IDataAgentEntity message in messages)
                    {
                        messagesToSend.Add(new Message(Encoding.UTF8.GetBytes(EntityConverter.Serialize(message))));
                    }

                    Logger.Log(LogLevel.Debug, $"Start publishing message for producer <{streamAttribute.StreamName}>. <{messagesToSend.Count}> message to send. PublishId <{PublishId}>");
                    await producer.Send(PublishId, messagesToSend, CompressionType.Gzip);
                    Logger.Log(LogLevel.Debug, $"Finished publishing message for producer <{streamAttribute.StreamName}>");
                    PublishId++;
                }
                catch (AggregateException agex)
                {
                    foreach (var innerException in agex.InnerExceptions)
                    {
                        Logger.Log(LogLevel.Error, innerException.ToString());
                    }
                }
                catch (Exception ex)
                {
                    Logger.Log(LogLevel.Error, ex.ToString());
                }
            }));
        }

[Gsantomaggio]

Thank you, ok you are using the sub-entry batch. That helps to understand where to debug.

[Gsantomaggio]

You are using the RawProducer class, which is not thread-safe. In general, you should use the Producer class.

I suspect that your problem is related to thread status.

The code you should use:

  var producer = await Producer.Create(new ProducerConfig(system, Stream), producerLogger);
        Console.WriteLine("Start sending messages ...");

        for (ulong zIndex = 0; zIndex < 500000; zIndex++)
        {
            var messagesToSend = new List<Message>();
            for (ulong i = 0; i < <<XXX_SUBENTRY_VALUE>>; i++)
            {
                messagesToSend.Add(new Message(new byte[50]));
            }

            await producer.Send(messagesToSend, CompressionType.Gzip).ConfigureAwait(false);
        }

Note:

• you should use the Producer class so you can handle the MessageTimeOut and lot of automatic stuff
• you should add .ConfigureAwait(false);
  -You should not use the Reference field. As written on the documentation: https://rabbitmq.github.io/rabbitmq-stream-dotnet-client/stable/htmlsingle/index.html#creating-a-producer:

Producer instances are thread-safe when the `Reference` is not set. Starting from version 1.2.0 the Reference field is deprecated. Reference is needed for deduplication. See the Deduplication section for more details.

The code looks a bit complex, I am trying to reproduce the issue without success.
Please try to reproduce the issue using a more straightforward example.

[plewam]

Thank you very much for this in-detailled response. I have switched to your proposed solution. However I am still seeing the exception, see

CoreCLR Version: 6.0.1623.17311 .NET Version: 6.0.16 Description: The process was terminated due to an unhandled exception. Exception Info: System.AggregateException: One or more errors occurred. (Operation is not valid due to the current state of the object.) ---> System.InvalidOperationException: Operation is not valid due to the current state of the object. at System.Threading.Tasks.Sources.ManualResetValueTaskSourceCore1.SignalCompletion() at System.Threading.Tasks.Sources.ManualResetValueTaskSourceCore1.SetException(Exception error) at RabbitMQ.Stream.Client.ManualResetValueTaskSource1.SetException(Exception error) in /_/RabbitMQ.Stream.Client/Client.cs:line 705 at RabbitMQ.Stream.Client.Client.<>c__542.<Request>b__54_0(Object valueTaskSource) in /_/RabbitMQ.Stream.Client/Client.cs:line 354 at System.Threading.CancellationTokenSource.Invoke(Delegate d, Object state, CancellationTokenSource source) at System.Threading.CancellationTokenSource.CallbackNode.<>c.<ExecuteCallback>b__9_0(Object s) at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state) --- End of stack trace from previous location --- at System.Threading.CancellationTokenSource.CallbackNode.ExecuteCallback() at System.Threading.CancellationTokenSource.ExecuteCallbackHandlers(Boolean throwOnFirstException) --- End of inner exception stack trace --- at System.Threading.CancellationTokenSource.ExecuteCallbackHandlers(Boolean throwOnFirstException) at System.Threading.CancellationTokenSource.NotifyCancellation(Boolean throwOnFirstException) at System.Threading.CancellationTokenSource.TimerCallback(Object state) at System.Threading.TimerQueueTimer.Fire(Boolean isThreadPool) at System.Threading.TimerQueue.FireNextTimers()

This is how my publish method looks now.

public async Task<bool> PublishMessage(List<IDataAgentEntity> messages, Type type)
        {
            bool result = false;
            try
            {
                if (messages == null || messages.Count == 0)
                {
                    return result;
                }

                if (!await OpenOrVerifyConnection())
                {
                    return result;
                }

                StreamAttribute? streamAttribute = (StreamAttribute?)Attribute.GetCustomAttribute(type, typeof(StreamAttribute));

                if (streamAttribute == null)
                {
                    return result;
                }

                if (Streams.Contains(streamAttribute.StreamName) == false)
                {
                    if (await Stream.StreamExists(streamAttribute.StreamName) == false)
                    {
                        await Stream.CreateStream(new StreamSpec(streamAttribute.StreamName)
                        {
                            MaxLengthBytes = StreamLengthBytes //10 GBytes
                        });

                        Streams.Add(streamAttribute.StreamName);
                    }
                }

                if (Producer == null)
                {
                    Producer = await Producer.Create(
                    new ProducerConfig(this.Stream, streamAttribute.StreamName)
                    {
                        TimeoutMessageAfter = new TimeSpan(0,0,30)
                    });
                }

                List<Message> messagesToSend = new List<Message>();
                foreach (IDataAgentEntity message in messages)
                {
                    messagesToSend.Add(new Message(Encoding.UTF8.GetBytes(EntityConverter.Serialize(message))));
                }

                await Producer.Send(messagesToSend, CompressionType.Gzip).ConfigureAwait(false);
                result = true;
            }
            catch (AggregateException agex)
            {
                foreach (var innerException in agex.InnerExceptions)
                {
                    Logger.Log(LogLevel.Error, innerException.ToString());
                }
            }
            catch (Exception ex)
            {
                Logger.Log(LogLevel.Error, ex.ToString());
            }
            return result;
        }

[Gsantomaggio]

@plewam, thank you but :

• your code is not complete.
• the code is a bit complex.
• We'd need an easy way to reproduce the issue or a repo that we can build.
• We are not able to build the code.

For example: Where the Producer variable is defined?

[plewam]

Good morning, I have been running stable since 2 days now with millions of pushed datasets. I have removed the following code:
if (!await OpenOrVerifyConnection()) { return result; }

Inside the method I have checked whether the Stream System was still valid by checking the IsOpen property. I hope the StreamSystem auto recovers in case of connection failure as I currently have no recovery mechanisms. Maybe I still need to do some testing around that scenario. Anyways thanks for your help @Gsantomaggio

[Gsantomaggio]

@plewam ok. Please keep me updated. Producer and Consumer classes have the auto-recovery mechanism.
Note that here:

new ProducerConfig(this.Stream, streamAttribute.StreamName)
                    {
                        TimeoutMessageAfter = new TimeSpan(0,0,30)
                    });

You are not handling the confirmation messages. If a message is not confirmed, you won't be aware.

Read here: https://rabbitmq.github.io/rabbitmq-stream-dotnet-client/stable/htmlsingle/index.html#_sending_messages the ConfirmationHandler

[jonnepmyra]

We've seen this type of exception with the same consequences aswell and I agree with @Gsantomaggio. I'd guess it's a threading problem caused by the usage of the lib, and not a problem with the lib itself.

There's alot of deadlock prone code in the supplied example (sync over async with blocking Wait() and Task.Result) which should be fixed.

[Gsantomaggio]

Will close due of inactivity

[plewam]

Hi @Gsantomaggio, I am still struggling with the above AggregateException. I can control how often it happens by changing values for TimeoutMessageAfter, MaxInFlight and MessagesBufferSize. The smaller values I use the more it happens. In my code I removed all wait() and Task.Result occurences so that it is now completely async.

What I additionally see is that I get a ClientTimeoutErrorduring message confirmation.

Any recommendation for me?

Kind regards

[Gsantomaggio]

@plewam We analysed the logs and tried to reproduce the issue without success.

(classic queues) never have these disconnect problems.

Stream-Plugin and AMQP use different ports so it could be. Maybe the server logs could help.

This problem btw is not System.AggregateException this is another kind of problem, right?

[Gsantomaggio]

@plewam we are open to help; feel free to open another discussion.

[plewam]

Thanks @Gsantomaggio, really appreciate it. I am now configuring the logging of RabbitMQ as it is creating ridiculously large log files (>2 GB). Furthermore I have upgraded to your latest librariy version 1.7.2. Additionally I upgraded Erlang to 26.1.1 and RabbitMQ to 3.13-beta 6. I will now wait for the System.AggregateException to occur again. I will then provide you with my application and the RabbitMQ server logs.
If I realize that the errors and the exception I provided are two separate topics, I will open another discussion.

[plewam]

@Gsantomaggio, I just had a look again at the logs. The issue is gone when I use client side batching. No errors for the nearly two weeks.

[Gsantomaggio]

@plewam ok, thanks for the update!