diff --git a/.github/workflows/changelog.yml b/.github/workflows/changelog.yml
index 3218ff2..96814a8 100644
--- a/.github/workflows/changelog.yml
+++ b/.github/workflows/changelog.yml
@@ -30,7 +30,7 @@ on:
jobs:
changelog:
- permissions:
+ permissions :
contents: write # update-changelog job pushes CHANGELOG.md commits via checkout
pull-requests: write # preview-changelog job posts/updates the PR comment
uses: ktestify/.github/.github/workflows/reusable-changelog.yml@main
diff --git a/src/main/java/io/github/ktestify/io/core/AbstractSynchronousConsumer.java b/src/main/java/io/github/ktestify/io/core/AbstractSynchronousConsumer.java
new file mode 100644
index 0000000..9eac21d
--- /dev/null
+++ b/src/main/java/io/github/ktestify/io/core/AbstractSynchronousConsumer.java
@@ -0,0 +1,119 @@
+/*
+ * Copyright 2026 Nil MALHOMME (malhomme.nil+oss@icloud.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.github.ktestify.io.core;
+
+import io.github.ktestify.exceptions.ConsumerException;
+import io.github.ktestify.exceptions.FetchException;
+import io.github.ktestify.match.MatchContext;
+import io.github.ktestify.match.MatchResult;
+import io.github.ktestify.match.RecordMatcher;
+import io.github.ktestify.models.ConsumedRecord;
+import java.util.List;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Thin coordinator that wires a {@link RequestResponseClient} (transport) with a {@link RecordMatcher} (assertion) for
+ * synchronous, caller-initiated transports (HTTP, gRPC, SOAP, …), and exposes a single {@link #call()} entry point.
+ *
+ *
Sibling to {@code io.github.ktestify.io.kafka.AbstractKafkaConsumer}: same execute → match → return shape, adapted
+ * for a transport where the caller supplies the request explicitly instead of the fetcher blocking on a subscription.
+ *
+ *
This class contains no transport mechanics and no matching logic. Those
+ * responsibilities belong exclusively to {@link RequestResponseClient} and {@link RecordMatcher} respectively.
+ *
+ *
Client lifecycle
+ *
+ *
Unlike {@code AbstractKafkaConsumer}, this class does not close the client in a {@code finally}
+ * block. A {@link RequestResponseClient} is expected to be a longer-lived, connection-pooled client (like
+ * {@code java.net.http.HttpClient}) owned and closed by the plugin's shared scenario resources, not created and
+ * discarded per request.
+ *
+ * @param the request type
+ * @param the record value type
+ * @since 1.1.1
+ */
+@Slf4j
+public abstract class AbstractSynchronousConsumer extends AbstractConsumer {
+
+ protected final RequestResponseClient client;
+ protected final RecordMatcher matcher;
+
+ /**
+ * Primary constructor.
+ *
+ * @param properties the consumer properties map
+ * @param client the synchronous transport used to send the request
+ * @param matcher the assertion strategy to apply to the response
+ */
+ protected AbstractSynchronousConsumer(
+ Map properties, RequestResponseClient client, RecordMatcher matcher) {
+ super(properties);
+ this.client = client;
+ this.matcher = matcher;
+ log.debug(
+ "AbstractSynchronousConsumer created with client '{}' and matcher '{}'",
+ client.getClass().getSimpleName(),
+ matcher.getClass().getSimpleName());
+ }
+
+ /**
+ * Builds the transport-specific request to send. Called once per {@link #call()} invocation.
+ *
+ * @return the request to hand to {@link RequestResponseClient#execute(Object)}
+ */
+ protected abstract Req buildRequest();
+
+ /**
+ * Builds the {@link MatchContext} that is passed to the matcher.
+ *
+ *
Subclasses typically read this from their own context object (analogous to {@code ConsumerContext} for Kafka).
+ *
+ * @return the match context for this invocation
+ */
+ protected abstract MatchContext buildMatchContext();
+
+ /**
+ * Sends the request, then asserts the response with the configured matcher.
+ *
+ *
Lifecycle:
+ *
+ *
+ *
Build the request via {@link #buildRequest()}.
+ *
Call {@link RequestResponseClient#execute(Object)} — blocks until the response arrives or fails.
+ *
Pass the resulting records to {@link RecordMatcher#match(List, MatchContext)}.
+ *
+ *
+ * @return {@code true} if the matcher passed, {@code false} otherwise
+ * @throws ConsumerException if the execute or match step throws an unrecoverable error
+ */
+ @Override
+ public Boolean call() throws ConsumerException {
+ try {
+ Req request = buildRequest();
+ List> records = client.execute(request);
+ MatchContext matchContext = buildMatchContext();
+ MatchResult result = matcher.match(records, matchContext);
+
+ log.debug("Match result: passed={}, diff={}", result.isPassed(), result.getDiff());
+
+ return result.isPassed();
+
+ } catch (FetchException e) {
+ throw new ConsumerException(e.getMessage());
+ }
+ }
+}
diff --git a/src/main/java/io/github/ktestify/io/core/PollingRequestResponseClient.java b/src/main/java/io/github/ktestify/io/core/PollingRequestResponseClient.java
new file mode 100644
index 0000000..db385bf
--- /dev/null
+++ b/src/main/java/io/github/ktestify/io/core/PollingRequestResponseClient.java
@@ -0,0 +1,123 @@
+/*
+ * Copyright 2026 Nil MALHOMME (malhomme.nil+oss@icloud.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.github.ktestify.io.core;
+
+import io.github.ktestify.exceptions.FetchException;
+import io.github.ktestify.models.ConsumedRecord;
+import java.util.List;
+import java.util.function.Predicate;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Generic {@link RequestResponseClient} decorator that retries {@link #execute(Object)} against a delegate client until
+ * a caller-supplied predicate on the resulting records passes, or a timeout elapses.
+ *
+ *
Useful for "eventually consistent" APIs, for example asserting that an HTTP endpoint eventually returns 200 once
+ * an asynchronous side-effect completes, without every plugin re-implementing its own poll loop.
+ *
+ *
On timeout the last result obtained is returned rather than throwing, so the subsequent
+ * {@code RecordMatcher} failure message shows the real final state instead of a generic timeout string. A
+ * {@link FetchException} raised by the delegate is only propagated when no successful attempt has been made yet.
+ *
+ * @param the request type
+ * @param the record value type
+ * @since 1.1.1
+ */
+@Slf4j
+public class PollingRequestResponseClient implements RequestResponseClient {
+
+ private final RequestResponseClient delegate;
+ private final Predicate>> untilPredicate;
+ private final long timeoutMs;
+ private final long pollIntervalMs;
+
+ /**
+ * Creates a polling decorator.
+ *
+ * @param delegate the underlying client actually performing the call
+ * @param untilPredicate the success condition evaluated against each result
+ * @param timeoutMs total time budget in milliseconds
+ * @param pollIntervalMs sleep between attempts in milliseconds
+ */
+ public PollingRequestResponseClient(
+ RequestResponseClient delegate,
+ Predicate>> untilPredicate,
+ long timeoutMs,
+ long pollIntervalMs) {
+ this.delegate = delegate;
+ this.untilPredicate = untilPredicate;
+ this.timeoutMs = timeoutMs;
+ this.pollIntervalMs = pollIntervalMs;
+ }
+
+ /**
+ * Repeatedly executes {@code request} until the predicate passes or the timeout elapses.
+ *
+ * @param request the request to send
+ * @return the first result satisfying the predicate, or the last result obtained before the timeout
+ * @throws FetchException if the delegate fails and no result has ever been obtained, or the thread is interrupted
+ */
+ @Override
+ public List> execute(Req request) throws FetchException {
+ long deadlineMs = System.currentTimeMillis() + timeoutMs;
+ List> lastResult = null;
+ FetchException lastFailure = null;
+
+ do {
+ try {
+ List> result = delegate.execute(request);
+ lastResult = result;
+ lastFailure = null;
+ if (untilPredicate.test(result)) {
+ log.debug("Polling predicate satisfied.");
+ return result;
+ }
+ } catch (FetchException e) {
+ log.debug("Polling attempt failed: {}", e.getMessage());
+ lastFailure = e;
+ }
+
+ if (System.currentTimeMillis() + pollIntervalMs >= deadlineMs) {
+ break;
+ }
+ sleep(pollIntervalMs);
+ } while (true);
+
+ if (lastResult == null) {
+ throw lastFailure != null
+ ? lastFailure
+ : new FetchException("Polling produced no result within " + timeoutMs + "ms.");
+ }
+
+ log.warn("Polling predicate never satisfied within {}ms, returning the last result obtained.", timeoutMs);
+ return lastResult;
+ }
+
+ /** Closes the delegate client. Idempotent. */
+ @Override
+ public void close() {
+ delegate.close();
+ }
+
+ private static void sleep(long ms) throws FetchException {
+ try {
+ Thread.sleep(ms);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new FetchException("Interrupted while polling for a response.");
+ }
+ }
+}
diff --git a/src/main/java/io/github/ktestify/io/core/RequestResponseClient.java b/src/main/java/io/github/ktestify/io/core/RequestResponseClient.java
new file mode 100644
index 0000000..4dc4c31
--- /dev/null
+++ b/src/main/java/io/github/ktestify/io/core/RequestResponseClient.java
@@ -0,0 +1,64 @@
+/*
+ * Copyright 2026 Nil MALHOMME (malhomme.nil+oss@icloud.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.github.ktestify.io.core;
+
+import io.github.ktestify.exceptions.FetchException;
+import io.github.ktestify.models.ConsumedRecord;
+import java.util.List;
+
+/**
+ * Transport-agnostic contract for synchronous, caller-initiated request/response transports (HTTP, gRPC, SOAP, …).
+ *
+ *
Sibling contract to {@link RecordFetcher}. Where {@code RecordFetcher} models "background stream, block until a
+ * record appears" (Kafka, Azure Blob polling), {@code RequestResponseClient} models "send a request right now and get
+ * an answer immediately".
+ *
+ *
Both contracts return {@link ConsumedRecord}, the common currency shared with every {@code RecordMatcher}, so the
+ * entire assertion layer is reused unchanged regardless of which contract a transport implements.
+ *
+ *
+ *
+ * @param the request type specific to the transport (e.g. an HTTP request spec)
+ * @param the type of the resulting record value (e.g. {@code String} for an HTTP body)
+ * @since 1.1.1
+ */
+public interface RequestResponseClient extends AutoCloseable {
+
+ /**
+ * Sends {@code request} and returns the result wrapped as a (typically single-element) list of
+ * {@link ConsumedRecord}.
+ *
+ *
A list is used, not a single object, purely to keep perfect symmetry with {@link RecordFetcher#fetch()} so
+ * both contracts feed the same {@code RecordMatcher} signature unchanged.
+ *
+ * @param request the request to send
+ * @return a non-null, non-empty list of consumed records (normally exactly one)
+ * @throws FetchException if the call fails (connection error, timeout, non-recoverable transport error)
+ */
+ List> execute(Req request) throws FetchException;
+
+ /**
+ * Releases all resources held by this client (connection pools, threads, etc.). Idempotent, calling {@code close()}
+ * more than once must be safe.
+ */
+ @Override
+ void close();
+}
diff --git a/src/main/java/io/github/ktestify/match/MatchContext.java b/src/main/java/io/github/ktestify/match/MatchContext.java
index b500ab9..58dba8b 100644
--- a/src/main/java/io/github/ktestify/match/MatchContext.java
+++ b/src/main/java/io/github/ktestify/match/MatchContext.java
@@ -17,6 +17,7 @@
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import lombok.Builder;
import lombok.Value;
@@ -72,6 +73,18 @@ public class MatchContext {
/** Expected value for {@link #matchKey}. */
String matchValue;
+ /**
+ * Expected transport-attribute key/value pairs (see
+ * {@link io.github.ktestify.models.ConsumedRecord#getAttributes()}).
+ *
+ *
Used by {@link io.github.ktestify.match.impl.AttributeRecordMatcher}. Example: {@code {"statusCode": "200"}}.
+ * Defaults to an empty map.
+ *
+ * @since 1.1.1
+ */
+ @Builder.Default
+ Map expectedAttributes = Collections.emptyMap();
+
/**
* Convenience accessor for single-record matchers. Returns the first element of {@link #matchFilePaths}, or
* {@code null} if the list is empty.
diff --git a/src/main/java/io/github/ktestify/match/RecordMatcherFactory.java b/src/main/java/io/github/ktestify/match/RecordMatcherFactory.java
index 92c6669..3826b08 100644
--- a/src/main/java/io/github/ktestify/match/RecordMatcherFactory.java
+++ b/src/main/java/io/github/ktestify/match/RecordMatcherFactory.java
@@ -16,17 +16,7 @@
package io.github.ktestify.match;
import io.github.ktestify.exceptions.ConsumerException;
-import io.github.ktestify.match.impl.AvroFieldsRecordMatcher;
-import io.github.ktestify.match.impl.AvroFileKeyRecordMatcher;
-import io.github.ktestify.match.impl.AvroFileRecordMatcher;
-import io.github.ktestify.match.impl.AvroKeyRecordMatcher;
-import io.github.ktestify.match.impl.FieldsRecordMatcher;
-import io.github.ktestify.match.impl.FileKeyRecordMatcher;
-import io.github.ktestify.match.impl.FileRecordMatcher;
-import io.github.ktestify.match.impl.KeyRecordMatcher;
-import io.github.ktestify.match.impl.NoOpRecordMatcher;
-import io.github.ktestify.match.impl.XPathRecordMatcher;
-import io.github.ktestify.match.impl.XmlRecordMatcher;
+import io.github.ktestify.match.impl.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.avro.generic.GenericRecord;
@@ -56,6 +46,14 @@ public final class RecordMatcherFactory {
public static final String METHOD_FIELDS_TO_MATCH = "methodFieldsToMatch";
public static final String METHOD_RECORD_KEY_MATCH = "methodRecordKeyMatch";
+ /**
+ * Asserts transport attributes (see {@link io.github.ktestify.models.ConsumedRecord#getAttributes()}). Only
+ * available for raw transports, since attributes are populated by synchronous request/response clients.
+ *
+ * @since 1.1.1
+ */
+ public static final String METHOD_MATCH_ATTRIBUTES = "methodMatchAttributes";
+
private RecordMatcherFactory() {}
// =========================================================================
@@ -83,10 +81,11 @@ public static RecordMatcher forRaw(String matchMethod) {
case METHOD_MATCH_XML -> new XmlRecordMatcher();
case METHOD_MATCH_XPATH -> new XPathRecordMatcher();
case METHOD_RECORD_KEY_MATCH -> new KeyRecordMatcher();
+ case METHOD_MATCH_ATTRIBUTES -> new AttributeRecordMatcher<>();
default ->
throw new ConsumerException("Unknown raw matchMethod '" + matchMethod + "'. "
+ "Valid values: methodMatchFile, methodMatchKeyValue, methodFieldsToMatch, "
- + "methodMatchXML, methodMatchXPath, methodRecordKeyMatch.");
+ + "methodMatchXML, methodMatchXPath, methodRecordKeyMatch, methodMatchAttributes.");
};
}
diff --git a/src/main/java/io/github/ktestify/match/impl/AttributeRecordMatcher.java b/src/main/java/io/github/ktestify/match/impl/AttributeRecordMatcher.java
new file mode 100644
index 0000000..5877944
--- /dev/null
+++ b/src/main/java/io/github/ktestify/match/impl/AttributeRecordMatcher.java
@@ -0,0 +1,79 @@
+/*
+ * Copyright 2026 Nil MALHOMME (malhomme.nil+oss@icloud.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.github.ktestify.match.impl;
+
+import io.github.ktestify.exceptions.ComparisonException;
+import io.github.ktestify.match.MatchContext;
+import io.github.ktestify.match.MatchResult;
+import io.github.ktestify.match.RecordMatcher;
+import io.github.ktestify.models.ConsumedRecord;
+import java.util.*;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Generic matcher that asserts one or more {@link ConsumedRecord#getAttributes()} entries against the expected values
+ * supplied via {@link MatchContext#getExpectedAttributes()}.
+ *
+ *
Transport-agnostic by design, this matcher is reused by any transport that populates {@code attributes} (HTTP
+ * status code today, gRPC status / MQ reason code / script exit code in the future). It operates on
+ * {@code List>} for any {@code V} since it never inspects {@link ConsumedRecord#getValue()}.
+ *
+ *
Matching rule: every key in {@code expectedAttributes} must be present in the actual record's {@code attributes}
+ * with an exactly-equal String value (case-sensitive). Only the first record in the list is used (single-record
+ * semantics, consistent with {@link FileRecordMatcher} and {@link KeyRecordMatcher}).
+ *
+ * @param the record value type, irrelevant to this matcher, kept for interface compatibility
+ * @since 1.1.1
+ */
+@Slf4j
+public class AttributeRecordMatcher implements RecordMatcher {
+
+ @Override
+ public MatchResult match(List> records, MatchContext context) throws ComparisonException {
+
+ Map expected = context.getExpectedAttributes();
+
+ if (expected == null || expected.isEmpty()) {
+ log.debug("No expected attributes configured, nothing to assert.");
+ return MatchResult.pass();
+ }
+
+ if (records == null || records.isEmpty()) {
+ throw new ComparisonException("AttributeRecordMatcher requires at least one record to compare.");
+ }
+
+ Map actual = records.get(0).getAttributes();
+ Map safeActual = actual != null ? actual : Collections.emptyMap();
+
+ List diffs = new ArrayList<>();
+ for (Map.Entry entry : expected.entrySet()) {
+ String actualValue = safeActual.get(entry.getKey());
+ if (!Objects.equals(entry.getValue(), actualValue)) {
+ diffs.add(
+ String.format("%s: expected '%s' but was '%s'", entry.getKey(), entry.getValue(), actualValue));
+ }
+ }
+
+ if (!diffs.isEmpty()) {
+ String diff = String.join(System.lineSeparator(), diffs);
+ log.error("Record attribute mismatch:\n{}", diff);
+ return MatchResult.fail(diff, expected.toString(), safeActual.toString());
+ }
+
+ log.info("All {} expected record attribute(s) matched.", expected.size());
+ return MatchResult.pass(expected.toString(), safeActual.toString());
+ }
+}
diff --git a/src/main/java/io/github/ktestify/models/ConsumedRecord.java b/src/main/java/io/github/ktestify/models/ConsumedRecord.java
index 28edbba..fd6dcd0 100644
--- a/src/main/java/io/github/ktestify/models/ConsumedRecord.java
+++ b/src/main/java/io/github/ktestify/models/ConsumedRecord.java
@@ -18,6 +18,7 @@
import java.time.Instant;
import java.util.Collections;
import java.util.Map;
+import lombok.Builder;
import lombok.Value;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
@@ -25,9 +26,12 @@
/**
* Immutable value object representing a single record that has been fetched from any IO source (Kafka, IBM MQ, etc.).
*
- *
This is the common currency that flows between the transport layer ({@code RecordFetcher}) and the
- * assertion layer ({@code RecordMatcher}). Matchers have zero dependency on Kafka or any other transport — they only
- * know about {@code ConsumedRecord}.
+ *
This is the common currency that flows between the transport layer ({@code RecordFetcher} and
+ * {@code RequestResponseClient}) and the assertion layer ({@code RecordMatcher}). Matchers have zero dependency on
+ * Kafka or any other transport — they only know about {@code ConsumedRecord}.
+ *
+ *
Synchronous transports (see {@code RequestResponseClient}) additionally populate {@link #attributes} with
+ * structured transport metadata such as an HTTP status code. Asynchronous transports leave it empty.
*
* @param the type of the record value
* @since 0.3.0
@@ -56,6 +60,78 @@ public class ConsumedRecord {
/** Transport-level headers / properties. Keys and values are Strings to stay transport-agnostic. */
Map headers;
+ /**
+ * Transport-specific structured metadata that does not belong under {@link #headers} (which models actual protocol
+ * headers).
+ *
+ *
Examples: HTTP status code and elapsed time, a future gRPC status code, an MQ reason code, a script exit code.
+ * Matchers that need this data use {@link io.github.ktestify.match.impl.AttributeRecordMatcher} or read it
+ * directly. Never {@code null}, defaults to an empty, immutable map.
+ *
+ * @since 1.1.1
+ */
+ Map attributes;
+
+ // -------------------------------------------------------------------------
+ // Constructors
+ // -------------------------------------------------------------------------
+
+ /**
+ * Full constructor including transport {@link #attributes}.
+ *
+ * @param source the source topic / queue / channel name
+ * @param partition the partition index, {@code 0} for non-partitioned sources
+ * @param offset the offset within the partition, {@code -1} when the source has no offset concept
+ * @param key the record key, may be {@code null}
+ * @param value the deserialized record value
+ * @param timestamp the timestamp at which the record was written to the source
+ * @param headers the transport-level headers, may be {@code null} (treated as empty)
+ * @param attributes the transport-specific metadata, may be {@code null} (treated as empty)
+ * @since 1.1.1
+ */
+ @Builder
+ public ConsumedRecord(
+ String source,
+ int partition,
+ long offset,
+ String key,
+ V value,
+ Instant timestamp,
+ Map headers,
+ Map attributes) {
+ this.source = source;
+ this.partition = partition;
+ this.offset = offset;
+ this.key = key;
+ this.value = value;
+ this.timestamp = timestamp;
+ this.headers = headers;
+ this.attributes = attributes != null ? attributes : Collections.emptyMap();
+ }
+
+ /**
+ * Backward-compatible constructor for transports that carry no {@link #attributes}. Delegates to the full
+ * constructor with an empty attribute map.
+ *
+ * @param source the source topic / queue / channel name
+ * @param partition the partition index, {@code 0} for non-partitioned sources
+ * @param offset the offset within the partition, {@code -1} when the source has no offset concept
+ * @param key the record key, may be {@code null}
+ * @param value the deserialized record value
+ * @param timestamp the timestamp at which the record was written to the source
+ * @param headers the transport-level headers, may be {@code null} (treated as empty)
+ */
+ public ConsumedRecord(
+ String source,
+ int partition,
+ long offset,
+ String key,
+ V value,
+ Instant timestamp,
+ Map headers) {
+ this(source, partition, offset, key, value, timestamp, headers, Collections.emptyMap());
+ }
+
// -------------------------------------------------------------------------
// Factory helpers
// -------------------------------------------------------------------------
@@ -63,6 +139,8 @@ public class ConsumedRecord {
/**
* Builds a {@code ConsumedRecord} from a Kafka {@link ConsumerRecord}.
*
+ *
Kafka has no use for {@link #attributes} today, so an empty map is supplied.
+ *
* @param the Kafka key type
* @param the Kafka value type
* @param record the Kafka consumer record
@@ -78,7 +156,8 @@ public static ConsumedRecord fromKafkaRecord(ConsumerRecord reco
key,
record.value(),
Instant.ofEpochMilli(record.timestamp()),
- headers);
+ headers,
+ Collections.emptyMap());
}
/**
diff --git a/src/test/java/io/github/ktestify/io/core/AbstractSynchronousConsumerTest.java b/src/test/java/io/github/ktestify/io/core/AbstractSynchronousConsumerTest.java
new file mode 100644
index 0000000..fe196b4
--- /dev/null
+++ b/src/test/java/io/github/ktestify/io/core/AbstractSynchronousConsumerTest.java
@@ -0,0 +1,167 @@
+/*
+ * Copyright 2026 Nil MALHOMME (malhomme.nil+oss@icloud.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.github.ktestify.io.core;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.*;
+
+import io.github.ktestify.exceptions.ConsumerException;
+import io.github.ktestify.exceptions.FetchException;
+import io.github.ktestify.match.MatchContext;
+import io.github.ktestify.match.MatchResult;
+import io.github.ktestify.match.RecordMatcher;
+import io.github.ktestify.models.ConsumedRecord;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+@DisplayName("AbstractSynchronousConsumer")
+class AbstractSynchronousConsumerTest {
+
+ private static final String REQUEST = "GET /orders";
+
+ private RequestResponseClient client;
+ private RecordMatcher matcher;
+ private MatchContext matchContext;
+ private TestConsumer consumer;
+
+ @SuppressWarnings("unchecked")
+ @BeforeEach
+ void setUp() {
+ client = mock(RequestResponseClient.class);
+ matcher = mock(RecordMatcher.class);
+ matchContext = MatchContext.builder()
+ .expectedAttributes(Map.of("statusCode", "200"))
+ .build();
+ consumer = new TestConsumer(client, matcher, matchContext);
+ }
+
+ // =========================================================================
+ // Happy path
+ // =========================================================================
+
+ @Nested
+ @DisplayName("call — success")
+ class Success {
+
+ @Test
+ @DisplayName("returns true when the matcher passes")
+ void returnsTrueWhenMatcherPasses() {
+ List> records = records();
+ when(client.execute(REQUEST)).thenReturn(records);
+ when(matcher.match(records, matchContext)).thenReturn(MatchResult.pass());
+
+ assertTrue(consumer.call());
+ }
+
+ @Test
+ @DisplayName("returns false when the matcher fails")
+ void returnsFalseWhenMatcherFails() {
+ List> records = records();
+ when(client.execute(REQUEST)).thenReturn(records);
+ when(matcher.match(records, matchContext)).thenReturn(MatchResult.fail("boom", "200", "500"));
+
+ assertFalse(consumer.call());
+ }
+
+ @Test
+ @DisplayName("sends the request built by buildRequest and matches with buildMatchContext")
+ void wiresRequestAndContext() {
+ when(client.execute(anyString())).thenReturn(records());
+ when(matcher.match(any(), any())).thenReturn(MatchResult.pass());
+
+ consumer.call();
+
+ verify(client).execute(eq(REQUEST));
+ verify(matcher).match(any(), eq(matchContext));
+ }
+
+ @Test
+ @DisplayName("does not close the client — the client outlives a single call")
+ void doesNotCloseClient() {
+ when(client.execute(anyString())).thenReturn(records());
+ when(matcher.match(any(), any())).thenReturn(MatchResult.pass());
+
+ consumer.call();
+
+ verify(client, never()).close();
+ }
+ }
+
+ // =========================================================================
+ // Failure path
+ // =========================================================================
+
+ @Nested
+ @DisplayName("call — failure")
+ class Failure {
+
+ @Test
+ @DisplayName("wraps a FetchException into a ConsumerException")
+ void wrapsFetchException() {
+ when(client.execute(anyString())).thenThrow(new FetchException("connection refused"));
+
+ ConsumerException exception = assertThrows(ConsumerException.class, () -> consumer.call());
+ assertEquals("connection refused", exception.getMessage());
+ verifyNoInteractions(matcher);
+ }
+ }
+
+ // =========================================================================
+ // Helpers
+ // =========================================================================
+
+ private static List> records() {
+ return List.of(new ConsumedRecord<>(
+ "http://localhost/orders",
+ 0,
+ -1L,
+ "GET",
+ "{}",
+ Instant.now(),
+ Collections.emptyMap(),
+ Map.of("statusCode", "200")));
+ }
+
+ /** Minimal concrete subclass exercising the abstract extension points. */
+ private static final class TestConsumer extends AbstractSynchronousConsumer {
+
+ private final MatchContext matchContext;
+
+ private TestConsumer(
+ RequestResponseClient client, RecordMatcher matcher, MatchContext context) {
+ super(Collections.emptyMap(), client, matcher);
+ this.matchContext = context;
+ }
+
+ @Override
+ protected String buildRequest() {
+ return REQUEST;
+ }
+
+ @Override
+ protected MatchContext buildMatchContext() {
+ return matchContext;
+ }
+ }
+}
diff --git a/src/test/java/io/github/ktestify/io/core/PollingRequestResponseClientTest.java b/src/test/java/io/github/ktestify/io/core/PollingRequestResponseClientTest.java
new file mode 100644
index 0000000..bc957f3
--- /dev/null
+++ b/src/test/java/io/github/ktestify/io/core/PollingRequestResponseClientTest.java
@@ -0,0 +1,157 @@
+/*
+ * Copyright 2026 Nil MALHOMME (malhomme.nil+oss@icloud.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.github.ktestify.io.core;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import io.github.ktestify.exceptions.FetchException;
+import io.github.ktestify.models.ConsumedRecord;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Predicate;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+@DisplayName("PollingRequestResponseClient")
+class PollingRequestResponseClientTest {
+
+ /** Short intervals keep the suite fast, no real waiting of any significance. */
+ private static final long TIMEOUT_MS = 300L;
+
+ private static final long POLL_INTERVAL_MS = 20L;
+
+ private static final Predicate>> STATUS_200 =
+ records -> "200".equals(records.get(0).getAttributes().get("statusCode"));
+
+ // =========================================================================
+ // Predicate satisfied
+ // =========================================================================
+
+ @Nested
+ @DisplayName("Predicate satisfied")
+ class PredicateSatisfied {
+
+ @Test
+ @DisplayName("returns immediately when the first attempt already passes")
+ void returnsOnFirstAttempt() {
+ StubRequestResponseClient delegate =
+ new StubRequestResponseClient().thenReturn(Map.of("statusCode", "200"));
+
+ try (PollingRequestResponseClient polling =
+ new PollingRequestResponseClient<>(delegate, STATUS_200, TIMEOUT_MS, POLL_INTERVAL_MS)) {
+
+ List> result = polling.execute("request");
+
+ assertEquals("200", result.get(0).getAttributes().get("statusCode"));
+ assertEquals(1, delegate.getExecuteCount());
+ }
+ }
+
+ @Test
+ @DisplayName("retries until the predicate passes, then stops")
+ void retriesUntilPredicatePasses() {
+ StubRequestResponseClient delegate = new StubRequestResponseClient()
+ .thenReturn(Map.of("statusCode", "404"))
+ .thenReturn(Map.of("statusCode", "404"))
+ .thenReturn(Map.of("statusCode", "200"))
+ .thenReturn(Map.of("statusCode", "500"));
+
+ try (PollingRequestResponseClient polling =
+ new PollingRequestResponseClient<>(delegate, STATUS_200, TIMEOUT_MS, POLL_INTERVAL_MS)) {
+
+ List> result = polling.execute("request");
+
+ assertEquals("200", result.get(0).getAttributes().get("statusCode"));
+ assertEquals(3, delegate.getExecuteCount());
+ }
+ }
+
+ @Test
+ @DisplayName("recovers from a transient transport failure")
+ void recoversFromTransientFailure() {
+ StubRequestResponseClient delegate = new StubRequestResponseClient()
+ .thenFail("connection refused")
+ .thenReturn(Map.of("statusCode", "200"));
+
+ try (PollingRequestResponseClient polling =
+ new PollingRequestResponseClient<>(delegate, STATUS_200, TIMEOUT_MS, POLL_INTERVAL_MS)) {
+
+ assertEquals(
+ "200", polling.execute("request").get(0).getAttributes().get("statusCode"));
+ }
+ }
+ }
+
+ // =========================================================================
+ // Timeout
+ // =========================================================================
+
+ @Nested
+ @DisplayName("Timeout")
+ class Timeout {
+
+ @Test
+ @DisplayName("returns the last result obtained so the matcher can report the real final state")
+ void returnsLastResultOnTimeout() {
+ StubRequestResponseClient delegate =
+ new StubRequestResponseClient().thenReturn(Map.of("statusCode", "503"));
+
+ try (PollingRequestResponseClient polling =
+ new PollingRequestResponseClient<>(delegate, STATUS_200, TIMEOUT_MS, POLL_INTERVAL_MS)) {
+
+ List> result = polling.execute("request");
+
+ assertEquals("503", result.get(0).getAttributes().get("statusCode"));
+ assertTrue(delegate.getExecuteCount() > 1, "expected several polling attempts");
+ }
+ }
+
+ @Test
+ @DisplayName("propagates the delegate failure when no result was ever obtained")
+ void propagatesFailureWhenNoResult() {
+ StubRequestResponseClient delegate = new StubRequestResponseClient().thenFail("connection refused");
+
+ try (PollingRequestResponseClient polling =
+ new PollingRequestResponseClient<>(delegate, STATUS_200, TIMEOUT_MS, POLL_INTERVAL_MS)) {
+
+ FetchException exception = assertThrows(FetchException.class, () -> polling.execute("request"));
+ assertEquals("connection refused", exception.getMessage());
+ }
+ }
+ }
+
+ // =========================================================================
+ // Lifecycle
+ // =========================================================================
+
+ @Nested
+ @DisplayName("Lifecycle")
+ class Lifecycle {
+
+ @Test
+ @DisplayName("close() delegates to the wrapped client")
+ void closeDelegates() {
+ StubRequestResponseClient delegate = new StubRequestResponseClient().thenReturn(Map.of());
+ PollingRequestResponseClient polling =
+ new PollingRequestResponseClient<>(delegate, STATUS_200, TIMEOUT_MS, POLL_INTERVAL_MS);
+
+ polling.close();
+
+ assertTrue(delegate.isClosed());
+ }
+ }
+}
diff --git a/src/test/java/io/github/ktestify/io/core/RequestResponseClientTest.java b/src/test/java/io/github/ktestify/io/core/RequestResponseClientTest.java
new file mode 100644
index 0000000..aae6f1d
--- /dev/null
+++ b/src/test/java/io/github/ktestify/io/core/RequestResponseClientTest.java
@@ -0,0 +1,71 @@
+/*
+ * Copyright 2026 Nil MALHOMME (malhomme.nil+oss@icloud.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.github.ktestify.io.core;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import io.github.ktestify.exceptions.FetchException;
+import io.github.ktestify.models.ConsumedRecord;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+@DisplayName("RequestResponseClient contract")
+class RequestResponseClientTest {
+
+ @Test
+ @DisplayName("execute returns a non-empty list of ConsumedRecord")
+ void executeReturnsRecords() {
+ try (StubRequestResponseClient client =
+ new StubRequestResponseClient().thenReturn(Map.of("statusCode", "200"))) {
+ List> records = client.execute("request");
+
+ assertNotNull(records);
+ assertEquals(1, records.size());
+ assertEquals("200", records.get(0).getAttributes().get("statusCode"));
+ }
+ }
+
+ @Test
+ @DisplayName("execute propagates transport failures as FetchException")
+ void executePropagatesFetchException() {
+ StubRequestResponseClient client = new StubRequestResponseClient().thenFail("connection refused");
+
+ FetchException exception = assertThrows(FetchException.class, () -> client.execute("request"));
+ assertEquals("connection refused", exception.getMessage());
+ }
+
+ @Test
+ @DisplayName("client is AutoCloseable and close() releases resources")
+ void closeReleasesResources() {
+ StubRequestResponseClient client = new StubRequestResponseClient().thenReturn(Map.of());
+
+ assertFalse(client.isClosed());
+ client.close();
+ assertTrue(client.isClosed());
+ }
+
+ @Test
+ @DisplayName("close() is idempotent")
+ void closeIsIdempotent() {
+ StubRequestResponseClient client = new StubRequestResponseClient().thenReturn(Map.of());
+
+ client.close();
+ assertDoesNotThrow(client::close);
+ assertTrue(client.isClosed());
+ }
+}
diff --git a/src/test/java/io/github/ktestify/io/core/StubRequestResponseClient.java b/src/test/java/io/github/ktestify/io/core/StubRequestResponseClient.java
new file mode 100644
index 0000000..ce0fbb2
--- /dev/null
+++ b/src/test/java/io/github/ktestify/io/core/StubRequestResponseClient.java
@@ -0,0 +1,78 @@
+/*
+ * Copyright 2026 Nil MALHOMME (malhomme.nil+oss@icloud.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.github.ktestify.io.core;
+
+import io.github.ktestify.exceptions.FetchException;
+import io.github.ktestify.models.ConsumedRecord;
+import java.time.Instant;
+import java.util.*;
+
+/**
+ * Minimal in-memory {@link RequestResponseClient} used by the {@code io.core} unit tests.
+ *
+ *
Returns a queue of pre-programmed responses, one per {@link #execute(String)} call, and repeats the last one once
+ * the queue is exhausted. Records how many times it was executed and whether it was closed.
+ */
+class StubRequestResponseClient implements RequestResponseClient {
+
+ private final Deque