[Feature]: Add a synchronous request/response transport contract

[nil-malh]

Pre-flight checklist

• I searched existing issues and this feature has not been requested yet.

Problem Statement

ktestify-core currently only exposes one transport shape:

public interface RecordFetcher<V> extends AutoCloseable {
    List<ConsumedRecord<V>> fetch() throws FetchException;
    void close();
}

This models a background stream that you poll until something appears (Kafka topic, Azure Blob polling, IBM MQ queue watch). It works well for those transports, but it is the wrong shape for synchronous, caller-initiated transports like HTTP: an HTTP call does not "wait for a record to appear", it sends a request right now and gets an answer immediately.

This gap matters because a plugin is being planned (ktestify-plugin-http) to let users test content chains where part of the pipeline is triggered by an API call instead of Kafka. Building that plugin on top of RecordFetcher would be dishonest (forcing a poll-based contract onto a call/response transport), and building it with its own ad hoc action/validation services would mean the plugin stops following the core API, defeating the purpose of the shared three-layer architecture (transport / orchestration / assertion) that every other transport in this project follows.

Rather than fix this inside the plugin, the fix belongs in ktestify-core: add a second, sibling transport contract for synchronous request/response transports, so HTTP (and any future gRPC, SOAP, or synchronous MQ request/reply transport) can plug into the exact same assertion layer (RecordMatcher, MatchContext, MatchResult, RecordMatcherFactory) that Kafka and Azure Blob already use, with zero new matcher code needed for body comparisons.

Proposed Solution

Add the following, fully additive, backward-compatible pieces to ktestify-core:

1. io/core/RequestResponseClient.java (new)
   Sibling contract to RecordFetcher<V>:

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

   Same return type as RecordFetcher.fetch(), so it feeds the exact same RecordMatcher<V> signature unchanged.

2. models/ConsumedRecord.java (edit)
   Add a new attributes: Map<String,String> field (default empty map) for transport-specific structured metadata that does not belong under headers (real protocol headers). Examples: HTTP status code, elapsed time, a future gRPC status code, an MQ reason code. Must stay backward compatible with existing callers (fromKafkaRecord, AzureBlobRecordFetcher, tests), either via a @Builder plus a legacy constructor overload, or an additive constructor overload defaulting attributes to Collections.emptyMap().

3. match/MatchContext.java (edit)
   Add expectedAttributes: Map<String,String> (default empty map), used by the new matcher below.

4. match/impl/AttributeRecordMatcher.java (new)
   Generic, transport-agnostic matcher that asserts record.getAttributes() entries against context.getExpectedAttributes(). Works for any V since it never inspects the record value. Reused later by any future transport that populates attributes (status codes, reason codes, exit codes, etc.), so no per-transport matcher class is needed going forward.

5. match/RecordMatcherFactory.java (edit)
   Add METHOD_MATCH_ATTRIBUTES constant and register AttributeRecordMatcher in forRaw(...). Update the ConsumerException message listing valid matchMethod values.

6. io/core/AbstractSynchronousConsumer.java (new)
   Orchestration base mirroring io.github.ktestify.io.kafka.AbstractKafkaConsumer 1:1 in shape and Javadoc style, but wired to RequestResponseClient instead of a per-call KafkaRecordFetcher:

   public abstract class AbstractSynchronousConsumer<Req, V> extends AbstractConsumer {
       protected abstract Req buildRequest();
       protected abstract MatchContext buildMatchContext();
   
       public Boolean call() throws ConsumerException {
           try {
               List<ConsumedRecord<V>> records = client.execute(buildRequest());
               return matcher.match(records, buildMatchContext()).isPassed();
           } catch (FetchException e) {
               throw new ConsumerException(e.getMessage());
           }
       }
   }

   Note: unlike AbstractKafkaConsumer, the RequestResponseClient is expected to be a longer-lived, connection-pooled client owned by the caller, not created and closed per invocation.

7. io/core/PollingRequestResponseClient.java (new, recommended in the same PR)
   Generic RequestResponseClient decorator that retries execute(...) against a delegate client until a caller-supplied predicate on the result passes, or a timeout elapses. Gives every synchronous-transport plugin "eventually consistent" polling semantics for free, instead of every plugin hand-rolling its own sleep loop (as currently happens independently in AzureBlobRecordFetcher and the plugin skeleton).

Alternatives Considered

No response

Affected Component

Other

Example Usage / API Sketch

Additional Context

Discussed in https://github.com/orgs/ktestify/discussions/54

^{Originally posted by williamgarr August 6, 2026}
Hello, in my app I have a content chain of consumers/producers. But a part of the chain is triggered with API calls instead of Kafka.

Ideally I want to test it with the rest of the content chain in my KTestify cucumber setup.

The first workaround tooling I thought of was :

• A generic script to launch API calls using curl
• The script is run by the KTestify cucumber run script step

That could work but it is IMO a little dirty.

An elegant solution could be :

• A dedicated HTTP client KTestify plugin
• Which adds specific steps to launch HTTP requests during the tests
• It could improve readability in the tests and also technical integration complexity due to custom scripts
• Also we can imagine more complex use cases in the future involving HTTP requests

Hope you like the idea ;)