Class ColumnarTestDriver

java.lang.Object
io.krabka.streams.test.ColumnarTestDriver
public final class ColumnarTestDriver extends Object
Runs a columnar topology without a broker.

The driver mirrors Kafka's TopologyTestDriver for the columnar execution model: records piped in are processed synchronously on the calling thread, and sink output is captured per topic in FIFO order. Because the driver wraps one BuiltColumnarTopology, stateful operators accumulate across calls exactly as they would across polls in production.

The driver does not close the topology; close the BuiltColumnarTopology (and its allocator) yourself, ideally with try-with-resources in the test.

Example

try (var built = topology.build()) {
    var driver = new ColumnarTestDriver(built);
    driver.pipeInput("input", 0, key, value, 0L);

    var output = driver.readOutput("output");
    assertThat(driver.isOutputEmpty("output")).isTrue();
}

  • Constructor Details

    • ColumnarTestDriver

      public ColumnarTestDriver(BuiltColumnarTopology topology)
      Creates a driver around a built topology.
      Parameters:
      topology - the topology to feed; the driver does not close it

  • Method Details

    • pipeInput

      public void pipeInput(String topic, int partition, byte[] key, byte[] value, long timestamp)
      Runs one record through the topology. Offsets start at zero for each topic partition.
      Parameters:
      topic - the source topic to pipe the record into
      partition - the source partition number
      key - the record key, or null for a keyless record
      value - the record value
      timestamp - the record timestamp in epoch milliseconds

    • pipeInput

      public void pipeInput(String topic, int partition, byte[] key, byte[] value, long timestamp, List<RecordHeader> headers)
      Runs one record, including immutable Kafka headers, through the topology.
      Parameters:
      topic - the source topic to pipe the record into
      partition - the source partition number
      key - the record key, or null for a keyless record
      value - the record value
      timestamp - the record timestamp in epoch milliseconds
      headers - the ordered Kafka headers to attach

    • pipeBatch

      public void pipeBatch(String topic, List<ConsumedRecord> records)
      Runs one fetched partition batch through the topology.

      All records are evaluated as one graph pass, which is how a production runner hands a poll's worth of records to the topology. Sink output is appended to the per-topic queues read by readOutput(String).

      Parameters:
      topic - the source topic the records were fetched from
      records - the records of the batch, all from the same topic
      Throws:
      RuntimeException - the fault registered with failNext(RuntimeException), if one is pending

    • failNext

      public void failNext(RuntimeException fault)
      Throws the supplied fault before the next batch evaluation, once.

      Registered faults are queued: each piped batch consumes one fault. Use this to test error policies and retry paths without a failing codec or operator.

      Parameters:
      fault - the exception the next pipeInput or pipeBatch call throws

    • isOutputEmpty

      public boolean isOutputEmpty(String topic)
      Returns whether a topic has no unread output.
      Parameters:
      topic - the sink topic to check
      Returns:
      true when every record produced to the topic has been read

    • outputSize

      public int outputSize(String topic)
      Returns the number of unread records for a topic.
      Parameters:
      topic - the sink topic to check
      Returns:
      the number of records available to readOutput(String)

    • readOutput

      public ProduceRecord readOutput(String topic)
      Reads the next record or throws if the topic has no output.
      Parameters:
      topic - the sink topic to read from
      Returns:
      the oldest unread record produced to the topic
      Throws:
      NoSuchElementException - if the topic has no unread output

    • drainOutput

      public List<ProduceRecord> drainOutput(String topic)
      Removes and returns all records for a topic.
      Parameters:
      topic - the sink topic to drain
      Returns:
      every unread record in production order; empty when there is none