# Core Concepts > Key domain types in ktestify-core, ConsumedRecord, MatchContext, MatchResult, ConsumerContext, and more. Source: https://docs.ktestify.xyz/docs/extend/core-concepts ## `ConsumedRecord` The **only** data type that crosses layer boundaries. It is the universal output of the transport layer (`RecordFetcher` for asynchronous transports, `RequestResponseClient` 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`. ```java @Value public class ConsumedRecord { 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 headers; // protocol headers (Kafka headers, HTTP response headers, ...) Map attributes; // since 1.1.1, transport metadata, never null, defaults to emptyMap() static ConsumedRecord fromKafkaRecord(ConsumerRecord 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 →](matchers/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`. ```java @Value @Builder public class MatchContext { String matchMethod; List matchFilePaths; // replaces old matchFilePath (String) List excludedFields; // defaults to emptyList() boolean strictMatching; String matchKey; List 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 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. ```java @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` Immutable builder that configures a consumer operation. Built by `ConsumerValidationService` in `ktestify-cucumber` from DataTable values. Key fields: | Field | Type | Description | |---|---|---| | `topic` | `Topic` | The output topic, must be OUTPUT type, validated in the builder | | `matchMethod` | `String` | One of the `RecordMatcherFactory.METHOD_*` constants | | `matchFilePaths` | `List` | Expected file paths (single or batch) | | `excludedFields` | `List` | Field names to ignore in comparison | | `expectedRecordKey` | `String` | Key filter, the record must match this key | | `keyMatchStrategy` | `KeyMatchStrategy` | How `expectedRecordKey` is matched: `EXACT` (default), `CONTAINS`, `STARTS_WITH`, `ENDS_WITH`, `REGEX` | | `batchMatchStrategy` | `BatchMatchStrategy` | How batch records are matched against `matchFilePaths`: `POSITIONAL` (default) or `LAZY` | | `readTimeout` | `long` | Milliseconds | | `consumerDeltaTime` | `long` | Milliseconds (DataTable seconds × 1000) | | `isBatchConsumer` | `boolean` | Enables batch fetch mode | | `batchSize` | `int` | Number of records to collect in batch mode | --- ## `Topic` ```java @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` 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`. ```java @FunctionalInterface public interface RecordMatcher { MatchResult match(List> 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. ```java 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 →](matchers/built-in-matchers) for the full mapping table. --- ## Synchronous transports (`1.1.1`) Alongside `RecordFetcher` (async, poll until a record appears), `ktestify-core` now ships a sibling contract for synchronous, caller-initiated transports: ```java public interface RequestResponseClient extends AutoCloseable { List> execute(Req request) throws FetchException; void close(); } ``` Its orchestration counterpart, `AbstractSynchronousConsumer`, mirrors `AbstractKafkaConsumer` (build a request, call `execute`, hand the result to a `RecordMatcher`, return `MatchResult.isPassed()`), and a generic decorator, `PollingRequestResponseClient`, 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 →](transports/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 →](../write-tests/dynamic-variables) for the full list of built-in types.