Нет EntityManager с фактической транзакцией при публикации события в Saga

Question

Я использую конфигурацию по умолчанию "axon 4.0.3 + Spring Boot 2 + Spring Data (PostgreSQL)".

Публикуя событие в EventStore и ожидая, что оно будет перехвачено @SagaEventHandler, который я получилследующее исключение:

javax.persistence.TransactionRequiredException: нет EntityManager с фактической транзакцией, доступной для текущего потока, - невозможно надежно обработать вызов «persist» в org.springframework.orm.jpa.SharedEntityManagerCreator $ SharedEntityMankeInination(SharedEntityManagerCreator.java:292) ~ [spring-orm-5.1.5.RELEASE.jar: 5.1.5.RELEASE] в com.sun.proxy. $ Proxy104.persist (неизвестный источник) ~ [na: na] в java.util.stream.ForEachOps $ ForEachOp $ OfRef.accept (ForEachOps.java:184) ~ [na: 1.8.0_191] в java.util.stream.ReferencePipeline $ 3 $ 1.accept (ReferencePipeline.java:193) ~ [na:1.8.0_191] в java.util.ArrayList $ ArrayListSpliterator.forEachRemaining (ArrayList.java:1382) ~ [na: 1.8.0_191] в java.util.stream.AbstractPipeline.copyInto (AbstractPipeline.java:481) ~ [na: 1.8.0_191] в java.util.stream.AbstractPipeline.wrapAndCopyInto (AbstractPipeline.java:471) ~ [na: 1.8.0_191] в java.util.stream.ForEachOps $ ForEachOp.valuSequential (ForEachOps.java:151) ~ [na: 1.8.0_191] в java.util.stream.ForEachOps $ ForEachOp $ OfRef.evaluateSequential (ForEachOps.java:174) ~ [na: 1.8.0_191] в java.util.stream.AbstractPipeline.evaluate (AbstractPipeline.java:234) ~ [na: 1.8.0_191] в java.util.stream.ReferencePipeline.forEach (ReferencePipeline.java:418) ~ [na: 1.8.0_191] в org.axonframework.eventsourcing.eventstore.jpa.) ~ [axon-eventsourcing-4.0.3.jar: 4.0.3]

Какая дополнительная конфигурация требуется для EventStore для обработки этого случая?

PS.Добавление @Transactional к методу решает эту проблему, но я не понимаю, почему это необходимо.

Пример минимального кода (следующая конечная точка 127.0.0.1:8080/1 работает, но другая 127.0.0.1:8080/ 1 нет):

@SpringBootApplication
class TestAxonApplication

class UserId(val userId: String = IdentifierFactory.getInstance().generateIdentifier()) : Serializable

class TestCommand(@TargetAggregateIdentifier val userId: UserId)

class TestedEvent(val userId: UserId)

fun main(args: Array<String>) {
    runApplication<TestAxonApplication>(*args)
}

@RestController
@RequestMapping
class Controller(var commandGateway: CommandGateway, var eventStore: EventStore) {

    @GetMapping("/1")
    fun done(): UserId? {
        return commandGateway.sendAndWait<UserId>(TestCommand(UserId()))
    }

    @GetMapping("/2")
    fun failure() {
        eventStore.publish(
                GenericEventMessage.asEventMessage<Void>(
                        TestedEvent(UserId())
                )
        )
    }

}

@Aggregate
class User() {

    @AggregateIdentifier
    private lateinit var userId: UserId

    @CommandHandler
    constructor(cmd: TestCommand) : this() {
        AggregateLifecycle.apply(TestedEvent(cmd.userId))
    }

    @EventHandler
    fun on(event: TestedEvent) {
        this.userId = event.userId
    }

}

@Saga
@ProcessingGroup("mySaga")
class MySaga {

    @StartSaga
    @SagaEventHandler(associationProperty = "userId")
    fun start(event: TestedEvent) {
        println("DONE ${event.userId.userId}")
    }

}

Answer 1

разница между вызовами заключается в том, что один из них проходит через командную шину, а другой пропускает его и публикуется непосредственно в шину событий. По умолчанию TransactionManager настроено на Командной шине. Однако это не относится к шине событий.

Это означает, что вы публикуете событие без активной транзакции. Hibernate это не нравится.

Решение состоит в том, чтобы поместить @Transactional в конечную точку, чтобы гарантировать, что транзакция активна при хранении событий.

Answer 2

Похоже, вы забыли настроить командную шину, чтобы разрешить управление транзакциями.Добавьте его, как в примере ниже, и он будет работать:

@Bean
public CommandBus commandBus(TransactionManager transactionManager) {
    return new SimpleCommandBus(transactionManager, NoOpMessageMonitor.INSTANCE);
}

Обновление

Моя ошибка, в первую очередь я использую для работы с Axon 3.3 (непоследние), и я подумал, что вы используете пользовательскую конфигурацию для хранилища событий.

При загрузке аксона по умолчанию используется хранилище событий памяти, и поэтому, если вы не определите пользовательское хранилище, транзакции не будут работать.

Это то, чего не хватает в вашей конфигурации:

@Bean
public EventStorageEngine eventStorageEngine(EntityManagerProvider entityManagerProvider, TransactionManager transactionManager) {
    return new JpaEventStorageEngine(entityManagerProvider, transactionManager);
}

@Bean
public PlatformTransactionManager transactionManager(EntityManagerFactory emf) {
    return new JpaTransactionManager(emf);
}

´´´

Это простой пример, который я сделал для других предложений, и он отлично работает.

Помните

Другим интересным моментом является то, что вы должны определить агрегатные репозитории, потому что Axon пытается найти репозиторий, который соответствует определенному агрегату.

@Bean
public Repository<User> documentAggregateRepository(EventStore eventStore) {
    return new EventSourcingRepository<>(User.class, eventStore);
}

То же самое для саг, вы должны зарегистрировать их илиsaga никогда не будет триггером

// remember to call the method sagaName + Configuration
// or you must set up the @Saga configurationBean name pointing this method
@Bean
public SagaConfiguration<MySaga> mySagaConfiguration() {
    return SagaConfiguration.subscribingSagaManager(MySaga.class);
}

В этом примере источник событий, но если вы интересуетесь устаревшими миграциями, возможно, вы будете использовать GenericJpaRepository во время процесса миграции (если вам нужен пример, просто дайте мне знать).

HTH.