MassTransit Saga状态机接收Kafka响应后偶发事件不触发

使用MassTransit Saga状态机与Kafka编排数据流时，遇到偶发异常：接收Kafka主题的ProducerValidationResult响应后，状态机对应的事件未触发。

Saga配置 #

private static void RiderConfiguration(IRiderRegistrationConfigurator config, Configuration configuration)
{
    config.AddInMemoryInboxOutbox();

    IMongoDatabase db = AddMongoSagaDatabase(configuration.MongoConfiguration);

    config.AddSagaStateMachine<PipelineStateMachine, PipelineRunState>()
        .MongoDbRepository(db, cfg =>
        {
            cfg.DatabaseName = configuration.MongoConfiguration.DatabaseName;
        });    
}

端点配置 #

private static IKafkaFactoryConfigurator AddConsumerToBuilder<TMessage>(
        IKafkaFactoryConfigurator configurator,
        IRiderRegistrationContext context,
        string topicName,
        string groupId,
        Configuration configuration)
        where TMessage : class
    {
        configurator.TopicEndpoint<string, TMessage>(
            topicName,
            new ConsumerConfig() { GroupId = groupId, AutoOffsetReset = AutoOffsetReset.Latest },
            endpointConfig =>
            {
                endpointConfig.SetValueDeserializer(new JsonDeserializer<TMessage>());

                endpointConfig.UseMessageRetry(r =>
                {
                    r.Immediate(configuration.KafkaConfiguration.RetryCount);
                    r.Handle<MongoDbConcurrencyException>();
                });
                endpointConfig.UseInMemoryOutbox(context);

                endpointConfig.ConfigureSaga<PipelineRunState>(context);
            });
        return configurator;
    }

生产者配置 #

private static void AddProducerToBuilder<TMessage>(this IRiderRegistrationConfigurator config, string topicName)
     where TMessage : class
 {
     config.AddProducer<string, TMessage>(
         topicName,
         (ctx, cfg) =>
         {
             cfg.BatchNumMessages = 1;
             cfg.Linger = TimeSpan.FromMilliseconds(1);
             cfg.SetValueSerializer(new JsonSerializer<TMessage>());
         });
 }

状态机事件定义 #

InstanceState(config => config.CurrentState);

Initially(
    When(FundamentalDataReceived)
        .Then(context => TransitionToState(context, Initializing))
        .Activity(a => a.OfType<InitializationActivity>()));

OnUnhandledEvent(async e => await e.Ignore());

DuringAny(
    When(ProducerValidationResult)
        .IfElse(
            ctx => ctx.Message.IsSuccess,
            successBinder => successBinder
                .Then(ctx => LogContext.Debug?.Log("[{CorrelationId}]ProducerValidation.Result.Success", ctx.CorrelationId))
                .Then(ctx => TransitionToState(ctx, Initialized))
                .Activity(a => a.OfInstanceType<RequestMovementActivity>()),
            failedBinder => failedBinder
                .Then(ctx => LogContext.Debug?.Log("[{CorrelationId}]ProducerValidation.Result.Fail", ctx.CorrelationId))
                .Then(ctx => TransitionToState(ctx, ProducerValidationFailed))
                .Activity(a => a.OfInstanceType<RequestQuarantineActivity>())));

流程说明 #

接收到基础数据后，InitializationActivity向Kafka主题发送消息；对应服务处理完成后返回ProducerValidationResult响应。多数场景下流程正常，但偶发接收响应后状态机事件未触发。

已启用调试日志，查阅官方文档示例后仍未定位问题根源，恳请有相关经验的开发者提供排查思路。

事件未触发的日志记录 #

状态机日志 #

生产者验证服务日志 #

内容的提问来源于stack exchange，提问作者user26706031