Wiretap

Publish

Publishing records

KafkaTestPublisher — push test data into Kafka without building temporary producers in every test class.

Package: io.typecheck.wiretap.core

Auto-detection

KafkaTestPublisher picks the right serialiser based on value type:

PropertyDefaultDescription
String—StringSerializer
Avro GenericRecord or SpecificRecord—Confluent KafkaAvroSerializer (mock schema registry)
Protobuf com.google.protobuf.Message—Confluent KafkaProtobufSerializer (same mock schema registry)
Anything else (DTOs, POJOs)—Jackson 3 JSON serializer

String records

PropertyDefaultDescription
send(topic, key, value)—Sends one keyed string record.
send(topic, value)—Sends one string record without a key.
send(topic, partition, key, value)—Sends to a specific partition.
send(topic, key, value, headers)—Sends with Iterable<Header> or UTF-8 Map<String,String> headers.
Java
KafkaTestPublisher.send("orders-topic", "order-123", "created");
KafkaTestPublisher.send("orders-topic", 0, "order-124", "priority");

Object publishing (JSON, Avro and Protobuf)

JSON DTO

Java
OrderEvent dto = OrderEvent.create("Widget", 2, new BigDecimal("19.99"));
KafkaTestPublisher.send("orders-topic", dto.getOrderId(), dto);

Avro GenericRecord

Java
GenericRecord avroRecord = new GenericData.Record(schema);
avroRecord.put("id", "avro-1");
avroRecord.put("message", "hello");
KafkaTestPublisher.send("avro-topic", "avro-key", avroRecord);

Avro SpecificRecord

Java
TradeEvent trade = TradeEvent.newBuilder()
    .setTradeId("trade-1")
    .setSymbol("AAPL")
    .setQuantity(100)
    .setPrice(150.25)
    .build();
KafkaTestPublisher.send("trades-topic", trade.getTradeId().toString(), trade);

Protobuf Message

Auto-detected the same way Avro is — anything that is a com.google.protobuf.Message uses the Confluent KafkaProtobufSerializer pointed at the mock registry:

Java
ShipmentEvent shipment = ShipmentEvent.newBuilder()
    .setShipmentId("ship-1")
    .setCarrier("ACME-EXPRESS")
    .setParcelCount(3)
    .setWeightKg(12.5)
    .build();
KafkaTestPublisher.send("shipments-topic", "ship-1", shipment);

Headers

Use Map<String, String> for the common UTF-8 case:

Java
KafkaTestPublisher.send(
    "orders-topic",
    order.getOrderId(),
    order,
    Map.of("X-Correlation-Id", "corr-123")
);

For binary header values, pass Kafka Header instances:

Java
KafkaTestPublisher.send(
    "trades-topic",
    trade.getTradeId().toString(),
    trade,
    List.of(new RecordHeader("X-Trace-Id", traceIdBytes))
);

Batch publishing

PropertyDefaultDescription
sendBatch(topic, Map<String, ?> records)—Sends a batch of keyed records.
sendBatch(topic, List<?> values)—Sends a batch of unkeyed records.
Java
KafkaTestPublisher.sendBatch("orders-topic", List.of(dto1, dto2));

Explicit bootstrap servers

Every method above has a twin whose first argument is a bootstrap-servers address. The forms without it publish to the broker the most recently started test class is wired to, which is what you want inside a test; the explicit forms are exact, which is what you want in a helper that outlives one test class or in a suite that mixes embedded and passthrough classes.

Java
// Whatever cluster this context is wired to — embedded or passthrough
@Value("${spring.kafka.bootstrap-servers}")
private String bootstrapServers;

// Or, in an embedded suite, straight from the broker bean
@Autowired
private EmbeddedKafkaBroker broker;   // broker.getBrokersAsString()

KafkaTestPublisher.send(bootstrapServers, "orders-topic", "order-123", order);
KafkaTestPublisher.sendBatch(bootstrapServers, "orders-topic", List.of(dto1, dto2));

sendValue — keyless at an address

There is deliberately no three-argument send(bootstrapServers, topic, value): that signature is (String, String, String), which already means send(topic, key, value) against the broker under test. To publish without a key at an explicit address, name it — or pass a null key.

Java
// Both publish a keyless record at an explicit address
KafkaTestPublisher.sendValue(bootstrapServers, "orders-topic", order);
KafkaTestPublisher.send(bootstrapServers, "orders-topic", null, order);
PropertyDefaultDescription
send(bootstrapServers, topic, key, value)—Keyed record at an explicit address; value may be a String, DTO, Avro record, or Protobuf message.
sendValue(bootstrapServers, topic, value)—Keyless record at an explicit address, with the same auto-detection.
send(bootstrapServers, topic, partition, key, value)—Specific partition at an explicit address.
send(bootstrapServers, topic, key, value, headers)—With Iterable<Header> or UTF-8 Map<String,String> headers.
sendBatch(bootstrapServers, topic, records)—Batch of keyed records (Map) or unkeyed values (List) at an explicit address.

Custom serialisers

Java
KafkaTestPublisher.send(
    "orders-topic",
    "order-123",
    "created",
    new StringSerializer(),
    new StringSerializer()
);