Getting started

Requirements

Java 21 or later.
Kafka brokers new enough to serve topic IDs (3.7+).
processing.guarantee=exactly_once_v2 — enforced by Parsley.streams, which also pins the consumer to read_committed. Parsley's state is defined against EOS; there is no at-least-once mode.

Adding the dependency

<repositories>
    <repository>
        <id>central-snapshots</id>
        <url>https://central.sonatype.com/repository/maven-snapshots/</url>
        <releases><enabled>false</enabled></releases>
        <snapshots><enabled>true</enabled></snapshots>
    </repository>
</repositories>

<dependencies>
    <dependency>
        <groupId>io.github.tobyjamesclements</groupId>
        <artifactId>parsley</artifactId>
        <version>0.2.0-SNAPSHOT</version>
    </dependency>

    <!-- Only for a test suite that uses ContractProbes. -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams-test-utils</artifactId>
        <version>4.3.1</version>
        <scope>test</scope>
    </dependency>
</dependencies>

The same in Gradle:

repositories {
    mavenCentral()
    maven { url = uri("https://central.sonatype.com/repository/maven-snapshots/") }
}

dependencies {
    implementation("io.github.tobyjamesclements:parsley:0.2.0-SNAPSHOT")
    testImplementation("org.apache.kafka:kafka-streams-test-utils:4.3.1")   // ContractProbes
}

kafka-clients and kafka-streams are inherited and required; the coordinates alone run everything on this page. An application with no stage, using only CausalClock and CausalHeaders (plain clients), is the one exception and may drop 74 MiB of RocksDB with <exclusion> / exclude(group = "org.apache.kafka", module = "kafka-streams").

The shape: functional core, imperative edges

Your logic is a pure function. It receives a causally delivered Message and returns Emission values; with per-key state, it is a fold that also returns the next state. It holds no reference to any runtime, imports nothing from Kafka, and is unit-testable with plain equality. Everything imperative — gating, decoding, state persistence, stamping, partitioning, the transaction — lives in the stage runtime that hosts it.

Topics and codecs

Topics are typed values declared once — codecs live with the declaration, and a pipeline hop is the same Topic appearing as one stage's sink and another's source, which makes codec agreement across the hop hold by construction. A Codec is two pure functions between a value and its bytes; Kafka serdes (Avro, JSON Schema, and friends) bridge in with Codec.fromSerde(serde, topicName). The codec contract, hand-rolled codecs, and the schema-registry formats are covered in codecs and Avro.

Topic<String, Order>   orders      = Topic.of("orders", Codec.utf8(), orderCodec);
Topic<String, Payment> payments    = Topic.of("payments", Codec.utf8(), paymentCodec);
Topic<String, Settled> settlements = Topic.of("settlements", Codec.utf8(), settledCodec);

A stateless stage

A Handler maps one delivered message to its emissions. Emissions are created by the sink topic itself — topic.send(key, value) — so the payload type-checks at the construction site, and an emission to an undeclared sink fails loudly at the stamping site.

Stage settlement = Stage.named("settlement")
        .on(orders,   m -> List.of(settlements.send(m.key(), settle(m.value()))))
        .on(payments, m -> List.of(settlements.send(m.key(), apply(m.value()))))
        .into(settlements)
        .build();

Properties props = new Properties();
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");

try (CausalStreams app = Parsley.named("settlements-app", settlement).streams(props)) {
    app.start();
    // run until shutdown
}

The application id goes to Parsley.named rather than into the properties, because it names things: streams sets application.id from it, and it prefixes every topic and store the application owns — the same convention Kafka Streams uses, so a cluster's topic listing keeps an application's topics together. See expectations.

Every cause the stage consumes has already been delivered when a handler runs; that is the guarantee. A message carries its own coordinate — source topic, partition, offset, timestamp — and for a message released from the hold queue, the coordinate is its own, not the release trigger's. By default an emission inherits the handled message's timestamp; send(key, value, timestamp) overrides it.

A stateful stage

Declaring state gives the stage per-key state, and sources fold over it: the runtime resolves the state for the message's key (the declared initial value for an unseen key), applies the pure Fold, persists the returned state, and applies the returned emissions — all in the same transaction as the delivery. Causal order plus a pure fold makes the state a deterministic function of the delivered history.

Stage balances = Stage.named("balances")
        .state(balanceCodec, Balance::zero)
        .on(orders,   (bal, m) -> Step.of(bal.minus(m.value().total()),
                                          settlements.send(m.key(), settled(bal, m))))
        .on(payments, (bal, m) -> Step.of(bal.plus(m.value().amount())))
        .into(settlements)
        .build();

State is keyed by the message key's encoded bytes — the same agreement co-partitioning already requires — and scoped to the stage. Returning Step.of(null, ...) deletes the key's state. The stage name keys its stores, so keep it stable across deployments.

Testing your logic

Handlers and folds are pure functions over values with equality, so the primary tests need no runtime at all:

assertEquals(List.of(settlements.send("k", expected)),
        handler.handle(Message.of("orders", "k", order)));

assertEquals(Step.of(expectedBalance, settlements.send("k", expected)),
        fold.apply(balance, Message.of("payments", "k", payment)));

For the wiring, Parsley.named(applicationId, stage).testTopology() returns a fresh broker-less topology for TopologyTestDriver: pipe records and assert on outputs. Stamp presence is observable via the CausalHeaders name constants; gating behaviour itself is Parsley's contract, verified by Parsley's own suite. The test topology is for the driver only — under a real cluster it neither captures consumer positions nor resolves real topic identity, and the adapter fails closed at the first delivered record.

What the runtime wires for you

Parsley.streams installs a client supplier that records consumer positions after each poll (the liveness signal — see liveness), and resolves topic identity and sink end offsets through an admin client built from the same properties. A declared sink must exist before the application starts: init fails loudly rather than run with own-output claims silently off. The returned CausalStreams is a curated allowlist over the Kafka Streams runtime — lifecycle, state, listeners, metrics, lag — with no accessor to the underlying instance. The withheld members are absent for stated reasons, on the class Javadoc member by member: interactive queries can observe uncommitted transactional state, pausing an instance freezes the release of held records fleet-wide, and the rest duplicates what scaling by instances already expresses. metrics() includes the stage's own gauges and counters in the vc-metrics group, listed in the metrics reference.

Where next

Topology shapes — the common wiring shapes (linear, fan-out, fan-in, diamond, request and reply, event flows, cycles) and exactly what ordering each one gets.
Codecs and Avro — the codec contract, writing your own, bridging serdes, and schema-registry formats.
Ticks — time-driven policy as delivered records: the runtime emits a stamped tick per interval, and pure logic folds over it.
The contract — everything Parsley expects of you and everything it promises back, including the operational notes.
Verifying your application — the contract's checkable clauses as probes in your test suite and deploy pipeline.
Diagnosing held records — explainHolds(), the hold warnings, and what each diagnosis means when records are waiting at the gate.
Plain clients — stamping from plain producers, observing from plain consumers.