Read

Reading records

KafkaTopicReader — the low-friction way to consume records in integration tests without building ad-hoc consumers in each class.

Package: io.typecheck.wiretap.core

Auto-detection

The reader picks the right deserialiser based on the target type you provide:

Property	Default	Description
`SpecificRecord subclass`	—	Confluent `KafkaAvroDeserializer` with `specific.avro.reader=true`.
`GenericRecord`	—	Confluent `KafkaAvroDeserializer` with `specific.avro.reader=false`.
`Protobuf-generated Message subclass`	—	Confluent `KafkaProtobufDeserializer` with `specific.protobuf.value.type` naming the class.
`com.google.protobuf.DynamicMessage`	—	Same `KafkaProtobufDeserializer`, generic form — no generated class required.
`Anything else (DTOs, POJOs)`	—	Jackson 3 JSON deserializer.

String records

Property	Default	Description
`readRecords(topic)`	—	Reads available string records within the default timeout.
`readRecords(topic, expectedCount)`	—	Waits for a specific number of string records.
`readRecords(topic, expectedCount, timeout)`	—	Same, with a custom timeout.
`readSingleRecord(topic)`	—	Reads one string record.
`readSingleRecord(topic, timeout)`	—	Reads one string record with a custom timeout.

Java

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

ConsumerRecord<String, String> record =
    KafkaTopicReader.readSingleRecord("orders-topic");

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

ConsumerRecord<String, String> record =
    KafkaTopicReader.readSingleRecord("orders-topic");

Typed records (JSON, Avro and Protobuf)

Pass a Class<V> to readRecords() or readSingleRecord() and the deserialiser is selected automatically.

JSON DTO

Java

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

ConsumerRecord<String, OrderEvent> record =
    KafkaTopicReader.readSingleRecord("orders-topic", OrderEvent.class);

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

ConsumerRecord<String, OrderEvent> record =
    KafkaTopicReader.readSingleRecord("orders-topic", OrderEvent.class);

Avro GenericRecord

Java

KafkaTestPublisher.send("avro-topic", "avro-key", avroRecord);

ConsumerRecord<String, GenericRecord> record =
    KafkaTopicReader.readSingleRecord("avro-topic", GenericRecord.class);

KafkaTestPublisher.send("avro-topic", "avro-key", avroRecord);

ConsumerRecord<String, GenericRecord> record =
    KafkaTopicReader.readSingleRecord("avro-topic", GenericRecord.class);

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

ConsumerRecord<String, TradeEvent> record =
    KafkaTopicReader.readSingleRecord("trades-topic", TradeEvent.class);

TradeEvent trade = TradeEvent.newBuilder()
    .setTradeId("trade-1")
    .setSymbol("AAPL")
    .setQuantity(100)
    .setPrice(150.25)
    .build();

KafkaTestPublisher.send("trades-topic", trade.getTradeId().toString(), trade);

ConsumerRecord<String, TradeEvent> record =
    KafkaTopicReader.readSingleRecord("trades-topic", TradeEvent.class);

Protobuf: generated class and DynamicMessage

A generated message class reads back typed; DynamicMessage reads back generic with no generated class needed — the Protobuf analogue of GenericRecord.

Java

ShipmentEvent shipment = ShipmentEvent.newBuilder()
    .setShipmentId("ship-1")
    .setCarrier("ACME-EXPRESS")
    .setParcelCount(3)
    .setWeightKg(12.5)
    .build();

KafkaTestPublisher.send("shipments-topic", "ship-1", shipment);

// Typed, into the generated class
ConsumerRecord<String, ShipmentEvent> record =
    KafkaTopicReader.readSingleRecord("shipments-topic", ShipmentEvent.class);

// Generic, with no generated class on this classpath
ConsumerRecord<String, DynamicMessage> generic =
    KafkaTopicReader.readSingleRecord("shipments-topic", DynamicMessage.class);

ShipmentEvent shipment = ShipmentEvent.newBuilder()
    .setShipmentId("ship-1")
    .setCarrier("ACME-EXPRESS")
    .setParcelCount(3)
    .setWeightKg(12.5)
    .build();

KafkaTestPublisher.send("shipments-topic", "ship-1", shipment);

// Typed, into the generated class
ConsumerRecord<String, ShipmentEvent> record =
    KafkaTopicReader.readSingleRecord("shipments-topic", ShipmentEvent.class);

// Generic, with no generated class on this classpath
ConsumerRecord<String, DynamicMessage> generic =
    KafkaTopicReader.readSingleRecord("shipments-topic", DynamicMessage.class);

Custom deserialisers

Java

List<ConsumerRecord<String, String>> records = KafkaTopicReader.readRecords(
    "orders-topic",
    2,
    Duration.ofSeconds(5),
    new StringDeserializer(),
    new StringDeserializer()
);

List<ConsumerRecord<String, String>> records = KafkaTopicReader.readRecords(
    "orders-topic",
    2,
    Duration.ofSeconds(5),
    new StringDeserializer(),
    new StringDeserializer()
);