Ein Service soll mit anderen über Events zusammenarbeiten, aber Kafka steht noch nicht zur Verfügung. Diese Seite zeigt am Beispiel eines Billing-Service aus einem Fahrradverleih, wie man ihn trotzdem baut und testet: mit einem Receiver und einem Handler für jedes eingehende Event, einem Interface im Aggregate für die eigenen Events und Tests auf drei Ebenen. Später kommen für Kafka nur Adapter dazu, und keine Klasse muss geändert werden.

Sprache

English version

Beispielcode-Repository

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

Keywords

Event-getriebene Architektur, Microservices, Kafka, ohne Broker testen, Receiver, Handler, Orchestrator, Test-Double, Integrationstest, Tolerant Reader, Payload, MEA

• Einen Service ohne Broker entwickeln und testen
  • Das Beispiel: Abrechnung im Fahrradverleih
  • Ein Event hinein, ein Event hinaus
  • Was gegenüber dem Core dazukommt
  • Der Aufbau im Detail
• Konsumieren von Events (und darauf reagieren)
  • Die Receiver-Klasse
  • Wer den Receiver aufruft
  • Von der JSON-Payload zu einem Domain-Typen für die Event-Daten
  • Der Handler für den Event
• Das Publishen von Events
  • Ausgehende Events gehören meist (nicht immer) zu Aggregates
  • Ohne Broker testen
  • Später mit Kafka
  • Das Modul ausführen

Einen Service ohne Broker entwickeln und testen

Ein Service soll mit anderen über Events zusammenarbeiten, aber die gemeinsame Umgebung mit Kafka steht noch nicht. Die Frage dieses Moduls: Wie bauen Sie ihn so, dass er von Anfang an ohne Kafka läuft und sich testen lässt, und trotzdem später ohne Umbau an Kafka angeschlossen werden kann?

Ausgangspunkt ist das Modul service-development-1-core-functionality: die Geschäftslogik eines Billing-Service, aber ohne Weg nach draußen und ohne Tests. Dieses Modul fügt drei Dinge hinzu: einen Eingang für die Events eines anderen Service, einen Ausgang für die eigenen Events und Tests auf drei Ebenen. Der Code kommt dabei ganz ohne Kafka aus: Es gibt weder die Bibliothek noch einen Broker im Projekt. Das ist Absicht, denn ein Team soll mit einem sauberen Code anfangen können, bevor die gemeinsame Umgebung steht.

Das Beispiel: Abrechnung im Fahrradverleih

Das Beispiel setzt das Modul asyncapi-event-contracts fort. Der Billing-Service liest aus dem Topic rentals, in das der Rental-Service seine Events schreibt. Er braucht nur RentalEnded und reagiert darauf auf zwei Weisen:

• Er berechnet den Preis der Fahrt: 50 Cent für jede angefangene Viertelstunde, und veröffentlicht dann ein Event ChargeCreated.
• Er zählt die abgeschlossenen Fahrten je Kunde, als Grundlage für einen Treuerabatt.

RentalStarted liegt im selben Topic, wird aber nicht gebraucht: Dafür gibt es keinen Receiver.

Ein Event hinein, ein Event hinaus

Von außen gesehen ist die Funktion dieses Moduls eine Reaktion: Das konsumierte Event RentalEnded des Rental-Service löst sie aus, und am Ende steht das produzierte Event ChargeCreated. Dazwischen passiert alles im Billing-Service selbst. Er ruft den Rental-Service nicht auf, und er muss nicht wissen, wer ChargeCreated liest.

flowchart LR
    Rental["Rental-Service"]:::outside
    Topic[("Topic rentals")]:::outside
    subgraph Billing["Billing-Service"]
        Handler["RentalEndedHandler"]:::new
        Charge["charge:<br/>ChargingService<br/>Fahrt<br/>berechnen"]:::existing
        Ride["ride:<br/>RideCountService<br/>Fahrt<br/>zählen"]:::existing
        Events["charge:<br/>ChargeEvents<br/>(Interface)"]:::new
        Handler --> Charge
        Handler --> Ride
        Charge --> Events
    end
    Others["andere<br/>Services"]:::outside

    Rental --> Topic
    Topic -- "RentalEnded<br/>(konsumiert)" --> Handler
    Events -- "ChargeCreated<br/>(produziert)" --> 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["bestehende Klasse<br/>(aus Modul 1)"]:::existing ~~~ L2["neue Klasse"]:::new
    L2 ~~~ L3["außerhalb,<br/>im Test gemockt"]:::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

In diesem Modul gibt es noch keinen Broker: RentalEnded kommt in den Tests an, und ChargeCreated landet im Test-Double. Der Ablauf dazwischen ist schon derselbe wie später mit Kafka.

Was gegenüber dem Core dazukommt

Die Geschäftslogik bleibt, wie sie ist: Charge mit der Preisregel, RideCount, die Repositories und der RideCountService sind in beiden Modulen gleich. Neu sind:

• Das Interface ChargeEvents, das der ChargingService nach dem Speichern aufruft. Das ist die einzige Änderung an einer bestehenden Klasse.
• Das Package eventhandlers mit dem Handler, dem Record RentalEnded und, in infrastructure, dem Receiver für das Event RentalEnded.
• Die Tests und das Test-Double InMemoryChargeEvents.

Wenn Sie die beiden Modulordner nebeneinander vergleichen, sehen Sie genau diesen Schritt.

Der Aufbau im Detail

Der Service hat zwei Aggregates, charge und ride, jedes mit domain und application. Daneben steht das Package eventhandlers für die Reaktionen auf eingehende Events:

thkoeln.archilab.servicedevelopment
├── eventhandlers                 neu
│   ├── RentalEnded               der Payload in Domain-Typen
│   ├── RentalEndedHandler        die Reaktion auf RentalEnded
│   └── infrastructure
│       ├── RentalEndedReceiver   der Eingang für RentalEnded
│       └── RentalEventMapper     liest den Payload
├── charge
│   ├── domain                    Charge (mit der Preisregel), ChargeRepository
│   └── application               ChargingService, ChargeEvents (neu)
└── ride
    ├── domain                    RideCount, RideCountRepository
    └── application               RideCountService

Ein eingehendes Event gehört oft keinem einzelnen Aggregate: RentalEnded betrifft hier zwei. Deshalb liegt die Reaktion nicht in einem der Aggregates, sondern in einem Handler darüber. Der Weg einer Nachricht:

Topic rentals
    |    später: ein dünner Adapter für Kafka, im Modul noch nicht vorhanden
    v
RentalEndedReceiver.receive(payload als JSON)        <- hier beginnt unser Code
    |    Payload lesen und in Domain-Typen umwandeln: RentalEnded
    v
RentalEndedHandler.handle(RentalEnded)
    |    die Aggregates in der richtigen Reihenfolge aufrufen
    v
ChargingService, RideCountService
    |
    v
Interface ChargeEvents  ->  im Test: InMemoryChargeEvents      später: ein Kafka-Producer

Konsumieren von Events (und darauf reagieren)

Die Receiver-Klasse

Für jedes eingehende Event gibt es einen Receiver. Er kennt nur dieses eine Event:

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

Der Receiver bekommt nur den Payload als JSON und kennt keinen Broker-Client. Welcher Receiver für eine Nachricht zuständig ist, entscheidet, wer die Nachrichten empfängt: später ein Adapter für Kafka, der den Event-Typ aus dem Header liest, bis dahin die Tests. Ein Event, für das es keinen Receiver gibt, wird ignoriert.

Warum liegt der Receiver im Package infrastructure? Er ist zwar erstmal technologie-agnostisch. Das heißt, er ist nicht verbunden mit einer bestimmten Technologie wie z. B. Kafka oder Rabbit MQ. Aber er nimmt immerhin an, dass die Payload des Events als JSON ankommt. Das macht ihn zu Infrastruktur, also zu etwas, was Informationen über eine bestimmte Technologie für die Kommunikation mit der Außenwelt enthält.

Wer den Receiver aufruft

In Spring erwartet man an einem Listener eine Annotation wie @KafkaListener oder @RabbitListener, über die Spring die Methode aufruft, sobald eine Nachricht ankommt. Deshalb heißt RentalEndedReceiver nicht Listener: Er hat keine solche Annotation. Er ist eine gewöhnliche Methode, und das ist Absicht: Hier dockt jede Quelle von Nachrichten an, in diesem Modul die Tests, später ein Adapter für Kafka, ebenso gut einer für RabbitMQ. Die Annotation sitzt am Adapter, und dieses Modul braucht keinen Broker. Der Receiver kennt also das Format des fremden Payloads, aber keinen Broker. Der Handler darüber kennt nur noch Domain-Typen.

flowchart LR
    Test["RentalEndedReceiverIT<br/>(Test, dieses Modul)"]:::new
    Kafka["RentalsKafkaListener<br/>und RentalEndedSpecification<br/>(Kafka, Modul 3)"]:::later
    Rabbit["Methode mit<br/>@RabbitListener<br/>(RabbitMQ, denkbar)"]:::later
    Receiver["RentalEndedReceiver<br/>receive(payloadJson)"]:::new
    Handler["RentalEndedHandler<br/>handle(RentalEnded)"]:::new

    Test -- "Payload als JSON" --> Receiver
    Kafka -- "Payload als JSON" --> Receiver
    Rabbit -- "Payload als JSON" --> Receiver
    Receiver -- "Domain-Typen" --> Handler

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

Durchgezogen umrandet ist, was es in diesem Modul gibt, gestrichelt, was später dazukommt oder denkbar wäre. Alle Quellen liefern dasselbe: den Payload als JSON.

Wie viel der Adapter tun muss, hängt vom Broker ab. Bei RabbitMQ bekommt meist jeder Event-Typ eine eigene Queue, und eine Methode mit @RabbitListener( queues = "billing.rental-ended" ), die receive aufruft, genügt. Bei Kafka liegen mehrere Event-Typen in einem Topic: rentals enthält auch RentalStarted, damit alle Events einer Ausleihe in ihrer Reihenfolge bleiben. Ein @KafkaListener empfängt alles aus dem Topic, und jemand muss nach dem Event-Typ verteilen.

Von der JSON-Payload zu einem Domain-Typen für die Event-Daten

Der Receiver lässt das JSON zuerst vom RentalEventMapper lesen. Heraus kommt der Record RentalEnded. Dieser Record bildet den JSON-Payload (der ja erstmal nur ein String ist) als eigenen Domain-Typ ab. Hier reicht ein einfacher record, weil es tatsächlich nur ein Daten-Container ist. Er enthält genau die Informationen aus dem Event, die wir in unserem Kontext “Billing” brauchen. Dieser Record wird an den Handler übergeben (hier gekürzt):

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 enthält nur, was Billing braucht: die Ausleihe, den Kunden und die Dauer. Fahrrad und Stationen liest der Mapper gar nicht erst. So hängt der Rest des Service nicht an der Form des fremden Payloads, und neue oder geänderte Felder des Publishers führen nicht zu Fehlern - Felder, die Billing nicht braucht, werden ignoriert. Ein Payload, der nicht gelesen werden kann, löst eine InvalidMessageException aus.

Der Handler für den Event

Der Handler ruft die Aggregates nacheinander auf:

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

Hier sieht man an einer Stelle alle Reaktionen auf das Event und ihre Reihenfolge, und man kann sie im Debugger Schritt für Schritt durchgehen. Der Handler selbst enthält keine Geschäftsregel: Die Preisregel steht in Charge, das Zählen in RideCount. Er entscheidet nur, welches Aggregate was in welcher Reihenfolge tut. Jede Reaktion auf ein eingehendes Event beginnt in einem Handler, auch wenn nur ein Aggregate betroffen ist. So liegen alle Eingänge von außen in einem Package.

Jeder der beiden Services öffnet seine eigene Transaktion: eine Transaktion pro Aggregate.

Was passiert, wenn ein Schritt scheitert oder ein Event doppelt ankommt, zeigt dieses Modul noch nicht. Das ist Thema des Moduls service-development-4-failures-and-duplicates.

Das Publishen von Events

Ausgehende Events gehören meist (nicht immer) zu Aggregates

Das Event ChargeCreated kündigt eine Zustandsänderung der Gebühr für den Kunden (“Charge”) an, also veröffentlicht es das Aggregate charge. Der ChargingService kennt dafür nur ein Interface in seinem eigenen application-Package:

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

Welche Implementierung dahinter sitzt, entscheidet die Konfiguration. Im Modul gibt es nur eine, im Test-Verzeichnis: InMemoryChargeEvents sammelt die Events in einer Liste, statt sie zu senden. Später kommt ein Kafka-Producer dazu, hinter demselben Interface.

Ohne Broker testen

Die Tests prüfen den Service auf drei Ebenen. Keine davon braucht eine Kafka-Infrastruktur.

• ChargeTest prüft die Preisregel als reine Domain-Logik, ohne Spring und ohne Datenbank.
• RentalEndedHandlerIT ruft den Handler direkt mit dem Payload in Domain-Typen auf. So testen Sie, was Ihr Service mit einem Event tut, bevor überhaupt eine Nachricht ankommen kann.
• RentalEndedReceiverIT gibt dem Receiver Payloads in die Hand, wie es ein Kafka-Adapter täte, und prüft den ganzen Weg bis zum veröffentlichten 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 );
}

Die Namen zeigen die Ebene. ChargeTest ist ein Unit-Test: Er braucht kein Spring und läuft in Millisekunden. Die beiden anderen starten Spring und eine Datenbank, sie sind also Integrationstests und enden deshalb auf IT, so wie Sie es aus Softwaretechnik 2 kennen. mvn test führt nur die Unit-Tests aus, mvn verify zusätzlich die Integrationstests.

Die Charges liegen in einer H2-Datenbank im Speicher, die veröffentlichten Events im InMemoryChargeEvents. Weitere Tests prüfen, dass ein unlesbarer Payload abgelehnt wird und nichts bewirkt, und dass neue oder geänderte Felder, die Billing nicht liest, den Receiver nicht stören.

Die Integrationstests benutzen die echten Repositories, und das ist auch so gewollt. Sie sollen ja gerade prüfen, ob Service und Datenbank zusammenspielen. Allerdings gibt es eine Vereinfachung: Der Service läuft auf einer H2 im Speicher, und die Tests deshalb auch. In einem echten Service würde man zum Beispiel eher PostgreSQL nutzen. Eine H2 im Test kann Unterschiede übersehen, etwa im SQL-Dialekt. Ein Test gegen die echte Datenbank bräuchte dann aber eine laufende Infrastruktur-Umgebung. Er läuft deshalb nicht mehr mit mvn verify allein, sondern nur, wenn diese Umgebung gestartet ist.

Was diese Tests nicht zeigen, ist was Kafka tatsächlich liefert, mit Offsets, Rebalancing und echten Headers. Das prüft später ein Integrationstest mit einem Broker. Das Ziel hier ist, dass Sie bis dahin die ganze Logik Ihres Service ohne Broker entwickeln und absichern können.

Später mit Kafka

Sobald Kafka zur Verfügung steht, kommen nur Adapter dazu, und keine Klasse dieses Moduls ändert sich. Auf der Eingangsseite liest ein Adapter die Nachricht aus dem Topic, erkennt am Header den Event-Typ und ruft den passenden Receiver mit dem Payload auf. Auf der Ausgangsseite implementiert ein Producer das Interface ChargeEvents, und InMemoryChargeEvents bleibt im Test. Der Rest des Service und seine Tests bleiben, wie sie sind.

Das Modul service-development-3-with-infrastructure zeigt genau diesen Schritt, mit den Kafka-Hilfsbibliotheken des Microservice Dungeon.

Das Modul ausführen

Voraussetzung: Java 21 und Maven. Aus dem Hauptverzeichnis des Repositories:

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

Die Tests laufen ohne Kafka und ohne Docker; die Datenbank ist eine H2 im Speicher.