feat: add attributes support to ConsumedRecord and related matchers

.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

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.
+ *
+ * <p>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.
+ *
+ * <p>This class contains <strong>no transport mechanics</strong> and <strong>no matching logic</strong>. Those
+ * responsibilities belong exclusively to {@link RequestResponseClient} and {@link RecordMatcher} respectively.
+ *
+ * <h2>Client lifecycle</h2>
+ *
+ * <p>Unlike {@code AbstractKafkaConsumer}, this class does <strong>not</strong> 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 <Req> the request type
+ * @param <V> the record value type
+ * @since 1.1.1
+ */
+@Slf4j
+public abstract class AbstractSynchronousConsumer<Req, V> extends AbstractConsumer {
+
+    protected final RequestResponseClient<Req, V> client;
+    protected final RecordMatcher<V> 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<String, String> properties, RequestResponseClient<Req, V> client, RecordMatcher<V> 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.
+     *
+     * <p>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.
+     *
+     * <p>Lifecycle:
+     *
+     * <ol>
+     *   <li>Build the request via {@link #buildRequest()}.
+     *   <li>Call {@link RequestResponseClient#execute(Object)} — blocks until the response arrives or fails.
+     *   <li>Pass the resulting records to {@link RecordMatcher#match(List, MatchContext)}.
+     * </ol>
+     *
+     * @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<ConsumedRecord<V>> 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());
+        }
+    }
+}

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.
+ *
+ * <p>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.
+ *
+ * <p>On timeout the <em>last</em> 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 <Req> the request type
+ * @param <V> the record value type
+ * @since 1.1.1
+ */
+@Slf4j
+public class PollingRequestResponseClient<Req, V> implements RequestResponseClient<Req, V> {
+
+    private final RequestResponseClient<Req, V> delegate;
+    private final Predicate<List<ConsumedRecord<V>>> 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<Req, V> delegate,
+            Predicate<List<ConsumedRecord<V>>> 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<ConsumedRecord<V>> execute(Req request) throws FetchException {
+        long deadlineMs = System.currentTimeMillis() + timeoutMs;
+        List<ConsumedRecord<V>> lastResult = null;
+        FetchException lastFailure = null;
+
+        do {
+            try {
+                List<ConsumedRecord<V>> 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.");
+        }
+    }
+}

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, …).
+ *
+ * <p>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".
+ *
+ * <p>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.
+ *
+ * <p>Implementations exist per transport:
+ *
+ * <ul>
+ *   <li>{@code HttpRequestResponseClient} — HTTP / HTTPS (ktestify-plugin-http)
+ *   <li>{@code GrpcRequestResponseClient} — gRPC (future)
+ * </ul>
+ *
+ * @param <Req> the request type specific to the transport (e.g. an HTTP request spec)
+ * @param <V> the type of the resulting record value (e.g. {@code String} for an HTTP body)
+ * @since 1.1.1
+ */
+public interface RequestResponseClient<Req, V> extends AutoCloseable {
+
+    /**
+     * Sends {@code request} and returns the result wrapped as a (typically single-element) list of
+     * {@link ConsumedRecord}.
+     *
+     * <p>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<ConsumedRecord<V>> 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();
+}

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()}).
+     *
+     * <p>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<String, String> expectedAttributes = Collections.emptyMap();
+
     /**
      * Convenience accessor for single-record matchers. Returns the first element of {@link #matchFilePaths}, or
      * {@code null} if the list is empty.

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<String> 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.");
         };
     }