Kafka

Kafka carries what crosses a service's boundary: the streams a context publishes, the commands other contexts send it, and the receipts that come back. Inside a service, the log is the bus and Kafka is not involved. This page explains which topics a service uses, how to create them, and how to configure the transport.

When a service needs Kafka

A service needs Kafka when one of these holds:

• it publishes its aggregates' streams for other contexts or external systems;
• it subscribes to another context's published stream;
• its sagas send commands to another context, or receive their receipts;
• other contexts send commands to it;
• an aggregate type's commands are routed through Kafka for durable intake.

A node whose sagas send commands to another context does not start without a command transport. For local development, wow.topics.in-memory: true keeps the topics in the JVM instead, so services in one process reach each other without a broker.

The transport is wired when the wow-kafka module is on the classpath and wow.kafka.bootstrap-servers is set. The eShop's ordering service connects with its kafka profile:

wow:
  kafka:
    # Creates the topics the service owns when they are missing, for trying it out; production validates them, the
    # default (decision D168).
    topics:
      ddl: auto
    bootstrap-servers: localhost:9092
    # Execute the commands other contexts send to wow.eshop-ordering.commands (the payment's), which has as many
    # partitions as the context's log (wow.log.partitions, 64 by default).
    command-intake: true

Topics

Each context owns its topics; the topics of other contexts are theirs.

Topic	Written by	Read by
wow.<context>.<aggregate>.state	This context's Kafka sink, one record per commit of each published aggregate	Other contexts' subscriptions, external consumers
wow.<context>.commands	Other contexts' sagas and timers; this context's own nodes for aggregate types routed through Kafka	This context
wow.<context>.receipts	Other contexts, when they reject a command this context sent, or commit one it asked a receipt for	This context's sagas

An aggregate is published by default; it can stay unpublished, publish without its state, or name another topic. wow.kafka.topic-prefix puts a prefix before every topic name, such as staging., on both the writing and the reading side. Every record is an envelope keyed by the aggregate, so one aggregate's records stay in order (Wire protocols).

Partitions

• A command topic has exactly as many partitions as its context's log (wow.log.partitions). A command's partition in the topic is its aggregate's partition in the log, so the node that reads it is the node that owns the aggregate. With another count, commands would reach nodes that do not own their aggregates, so a node that reads a command topic checks the count at startup and refuses to start on a mismatch. Never add partitions to a command topic.
• A state or receipts topic can have any number of partitions, and partitions can be added while services run. A reading node looks for new partitions every wow.kafka.topics.partition-refresh (30 seconds) and reads them from their beginning, without a restart.

Creating topics

At startup, a node checks the topics its contexts own, as wow.kafka.topics.ddl says:

ddl	At startup
validate (default)	The node does not start while one of its topics is missing, or its command topic does not have exactly the log's partitions. The message names each such topic.
auto	The node first creates the missing topics, with the log's partition count, wow.kafka.topics.replication-factor, wow.kafka.topics.min-insync-replicas and wow.kafka.topics.configs.*, then validates. It never alters an existing topic.
none	The node checks nothing, except that a command topic it reads still has the log's partition count.

Use auto for development and tests. In production, create the topics yourself and keep validate, which catches a missing one at the first start. For the eShop's ordering context:

kafka-topics --bootstrap-server kafka:9092 --create --topic wow.eshop-ordering.commands --partitions 64
kafka-topics --bootstrap-server kafka:9092 --create --topic wow.eshop-ordering.receipts --partitions 64
kafka-topics --bootstrap-server kafka:9092 --create --topic wow.eshop-ordering.order.state --partitions 64

No node ever creates a topic as a side effect of reading it, so a stream that does not exist yet is waited for, not created with the broker's default partition count.

Durability

A command accepted on a command topic must survive the loss of a broker. That holds only when the topic's replication factor and min.insync.replicas are both at least 2. At startup, a node warns about each command topic of its contexts that is below that, and about a configuration that would create one, unless a development profile is active (dev, local, test or demo).

The producer is idempotent and waits for every in-sync replica (acks=all), whatever wow.kafka.properties says.

Durable command intake

By default a command is routed over the cluster channel to the owner of its aggregate, and ACCEPTED only means it passed admission: a node that stops before running it loses it. For commands that must not be lost once accepted, such as asynchronous intake or absorbing load peaks, route their aggregate type through Kafka with wow.aggregates.[<context>.<aggregate>].routing: kafka:

• The node that admits such a command writes it to the context's command topic and waits for the broker. A request that waits only for accepted is answered 202 once the broker has acknowledged it; from then on, the command is durable.
• The owner of the aggregate's partition reads it from the topic and executes it, so one aggregate's commands keep their order. A longer wait, such as snapshot, goes on as usual.
• If the broker does not acknowledge within the wait's timeout, the wait ends at ACCEPTED, failed. A retry with the same request id is not decided twice.

Only commands admitted at the edge take this route. A command a saga or a timer sends is already durable, because its sender retries until it commits.

A context whose aggregate types are routed through Kafka reads its command topic whatever wow.kafka.command-intake says. Otherwise, set wow.kafka.command-intake: true on a service that should execute the commands other contexts send it.

Reading other contexts' streams

A service reads other contexts' topics without a consumer group. Its leases assign the topic's partitions to its nodes, the same way they assign the log's, and the offsets are kept in the service's own checkpoint store, beside its other checkpoints. One consumer per node and topic feeds every subscription that reads the topic. So there is no consumer group to manage, reset or monitor in Kafka; a subscription's position is its checkpoint.

When a partition no longer holds the offset a subscription is to read from, because retention deleted it or the topic was created again, that is an offset gap. wow.kafka.offset-gap decides what happens:

• fail (default): the read fails and the subscription stays where it is, logging errors;
• earliest: the subscription reads on from the earliest retained offset, skipping what was lost.

Either way the gap is logged as an error and counted in the metric wow.kafka.offset.gaps; alert on it. Set each topic's retention longer than any subscription may fall behind.

To recover a subscription stopped by a gap, set wow.kafka.offset-gap: earliest and restart its nodes: the read resumes from the earliest retained offset. The skipped range is lost to that subscription, and stays counted in wow.kafka.offset.gaps. Set the option back to fail afterwards to keep the strict default. When the loss is not acceptable and what was published is still complete elsewhere, rebuild under a new subscription name instead.

Client settings

wow.kafka.properties.* passes settings to every producer and consumer as they are, such as security.protocol and SASL settings, client.id or linger.ms. A few settings belong to Wow and cannot be overridden, because the guarantees rest on them: the idempotent producer with acks=all, consumers without a group that commit nothing and reset no offset on their own, and no topic creation on read.