Guides

Seeding topics from fixtures

@KafkaSeed is the @Sql of this library: declare the records a test starts from in a file, and the harness publishes them — before the class, before the method, and again after every cleanup, so each test sees the state it declared.

Three entry points

All three read the same fixture files and behave identically otherwise:

Suite — wiretap.test.seed.locations, the state every test class in the context starts from.
Class — @KafkaSeed on the test class: that class's declared starting state.
Method — @KafkaSeed on a test method: extra records for that one test.

OrderProjectionIT.java

@SpringBootTest
@EnableWiretapKafkaBroker
@KafkaSeed("classpath:seeds/orders.json")                       // topic "orders"
class OrderProjectionIT {

    @Test
    void projectsAnOverdueOrder() {
        assertThatTopic("orders").hasRecordCount(3);
    }

    @Test
    @KafkaSeed(topic = "orders", locations = "classpath:seeds/edge-case-orders.json")
    @KafkaSeed("classpath:seeds/payments.ndjson")
    void projectsAnEdgeCase() {
        // "orders" carries the class fixture plus the edge cases, in that order.
        // "payments" exists only for this method.
    }
}

@SpringBootTest
@EnableWiretapKafkaBroker
@KafkaSeed("classpath:seeds/orders.json")                       // topic "orders"
class OrderProjectionIT {

    @Test
    void projectsAnOverdueOrder() {
        assertThatTopic("orders").hasRecordCount(3);
    }

    @Test
    @KafkaSeed(topic = "orders", locations = "classpath:seeds/edge-case-orders.json")
    @KafkaSeed("classpath:seeds/payments.ndjson")
    void projectsAnEdgeCase() {
        // "orders" carries the class fixture plus the edge cases, in that order.
        // "payments" exists only for this method.
    }
}

@KafkaSeed

Package: io.typecheck.wiretap.core.seed. Repeatable, and valid on a test class or a test method.

Property	Default	Description
`value()` String[]	`{}`	Alias for `locations()`, for the common `@KafkaSeed("classpath:seeds/orders.json")` form.
`locations()` String[]	`{}`	Alias for `value()`, for when other attributes are set too. Declaring both is an error unless they are equal — the harness reports it rather than silently picking one. Each location names one fixture file: `classpath:`, `file:`, or a plain filesystem path. There is no directory scanning.
`topic()` String	`""`	Target topic. Empty means the fixture's file name without its extension, so `classpath:seeds/orders.json` seeds `orders` and one annotation over several locations seeds several topics.
`avroSchema()` String	`""`	An `.avsc` location — `classpath:`, `file:` or a plain path — whose record schema this seed's values are converted to and published Avro-encoded against the mock schema registry. Empty is the default JSON behaviour. See Avro fixtures.
`placeholders()` Class<? extends SeedPlaceholders>[]	`{}`	Providers for `${...}` bindings, consulted in declaration order. See bindings from providers.

A topic that does not exist yet is created first, through the same two-phase ledger every other wiretap topic goes through. A seeded topic then joins the per-method cleanup scope for the cluster it was seeded on — which is what makes seeding a pass-through cluster's pre-existing topic both safe to do and safe to clean.

Ordering and re-application

Within one test method the order is always suite seeds, then class-level seeds, then method-level seeds — so a class fixture lands on top of the suite fixture, and a method fixture on top of both. Within one level, annotations apply in declaration order.

What survives a per-method cleanup is deliberately asymmetric:

Class seeds are always re-applied. A class-level @KafkaSeed is that class's declared starting state, in either seed mode.
Method seeds never are. They belong to the method that declared them and go away with its cleanup.
Suite seeds depend on wiretap.test.seed.mode — re-applied under per-class (the default), applied once at class start under per-method.

Fixture format

A .json array or .ndjson stream. The layout is detected from the first non-whitespace character — [ for an array, { for NDJSON — and NDJSON elements are found by tracking brace depth rather than trusting newlines, so blank lines are ignored and one element may span several lines.

Each element is written in one of two forms.

Envelope form — an object carrying a value field. value is required (that is what makes it an envelope); key and headers are optional. A string value is published as its raw UTF-8 text; anything else is serialised with the suite's ObjectMapper — the same rule the publisher applies to a String versus a DTO.

seeds/orders.json — envelope form

[
  {
    "key": "order-1",
    "value": {
      "status": "NEW",
      "total": 4299
    }
  },
  {
    "key": "order-2",
    "value": {
      "status": "PAID",
      "total": 1999
    },
    "headers": {
      "source": "fixture",
      "X-Correlation-Id": "corr-2"
    }
  },
  {
    "key": "order-3",
    "value": "raw string payload, published as UTF-8 text"
  }
]

[
  {
    "key": "order-1",
    "value": {
      "status": "NEW",
      "total": 4299
    }
  },
  {
    "key": "order-2",
    "value": {
      "status": "PAID",
      "total": 1999
    },
    "headers": {
      "source": "fixture",
      "X-Correlation-Id": "corr-2"
    }
  },
  {
    "key": "order-3",
    "value": "raw string payload, published as UTF-8 text"
  }
]

Shorthand form — the element simply is the value, with a null key and no headers. Handy for a fixture where keys do not matter.

The form is decided per element — an object carrying a value field is an envelope, anything else is shorthand — and then required to be the same for the whole fixture. A mixed fixture is an error naming the offending index, because the alternative is a fixture where one line quietly means something different from the line above it.

Keys, headers and tombstones

Key. A string, or absent/null for a keyless record. A number or boolean key is written out as its text form — keys go over the wire as strings, and refusing them would be pedantry rather than safety. An array or object key is refused.
Headers. A JSON object of scalars, written {"name": "value"}. Kafka headers carry bytes, so a header bound to an array or object is refused; a null header value is allowed and published as a null header.
Tombstones. An explicit "value": null publishes a tombstone — the record a compacted topic's reader has to handle.

seeds/compacted-orders.json — an explicit null value is a tombstone

[
  {
    "key": "order-1",
    "value": {
      "status": "NEW"
    }
  },
  {
    "key": "order-1",
    "value": null
  }
]

[
  {
    "key": "order-1",
    "value": {
      "status": "NEW"
    }
  },
  {
    "key": "order-1",
    "value": null
  }
]

Avro fixtures

Point avroSchema at an .avsc file and the very same fixtures are published Avro-encoded instead of as JSON bytes. Each value is converted to a GenericRecord against that schema and serialized with the Confluent KafkaAvroSerializer against the mock schema registry — the same serializer KafkaTestPublisher.send(topic, key, avroRecord) uses, so a listener or a typed read sees exactly what it would have seen from a real producer.

TradeProjectionIT.java

@SpringBootTest
@EnableWiretapKafkaBroker
@KafkaSeed(locations = "classpath:seeds/trades.json",
           avroSchema = "classpath:avro/TradeEvent.avsc")   // class level: every method
class TradeProjectionIT {

    @Test
    void readsTheSeededTradesTyped() {
        var records = KafkaTopicReader.readRecords("trades", GenericRecord.class, 2);
        assertThat(records.get(0).value().get("symbol")).hasToString("ACME");
    }

    @Test
    @KafkaSeed(topic = "trades",
               locations = "classpath:seeds/one-odd-lot.json",
               avroSchema = "classpath:avro/TradeEvent.avsc")   // method level: this test only
    void rejectsAnOddLot() {
        // "trades" carries the two class fixtures plus this method's odd lot.
    }
}

@SpringBootTest
@EnableWiretapKafkaBroker
@KafkaSeed(locations = "classpath:seeds/trades.json",
           avroSchema = "classpath:avro/TradeEvent.avsc")   // class level: every method
class TradeProjectionIT {

    @Test
    void readsTheSeededTradesTyped() {
        var records = KafkaTopicReader.readRecords("trades", GenericRecord.class, 2);
        assertThat(records.get(0).value().get("symbol")).hasToString("ACME");
    }

    @Test
    @KafkaSeed(topic = "trades",
               locations = "classpath:seeds/one-odd-lot.json",
               avroSchema = "classpath:avro/TradeEvent.avsc")   // method level: this test only
    void rejectsAnOddLot() {
        // "trades" carries the two class fixtures plus this method's odd lot.
    }
}

The fixture stays plain JSON in either form, and everything around the value is unchanged: keys stay strings (only the value is Avro-encoded, and there is no key schema), headers stay UTF-8, "value": null still publishes a real tombstone rather than an encoded nothing, and the topic joins the per-method cleanup scope as any seeded topic does. Placeholders run first, on the JSON, so ${quantity} bound to an Integer lands as a JSON number and converts into an Avro int.

How the JSON maps onto the schema:

enum — a string that must be one of the symbols. bytes and fixed — a base64 string, a fixed checked against its declared size.
Unions are written as the plain value, or null. Avro's own JSON encoding wraps a branch — {"string": "x"} — which is hostile to write by hand, so wiretap resolves a union by matching the value against the branches instead: null takes the null branch, an object takes the first record branch whose fields without defaults are all present. First match wins.
Logical types pass through as their underlying representation: a timestamp-millis takes a number, a uuid a string, a decimal a base64 string. An ISO-8601 string is not parsed into a date.
Defaults fill in an absent field. A missing field with no default, a field the schema does not declare, an unknown enum symbol and a JSON string where a record was expected are all errors naming the fixture, the element, the JSON path and both types.

The schema belongs to the annotation, so all of its locations share it and two @KafkaSeeds on one class can carry different schemas — or one with a schema and one without, which is how a class mixes an Avro topic and a JSON topic. Suite-level seeds cannot use this: wiretap.test.seed.locations is a property and a property carries locations and nothing else, so suite fixtures stay JSON-only.

Setting avroSchema needs org.apache.avro:avro on the test classpath — wiretap declares it as an optional dependency, so a JSON-only suite does not inherit it — and io.confluent:kafka-avro-serializer for the encoding. Either one missing is reported with the coordinate to add.

Protobuf fixtures

Point protoMessage at a generated Protobuf message class and the same fixtures are published Protobuf-encoded instead of as JSON bytes. Where avroSchema takes a schema file, protoMessage takes the generated class: runtime .proto parsing needs protoc, but the class already carries its own descriptor.

ShipmentProjectionIT.java

@SpringBootTest
@EnableWiretapKafkaBroker
@KafkaSeed(locations = "classpath:seeds/shipments.json",
           protoMessage = ShipmentEvent.class)   // topic "shipments"
class ShipmentProjectionIT {

    @Test
    void readsTheSeededShipmentsTyped() {
        var records = KafkaTopicReader.readRecords("shipments", ShipmentEvent.class, 2);
        assertThat(records.get(0).value().getCarrier()).isEqualTo("ACME-EXPRESS");
    }
}

@SpringBootTest
@EnableWiretapKafkaBroker
@KafkaSeed(locations = "classpath:seeds/shipments.json",
           protoMessage = ShipmentEvent.class)   // topic "shipments"
class ShipmentProjectionIT {

    @Test
    void readsTheSeededShipmentsTyped() {
        var records = KafkaTopicReader.readRecords("shipments", ShipmentEvent.class, 2);
        assertThat(records.get(0).value().getCarrier()).isEqualTo("ACME-EXPRESS");
    }
}

Everything else matches Avro: the fixture stays plain JSON, keys stay strings, headers stay UTF-8, "value": null still publishes a real tombstone, and placeholders run first on the JSON. Each value is converted to a Message against the class's descriptor and serialized with the Confluent KafkaProtobufSerializer against the mock schema registry — the same serializer KafkaTestPublisher.send(topic, key, message) uses. The mapping is proto3's own canonical JSON (JsonFormat), not a wiretap invention: field names in lowerCamelCase or the original snake_case (the fixture above uses one of each on purpose), enums by name, bytes as base64 — and an unknown field in the fixture is an error, matching the Avro converter's strictness.

avroSchema and protoMessage are mutually exclusive on one annotation — a seed has one encoding. Mix an Avro topic and a Protobuf topic with two separate @KafkaSeeds, exactly as a class mixes an Avro topic and a JSON topic. Like avroSchema, protoMessage cannot be combined with multiSources, and suite-level seeds cannot use it either.

Requires com.google.protobuf:protobuf-java and protobuf-java-util on the test classpath (both optional wiretap dependencies) and io.confluent:kafka-protobuf-serializer for the encoding; either one missing is reported with the coordinate to add. See the publishing and reading pages for auto-detected Protobuf publish and typed/DynamicMessage reads outside of seeding, and remember: never assert on a Protobuf message's toString() — it is the text format, not JSON, unlike Avro.

Programmatic multi-topic sources

A fixture file seeds one topic. When a scenario spans several topics whose records reference one another — an orders event and the billing.invoices events for the same order id — a KafkaSeedSource builds the whole thing in Java, so a value generated once is reused everywhere it appears. Separate fixture files can only approximate that with a placeholder duplicated across each file.

WholeOrderCase.java

@SpringBootTest
@EnableWiretapKafkaBroker
@KafkaSeed(multiSources = WholeOrderCase.class)   // keys decide the topics
class OrderChoreographyIT {
}

class WholeOrderCase implements KafkaSeedSource {

    @Override
    public Map<String, ? extends List<?>> records(KafkaSeedContext ctx) {
        String orderId = ctx.uuid();                        // generated ONCE, reused across topics

        return Map.of(
            "orders", List.of(
                SeedEvent.of(new OrderPlaced(orderId, "GBP", 4200))
                    .key(orderId)
                    .header("event-type", "OrderPlaced")
                    .header("trace-id", ctx.uuid())),        // per-event headers

            "billing.invoices", List.of(
                SeedEvent.of(new InvoiceRaised(orderId, 4200)).key(orderId),
                SeedEvent.tombstone(orderId)),               // a null-valued record

            "audit.trades", List.of(
                SeedEvent.of(tradeEvent).key(orderId)));     // an Avro record -> Avro-encoded
                                                             // (one topic = one format)
    }
}

@SpringBootTest
@EnableWiretapKafkaBroker
@KafkaSeed(multiSources = WholeOrderCase.class)   // keys decide the topics
class OrderChoreographyIT {
}

class WholeOrderCase implements KafkaSeedSource {

    @Override
    public Map<String, ? extends List<?>> records(KafkaSeedContext ctx) {
        String orderId = ctx.uuid();                        // generated ONCE, reused across topics

        return Map.of(
            "orders", List.of(
                SeedEvent.of(new OrderPlaced(orderId, "GBP", 4200))
                    .key(orderId)
                    .header("event-type", "OrderPlaced")
                    .header("trace-id", ctx.uuid())),        // per-event headers

            "billing.invoices", List.of(
                SeedEvent.of(new InvoiceRaised(orderId, 4200)).key(orderId),
                SeedEvent.tombstone(orderId)),               // a null-valued record

            "audit.trades", List.of(
                SeedEvent.of(tradeEvent).key(orderId)));     // an Avro record -> Avro-encoded
                                                             // (one topic = one format)
    }
}

The source returns Map<topic, List<SeedEvent>>: the keys are the topics — created if absent, at the configured partition count, and added to the per-method cleanup scope exactly as a fixture's topic is — so the per-file attributes have nowhere to apply. Setting multiSources together with any of value, locations, topic, avroSchema, protoMessage or placeholders is refused, naming the offending attribute; a fixture seed and a multi-source seed can still coexist as two separate @KafkaSeeds.

Each element is a SeedEvent:

Value. SeedEvent.of(value) takes a POJO — serialized to JSON with the configured mapper — an Avro GenericRecord/SpecificRecord, or a Protobuf Message, each published through its own Confluent serializer. The value type alone decides the encoding, per record, so one source can seed an Avro topic, a Protobuf topic and a JSON topic all at once; mixing formats on a single topic is refused (tombstones exempt). SeedEvent.tombstone(key) publishes a null-valued record.
Key and headers. .key(k) and .header(name, value) build the envelope, and because each SeedEvent carries its own headers, different events in the same list carry different headers. A null header value is a real header with no bytes.

A source is resolved the way a SeedPlaceholders provider is — a Spring bean when the context declares one, so it can inject the application's own collaborators, and a no-argument constructor otherwise — and records(...) is re-invoked on every application with a fresh KafkaSeedContext, so ctx.uuid() and ctx.now() yield fresh values per test method, the same freshness a re-read fixture gets. What is fresh is the call and the context, not necessarily the instance: a no-arg source is built anew each time, but a bean source is a singleton by default, so compute each application's records from ctx rather than from instance state you mutate across calls. The KafkaSeedContext also exposes ctx.binding("app.tenant-id") for a test property and ctx.objectMapper() for hand-building a node.

The keys carry the same commitment a fixture's topic does — seeding a topic puts its whole content under per-method deletion — so on a pass-through cluster the developer owns, the records already on a seeded topic are deleted between methods too. A multi-source makes that easier to overlook, since the keys can be computed at runtime rather than sitting as a static string in a file, so keep them simple and obvious.

Suite-level seeds

The seeds every test class in the context starts from, applied before any class-level @KafkaSeed. Written either as a comma-separated list or as a YAML list — both forms are read, which matters because the flat property lookup alone would silently bind nothing for the YAML form.

Properties

wiretap.test.seed.locations=classpath:seeds/orders.json,classpath:seeds/payments.ndjson
wiretap.test.seed.mode=per-class

wiretap.test.seed.locations=classpath:seeds/orders.json,classpath:seeds/payments.ndjson
wiretap.test.seed.mode=per-class

Property	Default	Description
`wiretap.test.seed.locations` List<String>	—	Fixture files to apply before any class-level seed, in declaration order. Each entry names one file; the topic comes from its file name, since a suite seed carries no `topic` attribute.
`wiretap.test.seed.mode` String	`per-class`	`per-class` re-applies the suite fixture after every per-method cleanup, so each method sees the same starting state. `per-method` applies it once at class start, so a method that consumes it leaves the next one an empty topic — which is what you want when the test is “the second reader sees nothing”. Compared case-insensitively after trimming; any other value fails the context immediately, naming the two it accepts.

Suite seeds carry no placeholder providers of their own — they are configuration, so the only bindings available to them are the built-ins, the Environment and inline defaults.

Placeholders

Fixtures may contain ${name} placeholders, so one file can serve many tests. Resolution happens per occurrence: two ${#uuid} in one fixture are two different ids.

seeds/placeholder-orders.json

[
  {
    "key": "order-${orderId}",
    "value": {
      "customer": "${customerId}",
      "quantity": "${quantity}",
      "tenant": "${app.tenant-id}",
      "status": "${status:NEW}",
      "traceId": "${#uuid}",
      "createdAt": "${#now-P1D}",
      "dueAt": "${#now+PT2H}"
    }
  }
]

[
  {
    "key": "order-${orderId}",
    "value": {
      "customer": "${customerId}",
      "quantity": "${quantity}",
      "tenant": "${app.tenant-id}",
      "status": "${status:NEW}",
      "traceId": "${#uuid}",
      "createdAt": "${#now-P1D}",
      "dueAt": "${#now+PT2H}"
    }
  }
]

Property	Default	Description
`${name}`	—	Resolved from the seed's providers, then the Spring `Environment`, then the built-ins.
`${name:default}`	—	An inline default, used when nothing else bound `name`. Last resort before failure.
`$${name}`	—	Escape: emitted as a literal `${name}` and never resolved.
`${#uuid}` String	—	A fresh random UUID for this occurrence.
`${#now}` ISO-8601 instant	—	The current instant as a string. JSON has no date type, so the faithful rendering of an instant is a string — and a string is the same value whether it lands whole or interpolated inside a longer one, whatever the suite ObjectMapper does with dates.
`${#now+PT2H} / ${#now-P1D}` ISO-8601 instant	—	Shifted by an ISO-8601 duration (`PT2H`) or period (`P1D`). An offset that is neither is an error naming both forms.
`${#today}` ISO-8601 instant	—	Midnight UTC at the start of today.

A #-prefixed name that is not a built-in fails immediately rather than being reported as unbound — nothing you could bind would fix a typo'd built-in.

published literally as ${notAPlaceholder}

{ "value": { "template": "$${notAPlaceholder}" } }

{ "value": { "template": "$${notAPlaceholder}" } }

Built-ins

The built-ins and inline defaults need no binding at all, which is often the whole story for a fixture: ${#uuid} for a key that must be unique per run, ${#now-P1D} for “yesterday” in a test about overdue orders.

Bindings from providers

Annotations cannot carry object values, so named bindings come from SeedPlaceholders implementations listed in @KafkaSeed(placeholders = ...).

OrderPlaceholders.java

public class OrderPlaceholders implements SeedPlaceholders {

    private final CustomerCatalogue catalogue;   // injected when this is a bean

    public OrderPlaceholders(CustomerCatalogue catalogue) {
        this.catalogue = catalogue;
    }

    @Override
    public Map<String, Object> placeholders() {
        return Map.of(
            "orderId", "7",
            "customerId", catalogue.anyActiveCustomer().id(),
            "quantity", 3);        // 3 stays a JSON number, not "3"
    }
}

public class OrderPlaceholders implements SeedPlaceholders {

    private final CustomerCatalogue catalogue;   // injected when this is a bean

    public OrderPlaceholders(CustomerCatalogue catalogue) {
        this.catalogue = catalogue;
    }

    @Override
    public Map<String, Object> placeholders() {
        return Map.of(
            "orderId", "7",
            "customerId", catalogue.anyActiveCustomer().id(),
            "quantity", 3);        // 3 stays a JSON number, not "3"
    }
}

A provider is instantiated fresh for every seed application, so a binding computed in placeholders() is recomputed each time the fixture is re-applied. Under wiretap-spring-boot, a provider the context declares as a bean is resolved as one — which is what lets it inject the application's own collaborators; otherwise its no-argument constructor is used.

The chain, in order, first hit wins:

the seed's own providers, in declaration order
class-level providers inherited by a method seed (so method bindings win over class ones)
the Spring Environment, so "tenant": "${app.tenant-id}" picks up a test property with no provider involved
the built-ins
an inline default

Java

@SpringBootTest(
        // Name the configuration that declares the provider bean rather than relying on
        // nested-@TestConfiguration detection, so what makes the bean available is visible.
        classes = {TestApplication.class, SeedPlaceholderConfig.class},
        properties = "app.tenant-id=acme")
@EnableWiretapKafkaBroker
@KafkaSeed(locations = "classpath:seeds/placeholder-orders.json",
        placeholders = OrderPlaceholders.class)
class OrderSeedIT {
}

@SpringBootTest(
        // Name the configuration that declares the provider bean rather than relying on
        // nested-@TestConfiguration detection, so what makes the bean available is visible.
        classes = {TestApplication.class, SeedPlaceholderConfig.class},
        properties = "app.tenant-id=acme")
@EnableWiretapKafkaBroker
@KafkaSeed(locations = "classpath:seeds/placeholder-orders.json",
        placeholders = OrderPlaceholders.class)
class OrderSeedIT {
}

Typed values stay typed

A binding substituted for a whole JSON string value keeps its type: an Integer lands as a JSON number, a Map as a JSON object. The engine does this by parking whole-value bindings under unique tokens during the text rewrite and swapping them for real JSON nodes after parsing, so 3 does not arrive as "3".

A placeholder interpolated into a longer string, or into an object key, is substituted textually — so only a scalar binding can be used there. And a placeholder written outside a JSON string is a hard error: substitution happens before parsing, so an unquoted one could only ever produce invalid JSON.

When a placeholder is unbound

Nothing falls back to an empty string. Anything left unbound after the whole chain fails fast, and the report lists every miss with its file and JSON path rather than only the first — one run tells you everything the fixture needs.

The reverse case is a warning, not a failure: a binding name that no loaded fixture ever asked for is logged as a typo guard. Bindings a method seed inherits from its class are exempt, because an inherited binding legitimately goes unused by a method fixture and warning about that every time would train people to ignore the warning.

Related reading: Configuration for every wiretap.test.* key, Passthrough mode for what seeding does to a cluster you own, and Assertions for reading the seeded records back.