我正在开发一个使用Kafka并应用 Saga 模式的POC。我能够创建一个快乐的路径，在其中我可以从一个主题消费消息，并通过其他Kafka主题将其发送到多个服务。现在我正在尝试处理消费消息期间的异常。
正如你所看到的，我有一个简单的OrderRequestEvent类型的消费者。这个消费者接收一条消息，然后用这个apiService验证它。如果验证失败，它应该抛出一个异常。我的想法是在 Saga 中捕获/处理这个异常，改变状态并将其发送到另一个主题进行可能的重试，而不最终确定 Saga 。
我的消费者来了：

public class OrderManagementSystemConsumer : IConsumer<OrderRequestEvent>
    {
        private readonly ITopicProducer<CustomerValidationRequestEvent> customerValidationResponseEvent;
        private readonly IApiService apiService;

        public OrderManagementSystemConsumer(
            ITopicProducer<CustomerValidationRequestEvent> customerValidationResponseEvent, 
            IApiService apiService)
        {
            this.customerValidationResponseEvent = customerValidationResponseEvent;
            this.apiService = apiService;
        }

        public async Task Consume(ConsumeContext<OrderRequestEvent> context)
        {
            ArgumentNullException.ThrowIfNull(context, nameof(context));

            if (await this.apiService.ValidateIncomingRequestAsync(context.Message))
                throw new ArgumentException("Something wrong just happened");
        }
    }

字符串
我的 Saga 中处理OrderRequestEvent的部分

Initially(
            When(OrderRequestedEvent)
                .Then(context => LogContext.Info?.Log("Initializing saga: {0}", context.Saga.CorrelationId))
                .InitializeSaga()
                .Then(context => LogContext.Info?.Log("Validating Customer: {0}", context.Saga.CorrelationId))
                .SendingToCustomerValidation().LogSaga()
                .TransitionTo(ValidatingCustomer));

型
我注意到，即使我的异常在使用者中被正确地抛出， Saga 仍在正常地继续。显然，在使用者中抛出的异常对 Saga 没有任何影响。