Skip to main content

Architecture

KTestify is built on a strict three-layer separation of concerns. Each layer has one job and knows nothing about the other layers' implementation details.

Since version 1.1.1, the transport layer has two sibling contracts instead of one. RecordFetcher<V> models a background stream you poll until a record appears (Kafka, Azure Blob). RequestResponseClient<Req, V> models a synchronous, caller-initiated call where you send a request right now and get an answer immediately (HTTP, gRPC, SOAP). Both return List<ConsumedRecord<V>>, so the orchestration and assertion layers below stay identical no matter which contract a given transport implements.

ConsumedRecord<V> is the only data type that crosses layer boundaries.

Synchronous transports additionally populate a new attributes map on ConsumedRecord (an HTTP status code, a future gRPC status code, an MQ reason code) for metadata that is not a protocol header. Asynchronous transports leave it empty. See Core Concepts → for the full field list.

Why the separation matters​

The three layers are connected by exactly one type, ConsumedRecord<V>. Because the transport layer only ever emits that type and the assertion layer only ever consumes it, the two are fully decoupled. The practical payoff is threefold:

  • Swap transports without touching assertions. A new broker is a new RecordFetcher (or RequestResponseClient) implementation. Every RecordMatcher keeps working because it only sees ConsumedRecord.
  • Reuse matchers across transports. The same FileRecordMatcher asserts a Kafka String body and an HTTP response body, because both arrive as ConsumedRecord<String>.
  • Stay test-framework agnostic. ktestify-cucumber (or any future adapter) drives the engine through ConsumerContext / ProducerContext and reads back ConsumedRecord. It never imports a broker client.

Layer responsibilities​

Transport - RecordFetcher<V>​

The transport layer turns a concrete message broker into a stream of ConsumedRecord<V>. The contract itself is deliberately tiny, a single fetch() method plus close():

public interface RecordFetcher<V> extends AutoCloseable {
    List<ConsumedRecord<V>> fetch() throws FetchException;
    void close();
}

fetch() blocks until at least one record that passes the configured filters is available, or the read timeout expires. It returns a non-empty, unmodifiable list, or throws FetchException.

The only implementation today is KafkaRecordFetcher, which knows Kafka brokers, partitions, offsets, and the deduplication registry. Swapping Kafka for IBM MQ means writing a new IbmMqRecordFetcher<V> that implements the same interface. Nothing in the layers above changes.

Transport, synchronous - RequestResponseClient<Req, V>​

Knows: how to send one request and turn the answer into ConsumedRecord. Does NOT know: matchers, files, test frameworks, and does not block waiting for something to "appear".

public interface RequestResponseClient<Req, V> extends AutoCloseable {
    List<ConsumedRecord<V>> execute(Req request) throws FetchException;
    void close();
}

This is the contract an HTTP client plugin implements. See Synchronous Transports → for the full guide.

Orchestration - AbstractKafkaConsumer​

Knows: fetch → match → result wiring. Does NOT know: Kafka internals, comparison algorithms.

// AbstractKafkaConsumer.call(), simplified
try (KafkaRecordFetcher<K, V> fetcher = new KafkaRecordFetcher<>(context)) {
    List<ConsumedRecord<V>> records = fetcher.fetch();      // transport
    MatchContext matchCtx = buildMatchContext();
    MatchResult result = matcher.match(records, matchCtx);  // assertion
    return result.isPassed();
} catch (FetchException e) {
    throw new ConsumerException(e.getMessage());
}

Orchestration, synchronous - AbstractSynchronousConsumer<Req, V>​

Knows: buildRequest → execute → match → result wiring, for transports where the caller supplies the request explicitly instead of the fetcher blocking on a subscription. Does NOT know: HTTP, gRPC, or any transport internals.

// AbstractSynchronousConsumer.call(), simplified
try {
    Req request = buildRequest();
    List<ConsumedRecord<V>> records = client.execute(request);   // transport
    MatchResult result = matcher.match(records, buildMatchContext()); // assertion
    return result.isPassed();
} catch (FetchException e) {
    throw new ConsumerException(e.getMessage());
}

Unlike AbstractKafkaConsumer, the client is not closed in a finally block here. A RequestResponseClient is expected to be a longer lived, connection pooled client (like java.net.http.HttpClient), owned and closed by the plugin's shared scenario resources, not created and discarded per call.

Assertion - RecordMatcher<V>​

Knows: ConsumedRecord, expected values, comparison algorithm. Does NOT know: Kafka, IBM MQ, any transport.

@FunctionalInterface
public interface RecordMatcher<V> {
    MatchResult match(List<ConsumedRecord<V>> records, MatchContext context);
}

Module boundaries​

ktestify-core                      ktestify-cucumber
───────────────────────────────    ──────────────────────────────────
RecordFetcher<V>                   BackgroundStepDefinition
RequestResponseClient<Req,V>        ValidationStepDefinition
KafkaRecordFetcher              ◄──────── ConsumerContext (config only)
AbstractKafkaConsumer              ConsumerValidationService.** It uses `ConsumerContext` / `ProducerContext` (ktestify-core abstractions) to configure the engine and receives only `ConsumedRecord<V>` back. This is what keeps the Cucumber layer free of broker specifics and lets the same step definitions drive Kafka, HTTP, or any future transport without modification
PollingRequestResponseClient
RecordMatcher<V>
MatchContext / MatchResult
ConsumedRecord<V>

ktestify-cucumber must never import org.apache.kafka.*. It uses ConsumerContext / ProducerContext (ktestify-core abstractions) to configure the engine and receives only ConsumedRecord<V> back.

Class hierarchy​

Matchers hierarchy​

RecordMatcher<V>  (@FunctionalInterface)
├── NoOpRecordMatcher<V>
├── FileRecordMatcher
├── XmlRecordMatcher
├── XPathRecordMatcher
├── FieldsRecordMatcher
├── FileKeyRecordMatcher
├── KeyRecordMatcher
├── AttributeRecordMatcher<V>
├── AvroFileRecordMatcher
├── AvroFileKeyRecordMatcher
├── AvroFieldsRecordMatcher
└── AvroKeyRecordMatcher

See also​