A service is meant to work with others through events, but Kafka is not available yet. Using a billing service from a bike sharing system, this page shows how to build and test it anyway: with a receiver and a handler for each incoming event, an interface in the aggregate for its own events, and tests on three levels. Later, Kafka only adds adapters, and no class has to change.

Language

Deutsche Fassung

Example code repository

https://gitlab.com/archi-lab/public/learning/learn-event-driven-architecture/-/tree/main/service-development-2-with-interfaces-and-tests

Keywords

Event-driven architecture, Microservices, Kafka, Testing without a broker, Receiver, Handler, Orchestrator, Test double, Integration test, Tolerant reader, Payload

• Developing and testing a service without a broker
  • The example: billing in a bike sharing system
  • One event in, one event out
  • What is added to the core
  • The structure
• Consuming events (and reacting to them)
  • The receiver
  • Who calls the receiver
  • From the JSON payload to a domain type for the event data
  • The handler for the event
• Publishing events
  • Outgoing (produced) Events usually (though not always) belong to aggregates
  • Testing without a broker
  • Later with Kafka
  • Running the module

Developing and testing a service without a broker

English version. The German README.de.md is the source of truth and is the text embedded by the archi-lab.io infopage.

A service is meant to work with others through events, but the shared environment with Kafka is not up yet. The question of this module: how do you build it so that it runs and can be tested without Kafka from the very start, and can still be connected to Kafka later without a rebuild?

The starting point is the module service-development-1-core-functionality: the business logic of a billing service, but with no way out and no tests. This module adds three things: an entry point for the events of another service, a way out for its own events, and tests on three levels. The code needs no Kafka at all: there is neither the library nor a broker in the project. That is on purpose, because a team should be able to start with a clean codebase before the shared environment is up.

The example: billing in a bike sharing system

The example continues the module asyncapi-event-contracts. The billing service reads from the topic rentals, into which the rental service writes its events. It needs only RentalEnded and reacts to it in two ways:

• It prices the ride: 50 cents for every started quarter of an hour, and then publishes a ChargeCreated event.
• It counts the finished rides per customer, as the basis for a loyalty discount.

RentalStarted is in the same topic but is not needed: there is no receiver for it.

One event in, one event out

Seen from the outside, the functionality of this module is a reaction: the consumed event RentalEnded from the rental service triggers it, and it ends with the produced event ChargeCreated. Everything in between happens inside the billing service. It does not call the rental service, and it does not need to know who reads ChargeCreated.

flowchart LR
    Rental["Rental service"]:::outside
    Topic[("Topic rentals")]:::outside
    subgraph Billing["Billing service"]
        Handler["RentalEndedHandler"]:::new
        Charge["charge:<br/>ChargingService<br/>price<br/>the ride"]:::existing
        Ride["ride:<br/>RideCountService<br/>count<br/>the ride"]:::existing
        Events["charge:<br/>ChargeEvents<br/>(interface)"]:::new
        Handler --> Charge
        Handler --> Ride
        Charge --> Events
    end
    Others["other<br/>services"]:::outside

    Rental --> Topic
    Topic -- "RentalEnded<br/>(consumed)" --> Handler
    Events -- "ChargeCreated<br/>(produced)" --> Others

    classDef existing fill:#e8f5e9,stroke:#2e7d32,stroke-width:1px
    classDef new fill:#fff3cd,stroke:#ff9800,stroke-width:3px
    classDef outside fill:#e3f2fd,stroke:#1565c0,stroke-width:1px,stroke-dasharray:5 4
    style Billing fill:#fafafa,stroke:#9e9e9e

flowchart LR
    %% P1 and P2 are invisible spacers: they give the legend the width of the diagram above,
    %% so that both are scaled alike and the legend stays centered
    P1["_________________________________________________"]:::spacer ~~~ L1["existing class<br/>(from module 1)"]:::existing ~~~ L2["new class"]:::new
    L2 ~~~ L3["outside,<br/>mocked in tests"]:::outside ~~~ P2["_________________________________________________"]:::spacer

    classDef existing fill:#e8f5e9,stroke:#2e7d32,stroke-width:1px
    classDef new fill:#fff3cd,stroke:#ff9800,stroke-width:3px
    classDef outside fill:#e3f2fd,stroke:#1565c0,stroke-width:1px,stroke-dasharray:5 4
    classDef spacer fill:none,stroke:none,color:transparent

There is no broker in this module yet: RentalEnded arrives in the tests, and ChargeCreated ends up in the test double. The flow in between is already the same as later with Kafka.

What is added to the core

The business logic stays as it is: Charge with the price rule, RideCount, the repositories and the RideCountService are the same in both modules. New are:

• The interface ChargeEvents, which the ChargingService calls after saving. This is the only change to an existing class.
• The package eventhandlers with the handler, the record RentalEnded and, in infrastructure, the receiver for the event RentalEnded.
• The tests and the test double InMemoryChargeEvents.

Comparing the two module folders side by side shows exactly this step.

The structure

The service has two aggregates, charge and ride, each with domain and application. Beside them is the package eventhandlers for the reactions to incoming events:

thkoeln.archilab.servicedevelopment
├── eventhandlers                 new
│   ├── RentalEnded               the payload in domain types
│   ├── RentalEndedHandler        the reaction to RentalEnded
│   └── infrastructure
│       ├── RentalEndedReceiver   the entry point for RentalEnded
│       └── RentalEventMapper     reads the payload
├── charge
│   ├── domain                    Charge (with the price rule), ChargeRepository
│   └── application               ChargingService, ChargeEvents (new)
└── ride
    ├── domain                    RideCount, RideCountRepository
    └── application               RideCountService

An incoming event often belongs to no single aggregate: RentalEnded concerns two here. So the reaction does not live in one of the aggregates but in a handler above them. The path of a message:

Topic rentals
    |    later: a thin adapter for Kafka, not yet in the module
    v
RentalEndedReceiver.receive(payload as JSON)         <- our code starts here
    |    read the payload and turn it into domain types: RentalEnded
    v
RentalEndedHandler.handle(RentalEnded)
    |    call the aggregates in the right order
    v
ChargingService, RideCountService
    |
    v
Interface ChargeEvents  ->  in the tests: InMemoryChargeEvents      later: a Kafka producer

Consuming events (and reacting to them)

The receiver

There is one receiver for each incoming event. It knows only this one event:

@Component
public class RentalEndedReceiver {
    public void receive( String payloadJson ) {
        rentalEndedHandler.handle( mapper.toRentalEnded( payloadJson ) );
    }
}

The receiver gets only the payload as JSON and knows no broker client. Which receiver is responsible for a message is decided by whoever receives the messages: later an adapter for Kafka that reads the event type from the header, until then the tests. An event without a receiver is ignored.

Why is the receiver located in the infrastructure package? It is, at first glance, technology-agnostic. This means it is not tied to a specific technology such as Kafka or RabbitMQ. It does, however, assume that the event’s payload arrives as JSON. This makes it part of the infrastructure - that is, something that contains information about a specific technology used to communicate with the outside world.

Who calls the receiver

In Spring, you expect a listener to carry an annotation such as @KafkaListener or @RabbitListener, through which Spring calls the method as soon as a message arrives. That is why RentalEndedReceiver is not called a listener: it has no such annotation. It is an ordinary method, and that is on purpose: this is where any source of messages plugs in, in this module the tests, later an adapter for Kafka, just as well one for RabbitMQ. The annotation sits on the adapter, and this module needs no broker. So the receiver knows the format of the other service’s payload, but no broker. The handler above it knows only domain types.

flowchart LR
    Test["RentalEndedReceiverIT<br/>(test, this module)"]:::new
    Kafka["RentalsKafkaListener<br/>and RentalEndedSpecification<br/>(Kafka, module 3)"]:::later
    Rabbit["Method with<br/>@RabbitListener<br/>(RabbitMQ, possible)"]:::later
    Receiver["RentalEndedReceiver<br/>receive(payloadJson)"]:::new
    Handler["RentalEndedHandler<br/>handle(RentalEnded)"]:::new

    Test -- "payload as JSON" --> Receiver
    Kafka -- "payload as JSON" --> Receiver
    Rabbit -- "payload as JSON" --> Receiver
    Receiver -- "domain types" --> Handler

    classDef new fill:#fff3cd,stroke:#ff9800,stroke-width:3px
    classDef later fill:#f5f5f5,stroke:#9e9e9e,stroke-width:1px,stroke-dasharray:5 4

A solid border marks what exists in this module, a dashed one what comes later or would be possible. All sources deliver the same thing: the payload as JSON.

How much the adapter has to do depends on the broker. With RabbitMQ, each event type usually gets a queue of its own, and a method with @RabbitListener( queues = "billing.rental-ended" ) that calls receive is enough. With Kafka, several event types share a topic: rentals also holds RentalStarted, so that all events of one rental stay in order. A @KafkaListener receives everything in the topic, and someone has to dispatch by event type.

From the JSON payload to a domain type for the event data

The receiver first has the RentalEventMapper read the JSON. The result is the record RentalEnded. This record maps the JSON payload (which at first is just a string) to a domain type of our own. A simple record is enough here, because it really is just a data container. It contains exactly the information from the event that we need in our context “Billing”. This record is passed to the handler (shortened here):

private record RentalEndedPayload( UUID rentalId, UUID customerId, Instant startedAt, Instant endedAt ) {}

public RentalEnded toRentalEnded( String payloadJson ) {
    RentalEndedPayload payload = json.readValue( payloadJson, RentalEndedPayload.class );
    Duration rideDuration = Duration.between( payload.startedAt(), payload.endedAt() );
    return new RentalEnded( payload.rentalId(), payload.customerId(), rideDuration );
}

RentalEnded contains only what billing needs: the rental, the customer and the duration. The mapper does not even read bike and stations. That way the rest of the service does not depend on the shape of the foreign payload, and new or changed fields from the publisher do not cause errors - fields that billing does not need are ignored. A payload that cannot be read raises an InvalidMessageException.

The handler for the event

The handler calls the aggregates one after the other:

@Service
public class RentalEndedHandler {
    public void handle( RentalEnded rental ) {
        chargingService.chargeForRide( rental.rentalId(), rental.customerId(), rental.rideDuration() );
        rideCountService.countRideOf( rental.customerId() );
    }
}

All reactions to the event and their order are visible in one place, and you can step through them in the debugger. The handler itself holds no business rule: the price rule is in Charge, the counting in RideCount. It only decides which aggregate does what, in which order. Every reaction to an incoming event starts in a handler, even when only one aggregate is involved. That way all entry points from the outside are in one package.

Each of the two services opens its own transaction: one transaction per aggregate.

What happens when a step fails or an event arrives twice is not shown in this module yet. That is the topic of the module service-development-4-failures-and-duplicates, which is still to be written.

Publishing events

Outgoing (produced) Events usually (though not always) belong to aggregates

The event ChargeCreated announces a state change of the charge, so the charge aggregate publishes it. For that, the ChargingService knows only an interface in its own application package:

public interface ChargeEvents {
    void chargeCreated( Charge charge );
}

Which implementation sits behind it is up to the configuration. The module has only one, in the test directory: InMemoryChargeEvents collects the events in a list instead of sending them. Later a Kafka producer comes in, behind the same interface.

Testing without a broker

The tests check the service on three levels, and none of them needs Kafka:

• ChargeTest checks the price rule as plain domain logic, without Spring and without a database.
• RentalEndedHandlerIT calls the handler directly with the payload in domain types. This is how you test what your service does with an event before any message can arrive.
• RentalEndedReceiverIT hands payloads to the receiver the way a Kafka adapter would, and checks the whole path up to the published event.

@Test
void aFinishedRentalIsChargedAndThePublishedEventAnnouncesIt() {
    receiver.receive( RENTAL_ENDED );

    List<Charge> charges = chargeRepository.findAll();
    assertThat( charges ).hasSize( 1 );
    assertThat( charges.get( 0 ).getAmountInCents() ).isEqualTo( 100 );
    assertThat( chargeEvents.published() ).hasSize( 1 );
}

The names show the level. ChargeTest is a unit test: it needs no Spring and runs in milliseconds. The other two start Spring and a database, so they are integration tests and end in IT, as you know it from Softwaretechnik 2. mvn test runs only the unit tests, mvn verify the integration tests as well.

The charges are in an in-memory H2 database, the published events in InMemoryChargeEvents. Further tests check that an unreadable payload is rejected and has no effect, and that new or changed fields that billing does not read do not disturb the receiver.

The integration tests use the real repositories, and that is intended: they are meant to check exactly whether service and database work together. What is simplified is something else. For simplicity, the service runs on an in-memory H2, and so do its tests. A real service would have, for example, PostgreSQL there, and an H2 in the test can miss differences, for instance in the SQL dialect. A test against the real database needs a running environment. It therefore no longer runs with mvn verify alone, but only when that environment has been started.

What these tests do not show: how Kafka actually delivers, with offsets, rebalancing and real headers. An integration test with a broker checks that later. The goal here is that you can develop and secure the whole logic of your service without a broker until then.

Later with Kafka

As soon as Kafka is available, only adapters are added, and no class of this module changes. On the way in, an adapter reads the message from the topic, recognises the event type from the header and calls the matching receiver with the payload. On the way out, a producer implements the interface ChargeEvents, and InMemoryChargeEvents stays in the tests. The rest of the service and its tests stay as they are.

The module service-development-3-with-infrastructure shows exactly this step, with the Kafka helper libraries of the Microservice Dungeon.

Running the module

Requires Java 21 and Maven. From the root directory of the repository:

mvn verify -pl service-development-2-with-interfaces-and-tests

The tests run without Kafka and without Docker; the database is an in-memory H2.