Der Service, der bisher ohne Broker lief und getestet war, kommt an Kafka. Dafür ändert sich keine Klasse, es kommen nur Adapter dazu: ein Kafka-Listener pro Topic, der mit der consumer-lib nach dem Event-Typ im Header verteilt, und ein Producer hinter dem Interface des Aggregates, der über die outbox-lib sendet. Die Seite zeigt den Weg einer Nachricht, die Headers und warum die Tests weiter ohne Broker laufen.

Sprache

English version

Beispielcode-Repository

https://gitlab.com/archi-lab/public/learning/learn-event-driven-architecture/-/tree/main/service-development-3-with-infrastructure

Keywords

Event-getriebene Architektur, Microservices, Kafka, KafkaListener, Outbox, outbox-lib, consumer-lib, Headers, Adapter, Consumer Group, Microservice Dungeon, MEA

• Einen Service an Kafka anschließen
  • Ein Event hinein, ein Event hinaus, jetzt mit Kafka
  • Was dazukommt
  • Der Weg einer Nachricht
  • Der Eingang
  • Der Ausgang
  • Die Headers
  • Ohne Broker testen
  • Das Modul ausführen

Einen Service an Kafka anschließen

Das Modul service-development-2-with-interfaces-and-tests hat einen Service gebaut, der ohne Kafka läuft und getestet ist: mit einem Interface für die eigenen Events und einem Receiver für jedes eingehende Event. Dieses Modul schließt ihn an Kafka an. Die Frage: Was muss sich dafür ändern?

Die Antwort: nichts. Es kommen nur Adapter dazu, und keine Klasse aus Modul 2 wird geändert. Die Adapter benutzen die beiden Kafka-Hilfsbibliotheken des Microservice Dungeon, die outbox-lib zum Veröffentlichen und die consumer-lib zum Empfangen.

Ein Event hinein, ein Event hinaus, jetzt mit Kafka

Von außen gesehen tut der Service dasselbe wie in Modul 2: Das konsumierte Event RentalEnded löst die Reaktion aus, und am Ende steht das produzierte Event ChargeCreated. Neu ist, dass beide jetzt wirklich über Kafka laufen. Dafür kommen zwei Adapter dazu, auf jeder Seite einer, und alles dazwischen bleibt, wie es war.

flowchart LR
    Topic[("Topic rentals<br/>(vom<br/>Rental-Service)")]:::outside
    subgraph Billing["Billing-Service"]
        In["eventhandlers:<br/>RentalsKafkaListener<br/>RentalEndedSpecification<br/>(mit consumer-lib)"]:::new
        Handler["RentalEndedHandler"]:::existing
        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)"]:::existing
        Kafka["charge:<br/>KafkaChargeEvents<br/>(mit outbox-lib)"]:::new
        In --> Handler
        Handler --> Charge
        Handler --> Ride
        Charge --> Events
        Events --> Kafka
    end
    TopicOut[("Topic charges<br/>(für andere Services)")]:::outside

    Topic -- "RentalEnded<br/>(konsumiert)" --> In
    Kafka -- "ChargeCreated<br/>(produziert)" --> TopicOut

    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 2)"]:::existing ~~~ L2["neue Klasse"]:::new
    L2 ~~~ L3["außerhalb<br/>des Service"]:::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

Was dazukommt

Wenn Sie die beiden Modulordner vergleichen, finden Sie nur neue Dateien:

thkoeln.archilab.servicedevelopment
├── BillingServiceApplication      der Service, jetzt startbar
├── eventhandlers
│   └── infrastructure
│       ├── RentalEndedReceiver    unverändert aus Modul 2
│       ├── RentalEventMapper      unverändert aus Modul 2
│       └── kafka                  neu
│           ├── RentalsKafkaListener      liest das Topic rentals
│           └── RentalEndedSpecification  verbindet den Event-Typ mit dem Receiver
└── charge
    ├── domain                     unverändert
    ├── application                unverändert, mit dem Interface ChargeEvents
    └── infrastructure             neu
        └── KafkaChargeEvents      implementiert ChargeEvents

Die Kafka-Seite liegt immer in einem Package infrastructure. Beim Eingang hat sie dort ein eigenes Package kafka, neben dem Receiver aus Modul 2: Der Receiver ist die Stelle, an der jede Quelle von Nachrichten andockt, und ein Adapter für RabbitMQ käme als Package rabbitmq daneben. Beim Ausgang liegt sie unter dem Aggregate, das das Event veröffentlicht. domain und application wissen weiter nichts von Kafka.

Der Weg einer Nachricht

Topic rentals
    |
    v
RentalsKafkaListener.onRecord(ConsumerRecord)       <- neu: @KafkaListener
    |    die consumer-lib liest die Header und sucht die Specification zum Event-Typ
    v
RentalEndedSpecification.process(context)           <- neu: übersetzt nur Typen
    |
    v
RentalEndedReceiver.receive(payload als JSON)       <- ab hier alles wie in Modul 2
    v
RentalEndedHandler, ChargingService, RideCountService
    |
    v
Interface ChargeEvents  ->  KafkaChargeEvents  ->  outbox-lib  ->  Topic charges

Der Eingang

Für jedes Topic gibt es einen Kafka-Listener. Er gibt jeden Record an die consumer-lib weiter:

@KafkaListener( topics = "rentals" )
public void onRecord( ConsumerRecord<String, byte[]> record ) {
    chain.process( record );
}

Die Bibliothek liest aus den Headern den Event-Typ und ruft die EventSpecification auf, die für diesen Typ registriert ist. Für RentalEnded ist das eine kleine Klasse, die den Payload an den Receiver aus Modul 2 weitergibt:

@Override
public void process( KafkaEventContext<KafkaDefaultHeader, byte[]> context ) {
    receiver.receive( new String( context.value(), UTF_8 ) );
}

Ein Record, für dessen Typ keine Specification registriert ist, wird übersprungen. So wird RentalStarted ignoriert, ohne dass eine Zeile dafür geschrieben werden muss.

Warum nicht für jedes Event einen eigenen @KafkaListener? Zwei Listener auf demselben Topic in derselben Consumer Group teilen sich dessen Partitionen, und jeder bekäme nur einen Teil der Events. Mit einer eigenen Consumer Group für jeden läse dagegen jeder das ganze Topic und müsste alle fremden Events wegwerfen. Ein Listener pro Topic, der nach dem Header verteilt, vermeidet beides.

Der Ausgang

KafkaChargeEvents implementiert das Interface ChargeEvents aus Modul 2 und schreibt das Event über die outbox-lib:

@Override
public void chargeCreated( Charge charge ) {
    ChargeCreatedPayload payload = new ChargeCreatedPayload(
            charge.getId(), charge.getRentalId(), charge.getCustomerId(), charge.getAmountInCents() );
    outbox.write( new ChargeCreatedMessage( charge, json.writeValueAsBytes( payload ) ) );
}

ChargeCreatedMessage sagt der Bibliothek, wohin das Event geht und wie es heißt: Topic charges, Event-Typ ChargeCreated in Version 1, die Customer-ID als Key. Der Key bestimmt die Partition, und innerhalb einer Partition bleibt die Reihenfolge erhalten. Alle Charges eines Kunden kommen also in der Reihenfolge an, in der sie entstanden sind.

Der ChargingService ruft chargeCreated innerhalb seiner Transaktion auf. Die Bibliothek sendet das Event erst nach dem Commit, und bei einem Rollback gar nicht.

Fine print: der Direct Mode und das Outbox Pattern

Das Modul benutzt outbox-lib-direct. Diese Variante sendet direkt an Kafka und braucht keine Datenbank, gibt aber keine Zustellgarantie: Stürzt der Service nach dem Commit und vor dem Senden ab, ist das Event verloren. Sie ist für die Entwicklung gedacht. Die Core-Services des Microservice Dungeon benutzen outbox-lib-jpa: Dort wird das Event in derselben Transaktion wie die Charge in eine Outbox-Tabelle geschrieben, und ein eigener Service veröffentlicht es von dort, mindestens einmal und in der richtigen Reihenfolge. Beide Varianten haben dieselben Interfaces, der Wechsel ist nur eine andere Dependency. Die Details stehen im README der outbox-lib.

Die Headers

Beide Bibliotheken verwenden dieselben Headers, so wie die outbox-lib sie beim Senden setzt:

Header	Inhalt
event_id	eindeutige ID des Events, als Text
sequence_id	fortlaufende Nummer, 8 Byte
created_at	Zeitpunkt in Millisekunden seit 1970, 8 Byte
event_type	Name des Event-Typs, z. B. RentalEnded
event_type_version	Version des Event-Typs, 4 Byte
entity	welches Objekt sich geändert hat, z. B. charge.<id>

Den Event-Typ und seine Version liest die consumer-lib, um die passende Specification zu finden. Die Contracts im Modul asyncapi-event-contracts führen genau diese Headers auf.

Ohne Broker testen

Die Tests aus Modul 2 laufen hier unverändert mit. Die Kafka-Adapter sind nur mit dem Spring-Profil kafka aktiv, und die Tests laufen ohne dieses Profil: Statt KafkaChargeEvents gibt es dann nur InMemoryChargeEvents aus dem Test-Verzeichnis. Dazu kommen Tests für die neuen Adapter, die ebenfalls keinen Broker brauchen:

• KafkaChargeEventsTest prüft, was der Producer der Bibliothek übergibt: Topic, Event-Typ, Key und Payload. Die Outbox ist dabei eine kleine Attrappe, die sich das Event nur merkt.
• RentalsKafkaListenerIT gibt dem Kafka-Listener Records in die Hand, wie Kafka sie liefern würde: mit Key, Payload in Bytes und den Headers oben. Ein RentalEnded nimmt denselben Weg wie in Modul 2, RentalStarted und unbekannte Typen werden übersprungen.

@Test
void aRentalEndedRecordTakesTheSamePathAsInModuleTwo() {
    kafkaListener.onRecord( record( "RentalEnded", RentalEndedReceiverIT.RENTAL_ENDED ) );

    assertThat( chargeRepository.findAll() ).hasSize( 1 );
    assertThat( rideCountService.ridesOf( CUSTOMER_ID ) ).isEqualTo( 1 );
    assertThat( chargeEvents.published() ).hasSize( 1 );
}

Ob Kafka die Records wirklich so zustellt, zeigen diese Tests nicht. Das zeigt erst der Service, wenn er gegen einen echten Broker läuft.

Das Modul ausführen

Voraussetzung: Java 21 und Maven. Die Tests laufen ohne Kafka und ohne Docker. Aus dem Hauptverzeichnis des Repositories:

mvn verify -pl service-development-3-with-infrastructure

Um den Service gegen Kafka laufen zu lassen, brauchen Sie einen Broker, zum Beispiel aus der lokalen Umgebung des Microservice Dungeon. Dort ist Kafka von außerhalb von Docker unter localhost:29092 erreichbar; eine andere Adresse geben Sie mit der Umgebungsvariable KAFKA_BOOTSTRAP_SERVERS an. Die Topics rentals und charges müssen vorher angelegt sein, denn die outbox-lib fragt beim Start nach den Partitionen ihrer Topics. Dann:

mvn spring-boot:run -pl service-development-3-with-infrastructure -Dspring-boot.run.profiles=kafka

Beim Start meldet die outbox-lib in einer Warnung, dass sie im Direct Mode läuft. Das ist gewollt.