spring-cloud-stream-event-sourcing-testcontainers

The goal of this project is to create a Spring Boot application that manages users using Event Sourcing. So, besides the traditional create/update/delete, whenever a user is created, updated, or deleted, an event informing of this change is sent to Kafka. Furthermore, we will implement another Spring Boot application that listens to those events and saves them in Cassandra. Finally, we will use Testcontainers to perform end-to-end testing.

Note: In kubernetes-minikube-environment repository, it's shown how to deploy this project in Kubernetes (Minikube)

Proof-of-Concepts & Articles

On ivangfr.github.io, I have compiled my Proof-of-Concepts (PoCs) and articles. You can easily search for the technology you are interested in by using the filter. Who knows, perhaps I have already implemented a PoC or written an article about what you are looking for.

Additional Readings

• [Medium] Implementing a Kafka Producer and Consumer using Spring Cloud Stream
• [Medium] Implementing Unit Tests for a Kafka Producer and Consumer that uses Spring Cloud Stream
• [Medium] Implementing End-to-End testing for a Kafka Producer and Consumer that uses Spring Cloud Stream
• [Medium] Configuring Distributed Tracing with Zipkin in a Kafka Producer and Consumer that uses Spring Cloud Stream
• [Medium] Using Cloudevents in a Kafka Producer and Consumer that uses Spring Cloud Stream

Project Overview

flowchart TB
    subgraph users ["Users"]
        HTTP["REST Clients"]
        Browser["Browser"]
    end

    subgraph user-service ["user-service:9080\n(Spring Boot)"]
        USRestCtrl["UserController\n/api/users"]
        USSwagger["Swagger UI"]
        USService["UserService"]
        USRepo["UserRepository"]
        USEmitter["UserEmitter"]
    end

    MySQL[("MySQL\nusers")]

    subgraph event-service ["event-service:9081\n(Spring Boot)"]
        ESRestCtrl["UserEventController\n/api/events"]
        ESSwagger["Swagger UI"]
        ESService["UserEventService"]
        ESRepo["UserEventRepository"]
        ESListener["UserEventListener"]
    end

    Cassandra[("Cassandra\nuser_events")]

    subgraph infrastructure ["Infrastructure"]
        Kafka[("Kafka\ncom.ivanfranchin.userservice.user")]
        SchemaRegistry["Schema Registry"]
        Zipkin["Zipkin\nDistributed Tracing"]
    end

    HTTP -->|"CRUD operations"| USRestCtrl
    HTTP -->|"queries events"| ESRestCtrl
    Browser -->|"accesses"| USSwagger
    Browser -->|"accesses"| ESSwagger

    USRestCtrl -->|"calls"| USService
    USService -->|"calls"| USRepo
    USRepo -->|"saves"| MySQL
    USService -->|"sends events"| USEmitter
    USEmitter -->|"publishes to"| Kafka

    Kafka -->|"triggers"| ESListener
    ESListener -->|"calls"| ESService

    ESRestCtrl -->|"calls"| ESService
    ESService -->|"calls"| ESRepo
    ESRepo -->|"queries"| Cassandra
    ESRepo -->|"saves"| Cassandra

    USEmitter -->|"resolves schemas"| SchemaRegistry
    ESListener -->|"resolves schemas"| SchemaRegistry

Applications

• user-service

  Spring Boot Web Java application responsible for handling users. The user information is stored in MySQL. Once a user is created, updated or deleted, an event is sent to Kafka.

  • Endpoints:

       GET /api/users
       GET /api/users/{id}
      POST /api/users      -d {"email": "...", "fullName": "...", "active": ...}
       PUT /api/users/{id} -d {"email": "...", "fullName": "...", "active": ...}
    DELETE /api/users/{id}
    

  • Serialization format

    user-service can use JSON or Avro format to serialize data to the binary format used by Kafka. If we choose Avro, both services will benefit by the Schema Registry that is running as a Docker container. The serialization format to be used is defined by the value set to the environment variable SPRING_PROFILES_ACTIVE.

    Configuration	Format
    SPRING_PROFILES_ACTIVE=default	JSON
    SPRING_PROFILES_ACTIVE=avro	Avro

• event-service

  Spring Boot Web Java application responsible for listening events from Kafka and saving them in Cassandra.

  • Endpoints:

    GET /api/events?userId={userId} - Get events by user id
    

  • Deserialization

    Unlike user-service, event-service does not have specific Spring profile to select the deserialization format. Spring Cloud Stream provides a stack of MessageConverters that handle the conversion of many content-types, including application/json. Besides, as event-service has a bean registered, auto-configures an Apache Avro message converter for schema management.

Prerequisites

• Java 25 or higher;
• A containerization tool (e.g., Docker, Podman, etc.)

Start Environment

• In a terminal and inside the spring-cloud-stream-event-sourcing-testcontainers root folder run:

  docker compose up -d
  

• Wait for Docker containers to be up and running. To verify, run:

  docker ps -a
  

Running Applications with Maven

• user-service

  • In a terminal, make sure you are inside the spring-cloud-stream-event-sourcing-testcontainers root folder;

  • In order to run the application, you can pick between JSON or Avro:

    • Using JSON

      ./mvnw clean spring-boot:run --projects user-service
      

    • Using Avro

      ./mvnw clean spring-boot:run --projects user-service -Dspring-boot.run.profiles=avro
      

• event-service

  • In a new terminal, make sure you are inside the spring-cloud-stream-event-sourcing-testcontainers root folder;

  • Run the following command:

    ./mvnw clean spring-boot:run --projects event-service
    

Running Applications as Docker containers

• Build Docker Images

  • In a terminal, make sure you are inside the spring-cloud-stream-event-sourcing-testcontainers root folder;

  • Run the following script to build the Docker images:

    ./build-docker-images.sh
    

• Environment Variables

  • user-service

    Environment Variable	Description
    MYSQL_HOST	Specify host of the MySQL database to use (default localhost)
    MYSQL_PORT	Specify port of the MySQL database to use (default 3306)
    KAFKA_HOST	Specify host of the Kafka message broker to use (default localhost)
    KAFKA_PORT	Specify port of the Kafka message broker to use (default 29092)
    SCHEMA_REGISTRY_HOST	Specify host of the Schema Registry to use (default localhost)
    SCHEMA_REGISTRY_PORT	Specify port of the Schema Registry to use (default 8081)
    ZIPKIN_HOST	Specify host of the Zipkin distributed tracing system to use (default localhost)
    ZIPKIN_PORT	Specify port of the Zipkin distributed tracing system to use (default 9411)

  • event-service

    Environment Variable	Description
    CASSANDRA_HOST	Specify host of the Cassandra database to use (default localhost)
    CASSANDRA_PORT	Specify port of the Cassandra database to use (default 9042)
    KAFKA_HOST	Specify host of the message broker to use (default )