EventStore Configuration with Postgres [Axon 4]

Imported from Google Groups
1

The application was previously configured correctly but something changed and I’m slightly struggling in determining the root cause (it’s also getting late so plan to restart tomorrow).

`

2019-01-24 17:00:59.005 ERROR 15490 — [tTaskScheduler3] o.s.s.s.TaskUtils$LoggingErrorHandler : Unexpected error occurred in scheduled task.

org.axonframework.eventsourcing.eventstore.EventStoreException: An event with identifier [c668cd0e-6f3c-433f-a7f7-da58eed6bd2a] could not be persisted
at org.axonframework.eventsourcing.eventstore.AbstractEventStorageEngine.handlePersistenceException(AbstractEventStorageEngine.java:125) ~[axon-eventsourcing-4.0.jar:4.0]
at org.axonframework.eventsourcing.eventstore.jpa.JpaEventStorageEngine.appendEvents(JpaEventStorageEngine.java:281) ~[axon-eventsourcing-4.0.jar:4.0]
at org.axonframework.eventsourcing.eventstore.AbstractEventStorageEngine.appendEvents(AbstractEventStorageEngine.java:98) ~[axon-eventsourcing-4.0.jar:4.0]
at org.axonframework.eventsourcing.eventstore.AbstractEventStore.prepareCommit(AbstractEventStore.java:63) ~[axon-eventsourcing-4.0.jar:4.0]
at org.axonframework.eventhandling.AbstractEventBus.publish(AbstractEventBus.java:131) ~[axon-messaging-4.0.jar:4.0]
at org.axonframework.eventhandling.EventBus.publish(EventBus.java:51) ~[axon-messaging-4.0.jar:4.0]
at com.aramark.ess.command.domain.SyncScheduler$SyncExecutor.run(SyncScheduler.kt:64) ~[classes/:na]
at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54) ~[spring-context-5.0.9.RELEASE.jar:5.0.9.RELEASE]
at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:93) [spring-context-5.0.9.RELEASE.jar:5.0.9.RELEASE]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_162]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_162]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) [na:1.8.0_162]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) [na:1.8.0_162]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_162]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_162]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_162]
Caused by: javax.persistence.TransactionRequiredException: No EntityManager with actual transaction available for current thread - cannot reliably process ‘persist’ call
at org.springframework.orm.jpa.SharedEntityManagerCreator$SharedEntityManagerInvocationHandler.invoke(SharedEntityManagerCreator.java:289) ~[spring-orm-5.0.9.RELEASE.jar:5.0.9.RELEASE]
at com.sun.proxy.$Proxy133.persist(Unknown Source) ~[na:na]
at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184) ~[na:1.8.0_162]
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193) ~[na:1.8.0_162]
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1382) ~[na:1.8.0_162]
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481) ~[na:1.8.0_162]
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471) ~[na:1.8.0_162]
at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151) ~[na:1.8.0_162]
at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174) ~[na:1.8.0_162]
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[na:1.8.0_162]
at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418) ~[na:1.8.0_162]
at org.axonframework.eventsourcing.eventstore.jpa.JpaEventStorageEngine.appendEvents(JpaEventStorageEngine.java:276) ~[axon-eventsourcing-4.0.jar:4.0]
… 14 common frames omitted

`

The above exception is fired from
`

eventStore.publish(asEventMessage<Any>(event))

`

As stated in the title we’re using Postgres and I see events applied to aggregates in domain_event_entry.
This event is fired from a standalone component with it’s intended target being a Saga. The associatedId is defined and the event handler in the Saga is decorated with a @StartSaga

Finally here is the AxonConfiguration currently in place:

`

@Configuration
@AutoConfigureAfter(AxonAutoConfiguration::class)
class AxonFrameworkConfiguration {
    private val log: Logger = LoggerFactory.getLogger(this.javaClass)

    @Autowired
    fun registerInterceptors(commandBus: CommandBus) {
        commandBus.registerDispatchInterceptor(BeanValidationInterceptor())
    }

    @Bean
    fun snapshotterFactoryBean() = SpringAggregateSnapshotterFactoryBean()

    @Bean
    fun eventSchemaFactory(): EventTableFactory {
        return PostgresEventTableFactory.INSTANCE
    }

    @Bean
    fun eventSchema(): EventSchema {
        return EventSchema()
    }

    @Bean
    fun sagaSqlSchema(): SagaSqlSchema {
        return PostgresSagaSqlSchema()
    }
}

`

application yml

`

server:
  port: 8081
spring:
  application:
    name: ${SERVICE_NAME}
  datasource:
    url: jdbc:postgresql://${POSTGRES_HOST_DEV}:5432/${SERVICE_NAME}

    username: ${POSTGRES_USER_DEV}
    password: ${POSTGRES_PASSWORD_DEV}
    driver-class-name: org.postgresql.Driver
  jpa:
    hibernate:
      ddl-auto: create-drop
    database-platform: org.hibernate.dialect.PostgreSQLDialect
axon:
  snapshot:
    trigger:
      treshold:
        synchost: 100
        syncmodel: 100
        syncpoint: 100
  kafka:
    default-topic: axon-events
    producer:
      retries: 5
      transaction-id-prefix: esstx
    consumer:
      group-id: external-sync-group
    client-id: ${random.uuid}
    bootstrap-servers: ${KAFKA_URL}:9092
  distributed:
    enabled: true
  serializer:
    messages: jackson
  eventhandling:
    processors:
      syncsaga:
        mode: tracking
        source: kafkaMessageSource

`

Thank you in advance for any feedback I could get to help understand where I’ve misconfigured or misunderstood the framework.

2

I should note that I’ve attempted to perform a publish to the eventbus vs the event store to the same effect and accidentally pasted slightly stale code although the error is the same.

The event is being fired from a scheduled runnable therefore a different thread. I don’t know if this has an impact.

3

“Caused by: javax.persistence.TransactionRequiredException: No EntityManager with actual transaction available for current thread”.

Perhaps just need to ensure the scheduled runnable executes with a transaction? Something along the lines of…

@Scheduled(cron = "0 0 0 * * *", zone="GMT")
@Transactional
public void processXXX() {
    LOG.info("Processing...");

    ... Axon code

    ...
4

You may be right but I believe I’m still missing something. Here is the entire class and the respective scheduler.

@Component
@ProcessingGroup("sync-point")
internal class SyncScheduler(private val eventBus: EventBus,
                             private val taskScheduler: TaskScheduler,
                             private val syncpointRepository: Repository<SyncPoint>) {

    @Autowired
    lateinit var syncmodelService: SyncModelService

    @PostConstruct
    fun onLoad() {
        // TODO:
    }

    @EventHandler
    @AllowReplay(false)
    fun handle(event: SyncPointCreatedEvent) {
        syncpointRepository.load(event.id).execute {
            val syncModel = syncmodelService.findById(it.syncModelId)
            schedule(SyncExecutor(syncPointId = event.id,
                                  syncModelId = syncModel.id,
                                  syncHostId = syncModel.syncHostId),
                     syncModel.cronSchedule)
        }
    }

    @Transactional
    private inner class SyncExecutor(private val syncPointId: String,
                                     private val syncModelId: String,
                                     private val syncHostId: String): Runnable {
        @Transactional
        override fun run() {
            println("Sync Triggered for $syncPointId")
            val event = SyncPointTriggeredEvent(
                    UUID.randomUUID().toString(),
                    syncHostId,
                    syncModelId,
                    syncPointId)
            eventBus.publish(asEventMessage<Any>(event)) // CRASH HAPPENS HERE!
        }
    }

    fun schedule(task: Runnable, cronExpression: String) {
        taskScheduler.schedule(task, CronTrigger(cronExpression))
    }
}
5

I could imagine that - as you are using a private inner class and @Transactional is AOP based - Spring is not be able to build a dynamic proxy and intercept the call and thus will not be able to open a transaction.

6

I think Michael L is right on that one. I think your @Transactional annotations are not placed where Spring has any influence.

Also, EventHandling itself is already asynchronous. Consider doing these calls synchronously to simplify things for yourself.