[JAVA] Try using Axon Framework

I happened to have the opportunity to implement an in-house business flow. It's like application → approval by the person concerned → approval by the final manager.

Normally, you can develop with CRUD architecture with MVC of Spring, but I want to make it a CQRS + Event Sourcing architecture that I have been interested in for a long time, so I will evaluate the Axon Framework that realizes it.

Axon Framework

• CQRS framework.
• Also provides Event Sourcing function.
• Event Sourcing is not required.
• Also provides a function to link with Spring Boot.
• But spring-boot-devtools is incompatible and cannot be used, and development efficiency is a little inferior.
• The latest version at the time of writing this article is 4.3.
• Apache License v2.0

In addition, from version 4, a server for Microservices environment called Axon Server has been added, and it seems that it is recommended to cooperate with Axon Server and Axon Framework, but this time I will not handle it because I want to understand Axon Framework.

architecture

Below is a diagram of the version 3 documentation, which is easier to understand (I think) than the latest diagram.

• Developer implements Command Handler, Event Handler, and Thin Data Layer. The UI has nothing to do with Axon, so I'll omit it.
• When the Command Handler sends an event according to the business logic, Axon records the event in the Event Store and calls the Event Handler corresponding to the event.
• Event Handler at the bottom of the figure generates data for query. What kind of data store you use is optional.

Processing sequence

An image understood from Axon documentation and samples.

command

1. Send commands to Aggregate via Command Bus provided by Axon.
2. An instance of Aggregate corresponding to the command is created. If the corresponding Aggregate event has already been recorded, the event will be applied (played) to the instance.
3. Aggregate receives the command and sends an event to Event Bus according to the state of Aggregate.
4. Event Bus records the event and sends it to Aggregate.
5. Aggregate changes its state to the content according to the event.

Query

1. The Event Processor detects that the event was recorded in the Event Store and sends the event to the Thin Data Layer.
2. The Thin Data Layer stores data in a format that is easy to query.
3. The query is sent to the Thin Data Layer via Query Bus.
4. Thin Data Layer retrieves data from the Data Store and returns it.

sample

First, let's implement a process that just CRUDs the application for evaluation.

1. Dependencies

Add axon-spring-boot-starter (1) to the Spring Boot application. Since Axon Server is not used this time, axon-server-connector is excluded. Otherwise, you will get an error or warning about connection failure to Axon Server while the application is running.

JPA and JDBC are provided as engines for accessing the Event Store in Axon Framework. Since the processing to the Event Store is hidden by the Axon Framework, which one to choose will be a trade-off with the implementation method of Thin Data Layer. This time, we have added spring-boot-starter-data-jpa (②) to the JPA engine, which requires minimal setup.

• In actual development, I want to use Jooq, which I personally like, so in that case I will use the JDBC engine.

pom.xml(Excerpt)

  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.2.6.RELEASE</version>
    <relativePath/>
  </parent>

  <properties>
    <java.version>11</java.version>
    <kotlin.version>1.3.71</kotlin.version>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.axonframework</groupId>
      <artifactId>axon-spring-boot-starter</artifactId> <!-- ① -->
      <version>4.3.2</version>
      <exclusions>
        <exclusion>
          <groupId>org.axonframework</groupId>
          <artifactId>axon-server-connector</artifactId>
        </exclusion>
      </exclusions>
    </dependency>

    <!-- Web -->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-thymeleaf</artifactId>
    </dependency>

    <!-- DB -->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-data-jpa</artifactId> <!-- ② -->
    </dependency>
    <dependency>
      <groupId>com.h2database</groupId>
      <artifactId>h2</artifactId>
      <scope>runtime</scope>
    </dependency>

２. Aggregate This is the central part of business logic in Axon Framework. Since Aggregate is said to be a transaction boundary in DDD, it is likely that events will be managed on an Aggregate basis.

2.1. Commands and events

In the command, specify the identifier of the Aggregate to be sent in the property to which @TargetAggregateIdentifier (1) was assigned. On the other hand, there is no such specification because the event is sent to the same Aggregate instance that received the command, or to the instance-independent Thin Data Layer.

message.kt

interface IInvoiceBody {
  val invoiceTitle: String
}

interface IInvoiceHeader {
  val invoiceId: UUID
}

interface IInvoice: IInvoiceHeader, IInvoiceBody

data class CreateInvoiceCommand(
  @field:TargetAggregateIdentifier // ①
  override val invoiceId: UUID,
  val body: IInvoiceBody): IInvoiceBody by body, IInvoice

data class UpdateInvoiceCommand(
  @field:TargetAggregateIdentifier // ①
  override val invoiceId: UUID,
  val body: IInvoiceBody): IInvoiceBody by body, IInvoice

data class RemoveInvoiceCommand(
  @field:TargetAggregateIdentifier // ①
  override val invoiceId: UUID
): IInvoiceHeader

data class InvoiceCreatedEvent(private val invoice: IInvoice): IInvoice by invoice

data class InvoiceUpdatedEvent(private val invoice: IInvoice): IInvoice by invoice

data class InvoiceRemovedEvent(
  override val invoiceId: UUID
): IInvoiceHeader

• The reason why we use interfaces and delegates a lot is because we want to break the transfer of property values from commands to events, and it does not mean that it will not work unless we do this.

２.2. Aggregate Assign @AggregateIdentifier (1) to the property corresponding to @TargetAggregateIdentifier of the command. @Aggregate (②) is an annotation for Spring linkage, and also has the function of @Component annotation.

Assign @CommandHandler (③) to the method that processes the command. The command to create a new Aggregate is received by the constructor. However, the default constructor is also required when processing subsequent commands, as it only replays the event after instantiation.

The event is received by the method with @EventSourcingHandler (④) and the Aggregate status is updated. Call AggregateLifecycle.markDeleted () at the end of Aggregate life event.

InvoiceAggregate.kt

@Aggregate // ②
class InvoiceAggregate(): IInvoice {
  @AggregateIdentifier // ①
  override lateinit var invoiceId: UUID
  override lateinit var invoiceTitle: String

  @CommandHandler // ③
  constructor(command: CreateInvoiceCommand): this() {
    AggregateLifecycle.apply(InvoiceCreatedEvent(command))
  }

  @EventSourcingHandler // ④
  fun on(event: InvoiceCreatedEvent) {
    invoiceId = event.invoiceId
    invoiceTitle = event.invoiceTitle
  }

  @CommandHandler // ③
  fun handle(command: UpdateInvoiceCommand) {
    AggregateLifecycle.apply(InvoiceUpdatedEvent(command))
  }

  @EventSourcingHandler // ④
  fun on(event: InvoiceUpdatedEvent) {
    invoiceTitle = event.invoiceTitle
  }

  @CommandHandler // ③
  fun handle(command: RemoveInvoiceCommand) {
    AggregateLifecycle.apply(InvoiceRemovedEvent(command.invoiceId))
  }

  @EventSourcingHandler // ④
  fun on(event: InvoiceRemovedEvent) {
    AggregateLifecycle.markDeleted() //⑤
  }
}

3. Thin Data Layer The event recorded in the Event Store is delivered to the method with @EventHandler (1). Please note that it is different from @EventSourcingHandler. The query is received by the method with @QueryHandler (②), the data corresponding to the condition is acquired, and it is returned.

InvoiceService.kt

@Service
class InvoiceService(val invoiceRepo: InvoiceRepository) {
  class AllInvoicesQuery

  @QueryHandler // ②
  fun handle(query: AllInvoicesQuery): List<InvoiceEntity> {
    return invoiceRepo.findAll()
  }
  @EventHandler // ①
  fun on(event: InvoiceCreatedEvent) {
    invoiceRepo.save(InvoiceEntity(event.invoiceId, event.invoiceTitle))
  }
  @EventHandler // ①
  fun on(event: InvoiceUpdatedEvent) {
    invoiceRepo.save(InvoiceEntity(event.invoiceId, event.invoiceTitle))
  }
  @EventHandler // ①
  fun on(event: InvoiceRemovedEvent) {
    invoiceRepo.deleteById(event.invoiceId)
  }
}

4. Controller This is a Controller in Spring MVC that translates an HTTP request into a command and sends it to the Command Bus or Query Bus using the Command Gateway or Query Gateway. Sending commands to CommandGateway has an asynchronous send method and a synchronous sendAndWait method. I'm using the PRG pattern here, but I'm using sendAndWait because there may be overtaking.

InvoiceController.kt

@Controller
class InvoiceController(val commandGateway: CommandGateway, val queryGateway: QueryGateway) {

  companion object {
    const val REDIRECT_URL = "${UrlBasedViewResolver.REDIRECT_URL_PREFIX}/"
  }

  data class InvoiceBody(override val invoiceTitle: String): IInvoiceBody
  data class InvoiceRequest(
    override val invoiceId: UUID,
    override val invoiceTitle: String
  ): IInvoiceHeader, IInvoiceBody

  @GetMapping("/")
  fun topPage(model: Model): String {
    val invoices = queryGateway.query(InvoiceService.AllInvoicesQuery(), MultipleInstancesResponseType(InvoiceEntity::class.java)).get()
    model.addAttribute("invoices", invoices)
    return "index"
  }

  @PostMapping("/invoices")
  fun createInvoice(invoice: InvoiceBody): String {
    commandGateway.sendAndWait<Any>(CreateInvoiceCommand(UUID.randomUUID(), invoice))
    return REDIRECT_URL
  }

  @PostMapping("/invoices/update")
  fun updateInvoice(invoice: InvoiceRequest): String {
    commandGateway.sendAndWait<Any>(UpdateInvoiceCommand(invoice.invoiceId, invoice))
    return REDIRECT_URL
  }

  @PostMapping("/invoices/delete")
  fun deleteInvoice(@RequestParam invoiceId: UUID): String {
    commandGateway.sendAndWait<Any>(RemoveInvoiceCommand(invoiceId))
    return REDIRECT_URL
  }
}

Trouble

I can't see the result of the command in the query

For example, in the sample, even if I created an application (/ invoices) and redirected to display the page (/), I could not get the data. When I debugged, the @CommandHandler method sent an event and the @EventSourcingHandler method was called immediately, but the @EventHandler method was a little later. Apparently, Axon's default Event Processor, TrackingEventProcessor, runs in a separate thread and monitors for event additions at the default value of 5 seconds.

It's probably a good idea to tune the monitoring cycle for large systems, but for smaller ones it's better to switch to another SubscribingEventProcessor.

@SpringBootApplication
class SampleAxonApplication {
  @Autowired
  fun configure(configurer: EventProcessingConfigurer) {
    configurer.usingSubscribingEventProcessors()
  }
}

With SubscribingEventProcessor, @EventHandler's method is called from the same thread as Controller, so you can avoid query overtaking.

Impressions

Since various investigations have not been completed yet, it is still unclear whether it can withstand actual development. For example

• When changing the implementation of Aggregate due to specification change, the event has already been recorded, so what should I do?
• If a bug records an incorrect event, can the event be modified or deleted?
• I want an operation history, but is there a mechanism to get a list of events?
• If you want to ensure uniqueness between Aggregates, such as the application name, it seems to check in the Data Store before sending the command from the Controller, but I'm a little worried if that alone can handle it.

And so on. I would like to investigate these in the future.