Skip to main content

Core Concepts

ConsumedRecord<V>​

The only data type that crosses layer boundaries. It is the universal output of the transport layer (RecordFetcher<V> for asynchronous transports, RequestResponseClient<Req, V> for synchronous ones since 1.1.1) and the universal input of the assertion layer.

@Value
public class ConsumedRecord<V> {
    String  source;        // topic name, container name, request URL, ...
    int     partition;
    long    offset;
    String  key;
    V       value;         // String for raw, GenericRecord for Avro
    Instant timestamp;
    Map<String, String> headers;      // protocol headers (Kafka headers, HTTP response headers, ...)
    Map<String, String> attributes;   // NEW in 1.1.1, transport metadata, never null, defaults to emptyMap()

    static <V> ConsumedRecord<V> fromKafkaRecord(ConsumerRecord<String, V> record) { ... }
    MatchedRecord toMatchedRecord() { ... }
}

attributes holds structured transport metadata that does not belong under headers, for example an HTTP status code and elapsed time, a future gRPC status code, or an MQ reason code. Kafka and Azure Blob leave it empty. A synchronous transport plugin populates it, and AttributeRecordMatcher (see Built-in matchers →) asserts against it.

ConsumedRecord ships both a full constructor (accepting attributes) and a backward-compatible overload without it (defaults to Collections.emptyMap()), plus a @Builder, so existing transports keep compiling unchanged.

MatchedRecord​

Deduplication token, represents a record that has already been claimed by a consumer step. It holds topic + partition + offset + key + timestamp and is stored in the static deduplication registry.

MatchedRecord deliberately excludes processedTime from equals/hashCode so that two records from the same Kafka partition+offset are always considered the same, regardless of when they were processed.

MatchContext​

Immutable context object passed to a RecordMatcher. Built by AbstractKafkaConsumer.buildMatchContext() (or AbstractSynchronousConsumer.buildMatchContext() for synchronous transports) from the ConsumerContext.

@Value @Builder
public class MatchContext {
    String        matchMethod;
    List<String>  matchFilePaths;    // replaces old matchFilePath (String)
    List<String>  excludedFields;    // defaults to emptyList()
    boolean       strictMatching;
    String        matchKey;
    String        matchValue;
    Map<String,String> expectedAttributes;  // NEW in 1.1.1, defaults to emptyMap(), used by AttributeRecordMatcher

    // Convenience, for single-record matchers
    public String getMatchFilePath() {
        return matchFilePaths != null && !matchFilePaths.isEmpty()
            ? matchFilePaths.get(0) : null;
    }
}

expectedAttributes holds the key/value pairs to assert against a record's attributes map, for example {"statusCode": "200"}. It follows the same convention as excludedFields, always non-null, defaults to an empty map, and matchers check isEmpty() rather than null.

MatchResult​

Returned by every RecordMatcher. Carries a pass/fail flag plus the diff, expected, and actual strings for failure reporting.

@Value
public class MatchResult {
    boolean passed;
    String  diff;
    String  expected;
    String  actual;

    static MatchResult pass()  { ... }
    static MatchResult fail(String diff, String expected, String actual) { ... }
}

ConsumerContext<K, V>​

Immutable builder that configures a consumer operation. Built by ConsumerValidationService in ktestify-cucumber from DataTable values.

Key fields:

FieldTypeDescription
topicTopicThe output topic (must be OUTPUT type — validated in builder)
matchMethodStringOne of the ConfigConstants.method* values
matchFilePathsList<String>Expected file paths (single or batch)
excludedFieldsList<String>Field names to ignore in comparison
expectedRecordKeyStringKey filter — record must match this key
readTimeoutlongMilliseconds
consumerDeltaTimelongMilliseconds (DataTable seconds × 1000)
isBatchConsumerbooleanEnables batch fetch mode
batchSizeintNumber of records to collect in batch mode

Topic​

@Data @Builder
public class Topic {
    String     topicName;
    String     topicAlias;
    String     topicNamespace;
    Topic.Type topicType;          // INPUT or OUTPUT

    // Returns "namespace.topicName" or just "topicName" if no namespace
    String getNamespacedTopic() { ... }
}

RecordMatcherFactory​

Pure static factory, no DI, no singleton. Resolves the right RecordMatcher implementation based on matchMethod and whether the consumer is raw or Avro.

RecordMatcherFactory.forRaw("matchFile")   → FileRecordMatcher
RecordMatcherFactory.forAvro("matchFile")  → AvroFileRecordMatcher
RecordMatcherFactory.forRaw("matchXML")    → XmlRecordMatcher
RecordMatcherFactory.forAvro("matchXML")   → throws ConsumerException  ← not supported
RecordMatcherFactory.forRaw("methodMatchAttributes")  → AttributeRecordMatcher<>()   // NEW in 1.1.1, raw only

See Built-in matchers → for the full mapping table.

Synchronous transports (1.1.1)​

Alongside RecordFetcher<V> (async, poll until a record appears), ktestify-core now ships a sibling contract for synchronous, caller-initiated transports:

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

Its orchestration counterpart, AbstractSynchronousConsumer<Req, V>, mirrors AbstractKafkaConsumer (build a request, call execute, hand the result to a RecordMatcher, return MatchResult.isPassed()), and a generic decorator, PollingRequestResponseClient<Req, V>, adds retry-until-predicate-or-timeout semantics that any synchronous transport plugin can reuse instead of writing its own poll loop.

Full details, including the client lifecycle and how to implement a new synchronous transport, live on the Synchronous transports → page.

Dynamic variable system​

All file reads go through FileUtils.getFileContent(path), which transparently calls DynamicVariableProcessor.process(content) before returning. To add a new variable type:

  1. Implement DynamicVariable interface.
  2. Register it in DynamicVariableFactory.
  3. No changes needed to FileUtils or DynamicVariableProcessor.

See Dynamic variables → for the full list of built-in types.