The service that so far ran and was tested without a broker is connected to Kafka. No class changes for that, only adapters are added: one Kafka listener per topic that dispatches by the event type in the header with the consumer-lib, and a producer behind the aggregate’s interface that sends through the outbox-lib. The page shows the path of a message, the headers, and why the tests still run without a broker.

Language

Deutsche Fassung

Example code repository

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

Keywords

Event-driven architecture, Microservices, Kafka, KafkaListener, Outbox, outbox-lib, consumer-lib, Headers, Adapter, Consumer group, Microservice Dungeon

• Connecting a service to Kafka
  • One event in, one event out, now with Kafka
  • What is added
  • The path of a message
  • The way in
  • The way out
  • The headers
  • Testing without a broker
  • Running the module

Connecting a service to Kafka

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

The module service-development-2-with-interfaces-and-tests built a service that runs without Kafka and is tested: with an interface for its own events and a receiver for each incoming event. This module connects it to Kafka. The question: what has to change for that?

The answer: nothing. Only adapters are added, and no class from module 2 is changed. The adapters use the two Kafka helper libraries of the Microservice Dungeon, the outbox-lib for publishing and the consumer-lib for receiving.

One event in, one event out, now with Kafka

Seen from the outside, the service does the same as in module 2: the consumed event RentalEnded triggers the reaction, and it ends with the produced event ChargeCreated. What is new is that both now really travel through Kafka. For that, two adapters are added, one on each side, and everything in between stays as it was.

flowchart LR
    Topic[("Topic rentals<br/>(from the<br/>rental service)")]:::outside
    subgraph Billing["Billing service"]
        In["eventhandlers:<br/>RentalsKafkaListener<br/>RentalEndedSpecification<br/>(with consumer-lib)"]:::new
        Handler["RentalEndedHandler"]:::existing
        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)"]:::existing
        Kafka["charge:<br/>KafkaChargeEvents<br/>(with outbox-lib)"]:::new
        In --> Handler
        Handler --> Charge
        Handler --> Ride
        Charge --> Events
        Events --> Kafka
    end
    TopicOut[("Topic charges<br/>(for other services)")]:::outside

    Topic -- "RentalEnded<br/>(consumed)" --> In
    Kafka -- "ChargeCreated<br/>(produced)" --> 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["existing class<br/>(from module 2)"]:::existing ~~~ L2["new class"]:::new
    L2 ~~~ L3["outside<br/>the 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

What is added

Comparing the two module folders, you find only new files:

thkoeln.archilab.servicedevelopment
├── BillingServiceApplication      the service, now startable
├── eventhandlers
│   └── infrastructure
│       ├── RentalEndedReceiver    unchanged from module 2
│       ├── RentalEventMapper      unchanged from module 2
│       └── kafka                  new
│           ├── RentalsKafkaListener      reads the topic rentals
│           └── RentalEndedSpecification  connects the event type to the receiver
└── charge
    ├── domain                     unchanged
    ├── application                unchanged, with the interface ChargeEvents
    └── infrastructure             new
        └── KafkaChargeEvents      implements ChargeEvents

The Kafka side always lives in a package infrastructure. On the way in, it has a package kafka of its own there, next to the receiver from module 2: the receiver is where any source of messages plugs in, and an adapter for RabbitMQ would go next to it as a package rabbitmq. On the way out, it lives under the aggregate that publishes the event. domain and application still know nothing about Kafka.

The path of a message

Topic rentals
    |
    v
RentalsKafkaListener.onRecord(ConsumerRecord)       <- new: @KafkaListener
    |    the consumer-lib reads the headers and finds the specification for the event type
    v
RentalEndedSpecification.process(context)           <- new: only translates types
    |
    v
RentalEndedReceiver.receive(payload as JSON)        <- from here on everything as in module 2
    v
RentalEndedHandler, ChargingService, RideCountService
    |
    v
Interface ChargeEvents  ->  KafkaChargeEvents  ->  outbox-lib  ->  Topic charges

The way in

There is one Kafka listener per topic. It passes every record on to the consumer-lib:

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

The library reads the event type from the headers and calls the EventSpecification registered for that type. For RentalEnded this is a small class that passes the payload on to the receiver from module 2:

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

A record whose type has no registered specification is skipped. That is how RentalStarted is ignored, without a line written for it.

Why not a @KafkaListener of its own for each event? Two listeners on the same topic in the same consumer group share its partitions, and each would get only part of the events. With a consumer group of its own for each, every listener would read the whole topic instead and have to throw away all other events. One listener per topic that dispatches by header avoids both.

The way out

KafkaChargeEvents implements the interface ChargeEvents from module 2 and writes the event through the 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 tells the library where the event goes and what it is called: topic charges, event type ChargeCreated in version 1, the customer ID as key. The key decides the partition, and within a partition the order is kept. So all charges of one customer arrive in the order in which they were created.

The ChargingService calls chargeCreated inside its transaction. The library sends the event only after the commit, and not at all on a rollback.

Fine print: the direct mode and the outbox pattern

The module uses outbox-lib-direct. This variant sends straight to Kafka and needs no database, but gives no delivery guarantee: if the service crashes after the commit and before sending, the event is lost. It is meant for development. The core services of the Microservice Dungeon use outbox-lib-jpa: there, the event is written to an outbox table in the same transaction as the charge, and a separate service publishes it from there, at least once and in the right order. Both variants have the same interfaces; switching is only a different dependency. The details are in the README of the outbox-lib.

The headers

Both libraries use the same headers, as the outbox-lib sets them when sending:

Header	Content
event_id	unique ID of the event, as text
sequence_id	consecutive number, 8 bytes
created_at	point in time in milliseconds since 1970, 8 bytes
event_type	name of the event type, e.g. RentalEnded
event_type_version	version of the event type, 4 bytes
entity	which object changed, e.g. charge.<id>

The consumer-lib reads the event type and its version to find the matching specification. The contracts in the module asyncapi-event-contracts list exactly these headers.

Testing without a broker

The tests from module 2 run here unchanged. The Kafka adapters are only active with the Spring profile kafka, and the tests run without that profile: instead of KafkaChargeEvents there is then only InMemoryChargeEvents from the test directory. In addition there are tests for the new adapters, which need no broker either:

• KafkaChargeEventsTest checks what the producer hands to the library: topic, event type, key and payload. The outbox is a small stand-in that only remembers the event.
• RentalsKafkaListenerIT hands the Kafka listener records the way Kafka would deliver them: with key, payload as bytes and the headers above. A RentalEnded takes the same path as in module 2, RentalStarted and unknown types are skipped.

@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 );
}

Whether Kafka really delivers the records like this is not shown by these tests. Only the service running against a real broker shows that.

Running the module

Requires Java 21 and Maven. The tests run without Kafka and without Docker. From the root directory of the repository:

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

To run the service against Kafka, you need a broker, for example from the local environment of the Microservice Dungeon. There, Kafka is reachable from outside Docker at localhost:29092; give a different address with the environment variable KAFKA_BOOTSTRAP_SERVERS. The topics rentals and charges have to be created beforehand, because the outbox-lib asks for the partitions of its topics on startup. Then:

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

On startup, the outbox-lib logs a warning that it runs in direct mode. That is intended.