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. The class itself has existed since 0.3.0; the attributes map was added in 1.1.1.

@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;   // since 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. The attributes field is documented as never null and defaults to an empty map, so matchers can safely call isEmpty() rather than checking for null.

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 maintained by KafkaRecordFetcher.

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. The processedTime field is annotated with Lombok @With, meaning it is a mutable-on-copy timestamp used for reporting, not for identity.

The conversion is one way and cheap: ConsumedRecord.toMatchedRecord() builds the token, and KafkaRecordFetcher calls MATCHED_RECORDS.contains(record.toMatchedRecord()) before claiming a record and MATCHED_RECORDS.add(...) after. The registry is a ConcurrentHashMap.newKeySet() shared across all fetcher instances in the JVM, and KafkaRecordFetcher.clearMatchedRecords() wipes it. In the Cucumber flow that clear happens in the @Before hook at the start of every scenario, which is why two steps in the same scenario never double-claim a record. If you use KafkaRecordFetcher directly outside Cucumber, you must call clearMatchedRecords() yourself before each test.

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;
    List<String>  matchKeys;         // defaults to emptyList(), expected keys for batch matching
    String        matchValue;
    KeyMatchStrategy keyMatchStrategy;  // defaults to EXACT, controls how expectedRecordKey(s) are matched
    BatchMatchStrategy batchMatchStrategy; // defaults to POSITIONAL, controls batch file/key matching order
    Map<String,String> expectedAttributes;  // since 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: both default to an empty collection rather than null, and matchers check isEmpty() instead of null. The default buildMatchContext() in AbstractKafkaConsumer sets strictMatching to false and leaves matchKey, matchValue, and expectedAttributes unset.

keyMatchStrategy controls how expectedRecordKey is matched against the actual record key. It defaults to KeyMatchStrategy.EXACT for backward compatibility. Available strategies: EXACT, CONTAINS, STARTS_WITH, ENDS_WITH, REGEX. The strategy is applied at two layers: the fetch-time pre-filter in KafkaRecordFetcher.passesKeyFilter() (to skip non-matching records during polling) and the assertion-time key matchers (KeyRecordMatcher, FileKeyRecordMatcher, AvroKeyRecordMatcher, AvroFileKeyRecordMatcher).

batchMatchStrategy controls how batch matchers (FileRecordMatcher and AvroFileRecordMatcher) compare multiple consumed records against matchFilePaths. It defaults to BatchMatchStrategy.POSITIONAL (record[i] must match matchFilePaths.get(i)). When set to BatchMatchStrategy.LAZY, records are matched against matchFilePaths in any order using 1-to-1 matching without replacement.

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 pass(String expected, String actual)  { ... }
    static MatchResult fail(String diff, String expected, String actual) { ... }
    static MatchResult fail(String message)  { ... }
}

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 the builder
matchMethodStringOne of the RecordMatcherFactory.METHOD_* constants
matchFilePathsList<String>Expected file paths (single or batch)
excludedFieldsList<String>Field names to ignore in comparison
expectedRecordKeyStringKey filter, the record must match this key
keyMatchStrategyKeyMatchStrategyHow expectedRecordKey is matched: EXACT (default), CONTAINS, STARTS_WITH, ENDS_WITH, REGEX
batchMatchStrategyBatchMatchStrategyHow batch records are matched against matchFilePaths: POSITIONAL (default) or LAZY
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;
    TopicNamespace  topicNamespace;   // nested, holds namespace + namespaceAlias
    Topic.Type      topicType;        // INPUT or OUTPUT

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

RecordMatcher<V>​

The assertion contract. A matcher receives the records fetched by a RecordFetcher (or returned by a RequestResponseClient) and asserts them against the expected state carried by a MatchContext.

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

Implementations have zero dependency on Kafka, HTTP, or any transport. They only know about ConsumedRecord, which is exactly why the same matcher works for every transport. Concrete implementations live in io.github.ktestify.match.impl: NoOpRecordMatcher (always passes, for consume-only scenarios), FileRecordMatcher, XmlRecordMatcher, XPathRecordMatcher, FieldsRecordMatcher, FileKeyRecordMatcher, KeyRecordMatcher, AttributeRecordMatcher, and their Avro* counterparts.

RecordMatcherFactory​

Pure static factory, no DI, no singleton. Resolves the right RecordMatcher implementation based on matchMethod and whether the consumer is raw (String) or Avro (GenericRecord). Two typed entry points, forRaw(String) and forAvro(String), let the compiler enforce the value type.

RecordMatcherFactory.forRaw("methodMatchFile")        → FileRecordMatcher
RecordMatcherFactory.forAvro("methodMatchFile")       → AvroFileRecordMatcher
RecordMatcherFactory.forRaw("methodMatchXML")         → XmlRecordMatcher
RecordMatcherFactory.forAvro("methodMatchXML")        → throws ConsumerException  (not supported for Avro)
RecordMatcherFactory.forRaw("methodMatchAttributes")  → AttributeRecordMatcher<>()   // since 1.1.1, raw only

When matchMethod is null or blank, both methods return a NoOpRecordMatcher, which makes consume-only scenarios a first-class use case. The Avro path supports only methodMatchFile, methodMatchKeyValue, methodFieldsToMatch, and methodRecordKeyMatch. XML, XPath, and attribute matching are raw-only, so requesting them for an Avro topic throws ConsumerException. 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.