System.InvalidOperationException: The method 'OnMessage' or 'OnMessageAsync' has already been called.Īt .OnMessage(MessageReceivePump pump)Īt .OnMessageAsync(Func 2 callback, OnMessageOptions onMessageOptions) at .OnMessageAsync(Func2 callback, OnMessageOptions options)Īt .ctor(NamespaceContext context, ClientContext clientContext, IPipe 1 receivePipe, ClientSettings clientSettings, ITaskSupervisor supervisor, ISendEndpointProvider sendEndpointProvider, IPublishEndpointProvider publishEndpointProvider) at .-Send>d_6.MoveNext() - End of stack trace from previous location where exception was thrown - at .ThrowForNonSuccess(Task task) at .HandleNonSuccessAndDebuggerNotification(Task task) at .-Send>d_4.MoveNext() - End of stack trace from previous location where exception was thrown - at .Throw() at .-Send>d_4.MoveNext() - End of stack trace from previous location where exception was thrown - at .ThrowForNonSuccess(Task task) at .HandleNonSuccessAndDebuggerNotification(Task task) at .d_6.MoveNext() - End of stack trace from previous location where exception was thrown - at .Throw() at .d_6.MoveNext() - End of stack trace from previous location where exception was thrown - at .ThrowForNonSuccess(Task task) at .HandleNonSuccessAndDebuggerNotification(Task task) at .ValidateEnd(Task task) at _5.MoveNext() It healed only after restart of the host. Hope this will help you to find the issue.Īfter this has happend, some receive endpoints failed so hard that Masstransit stopped receiving any messages (only for those failed endpoints). After digging around in Masstransit sources, looks like the problem could be in PrepareQueueClientFilter in this line of code:Ĭontext.GetOrAddPayload(() => clientContext) Īfter retry attempt, context stays the same and contains old clientContext (with QueueClient), so GetOrAddPayload does nothing. Somehow, when receive logic is retried, the OnMessageAsync method is called several times on the same QueueClient object. This one (see stacktrace) looks like a bug in Masstransit. I am concerned if Masstransit can self-heal after those exceptions properly. It would be great if you could confirm\comment on my assumptions. With the None retry filter, it would signify that exception should not be retried by a retry/redelivery filter defined prior to the None filter.While using Masstransit 3.5 with Azure ServiceBus transport I've noticed several exceptions in logs (happening with multiple receive endpoints). If you ignore the exception, it isn’t handled by the retry filter, so you’d need to handle it with no retry: When configuring multiple retry policies, which is what happens in this case, you’d need to ensure that exceptions you do not want retried are properly overridden. DoSomethingSpecificException is retried using global policy.
IConsumerConfigurator consumerConfigurator)ĮndpointConfigurator.UseMessageRetry(r => Protected override void ConfigureConsumer(IReceiveEndpointConfigurator endpointConfigurator, Public class DoSomethingConsumerDefinition : R.Intervals(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(10), TimeSpan.FromSeconds(30)) I was thinking that something like this should work(bus configuration part): I would like to create system wide retry policy (retry x times every y seconds) but modify/overwrite it in specific consumer (some exception should be ignored and moved to error queue without retry)
0 kommentar(er)
