使用Subscribing Event Processor时异常发生后事件未回滚问题

我采用Oracle作为事件存储，使用Subscribing Event Processor处理事件。当事件处理器抛出异常时，期望能回滚事件存储，但该逻辑未生效。

已查阅资料：

• Axon FrameWork: How to rollback in the domain_event_entry table
• Spring boot Axon complete rollback
• Axon 3-Spring Boot Transaction Management
• Axon官方文档：Subscribing Event Processor文档

环境配置（Axon 4.9 + Spring） #

Axon配置类 #

@Configuration
public class AxonConfig {
  // 省略其他配置方法...
  public void configureProcessorDefault(EventProcessingConfigurer processingConfigurer) {
    processingConfigurer.usingSubscribingEventProcessors();
    processingConfigurer.registerDefaultListenerInvocationErrorHandler(configuration -> PropagatingErrorHandler.INSTANCE);
  }

  @Bean
  public CommandBus commandBus(MeterRegistry meterRegistry) {
    TransactionManager transactionManager = springTransactionManager();
    SimpleCommandBus commandBus = SimpleCommandBus
      .builder()
      .transactionManager(transactionManager)
      .rollbackConfiguration(RollbackConfigurationType.ANY_THROWABLE)
      .build();
    commandBus.registerHandlerInterceptor(new TransactionManagingInterceptor<>(transactionManager));
    return commandBus;
  }

  @Bean public EventStorageEngine eventStorageEngine() {
    return JdbcEventStorageEngine
    .builder()
    .snapshotSerializer(xStreamSerializer)
    .upcasterChain(upcasterChain)
    .persistenceExceptionResolver(persistenceExceptionResolver())
    .eventSerializer(eventSerializer)
    .connectionProvider(new SpringDataSourceConnectionProvider(dataSource))
    .transactionManager(springTransactionManager())
    .build();
  }
}

聚合根代码 #

public class Aggregate {
  @CommandHandler
  public void handle(MyCommand command,
    UnitOfWork parent) {
    apply(new CommandCalled());
  }
}

投影处理器代码 #

public class ProjectionHandler {
    @EventHandler
    public void on(CommandCalled event,
      UnitOfWork child) {
      throw new RuntimeException();
    }
}

命令发送代码 #

commandGateway.sendAndWait(new MyCommand());

我发现事件处理器中的UnitOfWork父实例是命令处理器中的UnitOfWork，尝试调用CurrentUnitOfWork.get().rollback()但未生效。

核心问题分析 #

订阅型事件处理器默认复用命令处理的UnitOfWork，但事件存储的提交时机早于事件处理器执行：

1. 命令处理阶段，聚合根apply()生成的事件会先写入domain_event_entry表，这一步在命令UnitOfWork的提交阶段完成。
2. 事件处理器在命令UnitOfWork提交后才执行，此时事件已持久化，后续触发的回滚无法撤销事件存储的写入。

具体修复步骤 #

1. 为事件处理器配置独立事务 #

修改AxonConfig中的处理器配置，让事件处理器在独立事务中执行，确保异常能触发自身事务回滚：

public void configureProcessorDefault(EventProcessingConfigurer processingConfigurer) {
    processingConfigurer.usingSubscribingEventProcessors();
    // 注册事务拦截器，为事件处理器绑定独立事务
    processingConfigurer.registerDefaultHandlerInterceptor(configuration -> 
        new TransactionManagingInterceptor<>(springTransactionManager())
    );
    // 保留传播错误处理器，确保异常能触发事务回滚
    processingConfigurer.registerDefaultListenerInvocationErrorHandler(configuration -> PropagatingErrorHandler.INSTANCE);
}

2. 简化命令总线事务配置（可选） #

当前命令总线的TransactionManagingInterceptor属于重复配置，SimpleCommandBus已通过transactionManager()设置事务管理器，可简化代码：

@Bean
public CommandBus commandBus(MeterRegistry meterRegistry) {
    TransactionManager transactionManager = springTransactionManager();
    return SimpleCommandBus.builder()
            .transactionManager(transactionManager)
            .rollbackConfiguration(RollbackConfigurationType.ANY_THROWABLE)
            .build();
}

3. 若需回滚事件存储，同步执行事件处理器 #

如果必须让事件处理器的异常触发事件存储回滚，需将事件处理纳入命令事务的同步执行流程，让异常能传播回命令事务：

public class Aggregate {
  @CommandHandler
  public void handle(MyCommand command, UnitOfWork parent) {
    CommandCalled event = new CommandCalled();
    apply(event);
    // 强制同步执行事件处理器，让异常传播回命令事务
    parent.attachTransactionSynchronization(() -> {
        EventBus eventBus = parent.getResource(EventBus.class);
        eventBus.publish(event, parent);
    });
  }
}

关键注意事项 #

• 订阅型处理器默认异步执行事件处理（命令事务提交后），此时异常无法影响命令事务，仅能回滚自身投影事务。
• 若业务允许事件存储与投影处理分离（事件存储成功但投影失败时，通过重试机制修复），推荐使用独立事务的订阅型处理器，无需回滚事件存储。
• 若需保证事件存储与投影的强一致性，也可改用Tracking Event Processor配合事务管理，或采用上述同步执行的方式。

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