带有AzureStorageQueues的NServiceBus由于消息属性更改而无法从输入队列中删除有毒消息

Question

我正在尝试一个新的NServiceBus项目，该项目使用Azure存储队列来进行消息传输和JSON序列化，并使用自定义消息展开逻辑，如下所示：

            var jsonSerializer = new Newtonsoft.Json.JsonSerializer();
            transportExtensions.UnwrapMessagesWith(cloudQueueMessage =>
            {
                using (var stream = new MemoryStream(cloudQueueMessage.AsBytes))
                using (var streamReader = new StreamReader(stream))
                using (var textReader = new JsonTextReader(streamReader))
                {
                    try
                    {
                        var jObject = JObject.Load(textReader);

                        using (var jsonReader = jObject.CreateReader())
                        {
                            // Try deserialize to a NServiceBus envelope first
                            var wrapper = jsonSerializer.Deserialize<MessageWrapper>(jsonReader);

                            if (wrapper.MessageIntent != default)
                            {
                                // This was a envelope message
                                return wrapper;
                            }
                        }

                        // Otherwise this was an EventGrid event
                        using (var jsonReader = jObject.CreateReader())
                        {
                            var @event = jsonSerializer.Deserialize<EventGridEvent>(jsonReader);

                            var wrapper = new MessageWrapper
                            {
                                Id = @event.Id,
                                Headers = new Dictionary<string, string>
                            {
                                { "NServiceBus.EnclosedMessageTypes", @event.EventType },
                                { "NServiceBus.MessageIntent", "Publish" },
                                { "EventGrid.topic", @event.Topic },
                                { "EventGrid.subject", @event.Subject },
                                { "EventGrid.eventTime", @event.EventTime.ToString("u") },
                                { "EventGrid.dataVersion", @event.DataVersion },
                                { "EventGrid.metadataVersion", @event.MetadataVersion },
                            },
                                Body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(@event.Data)),
                                MessageIntent = MessageIntentEnum.Publish
                            };

                            return wrapper;
                        }
                    }
                    catch
                    {
                        logger.Error("Message deserialization failed, sending message to error queue");
                        throw;
                    }
                }
            });

自定义消息展开逻辑对于格式正确的JSON消息正确工作，当将格式不正确的JSON消息放入输入队列时，自定义消息展开逻辑将在使用情况的第一行出错，在该行中我创建了jObject，即预期的行为。但是，当自定义消息展开逻辑失败时，该错误将被MessageRetrieved类中的逻辑捕获，该类是NServiceBus.Azure.Transports.WindowsAzureStorageQueues NuGet包（v8.2.0）的一部分，如下所示：

        public async Task<MessageWrapper> Unwrap()
        {
            try
            {
                Logger.DebugFormat("Unwrapping message with native ID: '{0}'", rawMessage.Id);
                return unwrapper.Unwrap(rawMessage);
            }
            catch (Exception ex)
            {
                await errorQueue.AddMessageAsync(rawMessage).ConfigureAwait(false);
                await inputQueue.DeleteMessageAsync(rawMessage).ConfigureAwait(false);

                throw new SerializationException($"Failed to deserialize message envelope for message with id {rawMessage.Id}. Make sure the configured serializer is used across all endpoints or configure the message wrapper serializer for this endpoint using the `SerializeMessageWrapperWith` extension on the transport configuration. Please refer to the Azure Storage Queue Transport configuration documentation for more details.", ex);
            }
        }

尝试捕获的第一行正常运行，将消息添加到已配置的错误队列中，但是，这样做时，它似乎正在更改原始消息的消息ID和popreceipt，如下所示：

Initial Message Values

Updated Message Values

然后，当下一行尝试从输入队列中删除原始消息时，它无法找到它，因为根据本文https://docs.microsoft.com/en-us/rest/api/storageservices/delete-message2#remarks，它需要原始消息ID和pop接收器，它们现在已更改，从而导致以下错误被抛出：

2020-04-20 14:17:58,603 WARN : Azure Storage Queue transport failed pushing a message through pipeline
Type: Microsoft.WindowsAzure.Storage.StorageException
Message: The remote server returned an error: (404) Not Found.
Source: Microsoft.WindowsAzure.Storage
StackTrace:
   at Microsoft.WindowsAzure.Storage.Core.Executor.Executor.EndExecuteAsync[T](IAsyncResult result) in c:\Program Files (x86)\Jenkins\workspace\release_dotnet_master\Lib\ClassLibraryCommon\Core\Executor\Executor.cs:line 50
   at Microsoft.WindowsAzure.Storage.Core.Util.AsyncExtensions.<>c__DisplayClass7.<CreateCallbackVoid>b__5(IAsyncResult ar) in c:\Program Files (x86)\Jenkins\workspace\release_dotnet_master\Lib\ClassLibraryCommon\Core\Util\AsyncExtensions.cs:line 121
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at NServiceBus.Transport.AzureStorageQueues.MessageRetrieved.<Unwrap>d__3.MoveNext() in C:\BuildAgent\work\3c19e2a032c05076\src\Transport\MessageRetrieved.cs:line 40
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at NServiceBus.Transport.AzureStorageQueues.MessagePump.<InnerReceive>d__7.MoveNext() in C:\BuildAgent\work\3c19e2a032c05076\src\Transport\MessagePump.cs:line 153
TargetSite: T EndExecuteAsync[T](System.IAsyncResult)

这是NServiceBus软件包逻辑的问题，还是我的自定义消息展开逻辑中的某些问题导致这些值发生更改？

Answer 1

这是一个错误。当展开失败时，该消息尚未通过处理管道。结果，正常的可恢复性不适用。 CloudQueueMessage需要被“克隆”，并将克隆发送到错误队列，而原始消息用于将其从输入队列中删除。我在GitHub上提出了bug issue，您可以在此跟踪该过程。