refactor: improve runtime log visibility across transports and matchers

src/main/java/io/github/ktestify/io/core/AbstractSynchronousConsumer.java

@@ -104,15 +104,31 @@ protected AbstractSynchronousConsumer(
     public Boolean call() throws ConsumerException {
         try {
             Req request = buildRequest();
+            log.info(
+                    "Executing synchronous request via '{}'.", client.getClass().getSimpleName());
             List<ConsumedRecord<V>> records = client.execute(request);
             MatchContext matchContext = buildMatchContext();
+            log.info(
+                    "Running matcher '{}' on {} response record(s).",
+                    matcher.getClass().getSimpleName(),
+                    records.size());
             MatchResult result = matcher.match(records, matchContext);
 
-            log.debug("Match result: passed={}, diff={}", result.isPassed(), result.getDiff());
+            if (result.isPassed()) {
+                log.info(
+                        "Synchronous validation PASSED with matcher '{}'.",
+                        matcher.getClass().getSimpleName());
+            } else {
+                log.error(
+                        "Synchronous validation FAILED with matcher '{}': {}",
+                        matcher.getClass().getSimpleName(),
+                        result.getDiff());
+            }
 
             return result.isPassed();
 
         } catch (FetchException e) {
+            log.error("Synchronous request execution failed: {}", e.getMessage());
             throw new ConsumerException(e.getMessage(), e);
         }
     }

src/main/java/io/github/ktestify/io/core/PollingRequestResponseClient.java

@@ -75,18 +75,23 @@ public List<ConsumedRecord<V>> execute(Req request) throws FetchException {
         long deadlineMs = System.currentTimeMillis() + timeoutMs;
         List<ConsumedRecord<V>> lastResult = null;
         FetchException lastFailure = null;
+        int attempt = 0;
+
+        log.info("Starting polling request (timeout={}ms, pollInterval={}ms).", timeoutMs, pollIntervalMs);
 
         do {
+            attempt++;
             try {
                 List<ConsumedRecord<V>> result = delegate.execute(request);
                 lastResult = result;
                 lastFailure = null;
                 if (untilPredicate.test(result)) {
-                    log.debug("Polling predicate satisfied.");
+                    log.info("Polling predicate satisfied on attempt #{}.", attempt);
                     return result;
                 }
+                log.debug("Polling attempt #{} did not satisfy predicate; retrying...", attempt);
             } catch (FetchException e) {
-                log.debug("Polling attempt failed: {}", e.getMessage());
+                log.debug("Polling attempt #{} failed: {}", attempt, e.getMessage());
                 lastFailure = e;
             }
 
@@ -97,12 +102,19 @@ public List<ConsumedRecord<V>> execute(Req request) throws FetchException {
         } while (true);
 
         if (lastResult == null) {
+            log.error(
+                    "Polling failed after {} attempt(s) within {}ms without obtaining any response.",
+                    attempt,
+                    timeoutMs);
             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);
+        log.warn(
+                "Polling predicate never satisfied after {} attempt(s) within {}ms, returning the last result obtained.",
+                attempt,
+                timeoutMs);
         return lastResult;
     }
 

src/main/java/io/github/ktestify/io/inputs/DynamicVariableProcessor.java

@@ -51,9 +51,13 @@ public String process(String input) {
             String format = matcher.group(GROUP_FORMAT) != null ? matcher.group(GROUP_FORMAT) : "";
 
             if (DynamicVariableFactory.isRegistered(variableName)) {
-                LOGGER.info("Processing dynamic variable: {} with format: {}", variableName, format);
                 DynamicVariable variable = DynamicVariableFactory.getVariable(variableName);
                 String replacement = variable.process(format);
+                LOGGER.info(
+                        "Resolved dynamic variable '{{{}{}}}': '{}'",
+                        variableName,
+                        format.isEmpty() ? "" : ":" + format,
+                        replacement);
                 matcher.appendReplacement(sb, Matcher.quoteReplacement(replacement));
             } else {
                 LOGGER.warn("Dynamic variable {} is not registered, keeping it unchanged.", variableName);

src/main/java/io/github/ktestify/io/kafka/AbstractKafkaConsumer.java

@@ -83,18 +83,26 @@ public Boolean call() throws ConsumerException {
         try (KafkaRecordFetcher<K, V> fetcher = new KafkaRecordFetcher<>(context)) {
             List<ConsumedRecord<V>> records = fetcher.fetch();
             MatchContext matchContext = buildMatchContext();
+            log.info(
+                    "Running matcher '{}' on {} fetched record(s) from topic '{}'.",
+                    matcher.getClass().getSimpleName(),
+                    records.size(),
+                    context.getTopic().getNamespacedTopic());
             MatchResult result = matcher.match(records, matchContext);
 
-            log.debug(
-                    "Match result for topic '{}': passed={}, diff={}",
-                    context.getTopic().getNamespacedTopic(),
-                    result.isPassed(),
-                    result.getDiff());
-
             if (!result.isPassed()) {
+                log.error(
+                        "Validation FAILED on topic '{}' with matcher '{}': {}",
+                        context.getTopic().getNamespacedTopic(),
+                        matcher.getClass().getSimpleName(),
+                        result.getDiff());
                 throw new AssertionError("Consumer validation failed for topic '"
                         + context.getTopic().getNamespacedTopic() + "': " + result.getDiff());
             }
+            log.info(
+                    "Validation PASSED on topic '{}' with matcher '{}'.",
+                    context.getTopic().getNamespacedTopic(),
+                    matcher.getClass().getSimpleName());
             return true;
 
         } catch (FetchException e) {

src/main/java/io/github/ktestify/io/kafka/AbstractKafkaProducer.java

@@ -99,6 +99,11 @@ protected Topic getTopic() {
     protected ProducerRecord<K, V> buildRecord(K key, V value) {
         var record = new ProducerRecord<>(getTopic().getNamespacedTopic(), key, value);
         if (!headers.isEmpty()) {
+            logger.debug(
+                    "Attaching {} header(s) to record for topic '{}': {}",
+                    headers.size(),
+                    getTopic().getNamespacedTopic(),
+                    headers.keySet());
             for (Header header : headers.entrySet().stream()
                     .map(e -> (Header) new RecordHeader(e.getKey(), e.getValue().getBytes(StandardCharsets.UTF_8)))
                     .toList()) {
@@ -110,9 +115,17 @@ var record = new ProducerRecord<>(getTopic().getNamespacedTopic(), key, value);
 
     protected String resolvePayload() {
         if (inlinePayload != null && !inlinePayload.isEmpty()) {
+            logger.debug(
+                    "Resolved inline payload ({} chars) for topic '{}'.",
+                    inlinePayload.length(),
+                    getTopic().getNamespacedTopic());
             return inlinePayload;
         }
         if (payloadFile != null) {
+            logger.debug(
+                    "Reading payload from file '{}' for topic '{}'.",
+                    payloadFile.getPath(),
+                    getTopic().getNamespacedTopic());
             return FileUtils.getFileContent(payloadFile);
         }
         throw new ProducerException("No payload content was provided");
@@ -121,15 +134,24 @@ protected String resolvePayload() {
     protected String resolveSchema() throws IOException, RestClientException {
         SchemaRegistryClient client = SchemaRegistryClientFactory.createClient();
         if (!StringUtils.isBlank(context.getSchemaName())) {
-            return client.getLatestSchemaMetadata(context.getSchemaName() + "-value")
-                    .getSchema();
+            String subject = context.getSchemaName() + "-value";
+            logger.info("Resolving latest Avro schema for subject '{}' from Schema Registry.", subject);
+            return client.getLatestSchemaMetadata(subject).getSchema();
         }
         String subject = context.getTopic().getNamespacedTopic() + "-value";
         if (context.getSchemaVersion() == null || context.getSchemaVersion().isEmpty()) {
+            logger.info("Resolving latest Avro schema for subject '{}' from Schema Registry.", subject);
             return client.getLatestSchemaMetadata(subject).getSchema();
         } else {
-            return client.getSchemaMetadata(subject, Integer.parseInt(context.getSchemaVersion()))
-                    .getSchema();
+            try {
+                int version = Integer.parseInt(context.getSchemaVersion());
+                logger.info("Resolving Avro schema for subject '{}' (version {}) from Schema Registry.", subject, version);
+                return client.getSchemaMetadata(subject, version).getSchema();
+            } catch (NumberFormatException e) {
+                throw new ProducerException(
+                        "Invalid schema version '" + context.getSchemaVersion() + "' for subject '" + subject + "'. Expected an integer.",
+                        e);
+            }
         }
     }
 }

src/main/java/io/github/ktestify/io/kafka/KafkaRecordFetcher.java

@@ -25,6 +25,7 @@
 import io.github.ktestify.models.ConsumedRecord;
 import io.github.ktestify.models.MatchedRecord;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
@@ -158,8 +159,9 @@ public void close() {
      * an explicit reset, for example from the {@code clear known messages} step.
      */
     public static void clearMatchedRecords() {
+        int count = MATCHED_RECORDS.size();
         MATCHED_RECORDS.clear();
-        log.debug("KafkaRecordFetcher: matched-records registry cleared.");
+        log.info("KafkaRecordFetcher: cleared {} record(s) from the matched-records deduplication registry.", count);
     }
 
     // =========================================================================
@@ -180,13 +182,17 @@ private void subscribeAndAwaitAssignment(long deadlineNanos) throws FetchExcepti
         long pollIntervalMs = frameworkConfig.getPollIntervalMillis();
         while (kafkaConsumer.assignment().isEmpty()) {
             if (System.nanoTime() >= deadlineNanos) {
+                log.error(
+                        "Timed out after {}ms waiting for partition assignment on topic '{}'.",
+                        readTimeoutMs,
+                        namespacedTopic);
                 throw new FetchTimeoutException("Timed out after " + readTimeoutMs
                         + "ms waiting for partition assignment on topic '" + namespacedTopic
                         + "'. Check that the broker is reachable and the topic exists.");
             }
             kafkaConsumer.poll(Duration.ofMillis(pollIntervalMs));
         }
-        log.debug("Partition assignment received for topic '{}'.", namespacedTopic);
+        log.info("Assigned partition(s) {} for topic '{}'.", kafkaConsumer.assignment(), namespacedTopic);
     }
 
     /**
@@ -244,7 +250,11 @@ private long calculateDeltaTime() {
      */
     private void seekToOffset(long deltaTimestamp, long deadlineNanos) {
         Set<TopicPartition> assignments = kafkaConsumer.assignment();
-        log.debug(MESSAGE_CONSUMER_SEEKING_TO_OFFSET, deltaTimestamp);
+        log.info(
+                "Seeking partitions {} to timestamp {} ({})",
+                assignments,
+                deltaTimestamp,
+                Instant.ofEpochMilli(deltaTimestamp));
 
         Map<TopicPartition, Long> timestampMap =
                 assignments.stream().collect(Collectors.toMap(tp -> tp, tp -> deltaTimestamp));
@@ -257,9 +267,14 @@ private void seekToOffset(long deltaTimestamp, long deadlineNanos) {
             log.trace(MESSAGE_CONSUMER_OFFSET_FOR_TIMESTAMP, offsetAndTimestamp);
             if (offsetAndTimestamp != null) {
                 kafkaConsumer.seek(topicPartition, offsetAndTimestamp.offset());
+                log.info(
+                        "Partition {} seeked to offset {} (timestamp={}).",
+                        topicPartition,
+                        offsetAndTimestamp.offset(),
+                        Instant.ofEpochMilli(offsetAndTimestamp.timestamp()));
             } else {
                 kafkaConsumer.seekToEnd(Collections.singletonList(topicPartition));
-                log.debug(MESSAGE_CONSUMER_OFFSETS_NULL_SEEK_TO_LATEST, topicPartition.partition());
+                log.info(MESSAGE_CONSUMER_OFFSETS_NULL_SEEK_TO_LATEST, topicPartition.partition());
             }
         });
     }
@@ -292,11 +307,15 @@ private List<ConsumedRecord<V>> pollUntilRecordFound(long deadlineNanos) throws
         }
         List<ConsumedRecord<V>> accumulated = new ArrayList<>();
 
-        log.debug(
-                "Polling topic '{}' in {} mode, target={} record(s).",
+        log.info(
+                "Polling topic '{}' in {} mode (target={} record(s), timeout={}ms, expectedKey={}).",
                 namespacedTopic,
                 batchMode ? "BATCH" : "SINGLE",
-                targetSize);
+                targetSize,
+                readTimeoutMs,
+                context.getExpectedRecordKey() != null
+                        ? "'" + context.getExpectedRecordKey() + "' (" + context.getKeyMatchStrategy() + ")"
+                        : "<any>");
 
         while (System.nanoTime() < deadlineNanos) {
             ConsumerRecords<K, V> batch = kafkaConsumer.poll(Duration.ofMillis(pollIntervalMs));
@@ -310,11 +329,18 @@ private List<ConsumedRecord<V>> pollUntilRecordFound(long deadlineNanos) throws
                     }
                     ConsumedRecord<V> consumed = ConsumedRecord.fromKafkaRecord(record);
                     if (!tryClaim(consumed)) {
-                        log.debug(MESSAGE_CONSUMER_RECORD_ALREADY_MATCHED, consumed.toMatchedRecord());
+                        log.info(MESSAGE_CONSUMER_RECORD_ALREADY_MATCHED, consumed.toMatchedRecord());
                         continue;
                     }
 
-                    log.info(MESSAGE_CONSUMER_GOT_RECORD_WITH_KEY_AND_VALUE, record.key(), record.value());
+                    log.info(
+                            "Claimed record from topic '{}' [partition={}, offset={}, key={}, timestamp={}].",
+                            namespacedTopic,
+                            record.partition(),
+                            record.offset(),
+                            record.key(),
+                            Instant.ofEpochMilli(record.timestamp()));
+                    log.debug(MESSAGE_CONSUMER_GOT_RECORD_WITH_KEY_AND_VALUE, record.key(), record.value());
                     accumulated.add(consumed);
 
                     // In SINGLE mode return immediately on the first passing record.
@@ -338,11 +364,18 @@ private List<ConsumedRecord<V>> pollUntilRecordFound(long deadlineNanos) throws
         if (!accumulated.isEmpty()) {
             // Partial batch: give the claimed records back before reporting the timeout.
             accumulated.forEach(this::release);
+            log.warn(
+                    "Timed out after {}ms waiting for {} record(s) on topic '{}': only {} collected (released back to registry).",
+                    readTimeoutMs,
+                    targetSize,
+                    namespacedTopic,
+                    accumulated.size());
             throw new FetchTimeoutException("Timed out after " + readTimeoutMs + "ms waiting for " + targetSize
                     + " record(s) on topic '" + namespacedTopic
                     + "': only " + accumulated.size() + " collected.");
         }
 
+        log.debug("Timed out after {}ms waiting for a record on topic '{}'.", readTimeoutMs, namespacedTopic);
         throw new FetchTimeoutException(
                 "Timed out after " + readTimeoutMs + "ms waiting for a record on topic '" + namespacedTopic + "'.");
     }

src/main/java/io/github/ktestify/io/kafka/impl/AvroKafkaProducer.java

@@ -73,6 +73,12 @@ protected void produce() {
             JsonObject payload =
                     AvroUtils.getJsonElementFromFile(resolvePayload()).getAsJsonObject();
             GenericRecord recordValue = (GenericRecord) AvroUtils.convertJsonToAvro(payload, schema);
+            logger.debug(
+                    "Producing Avro record to topic '{}' with key '{}' and schema '{}':\n{}",
+                    getTopic().getNamespacedTopic(),
+                    context.getRecordKey(),
+                    schema.getFullName(),
+                    recordValue);
             ProducerRecord<String, GenericRecord> record = buildRecord(context.getRecordKey(), recordValue);
             RecordMetadata metadata = producer.send(record).get();
             logger.info(
@@ -84,11 +90,23 @@ protected void produce() {
                     metadata.timestamp());
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
+            logger.error(
+                    "Interrupted while producing Avro message to topic '{}'",
+                    getTopic().getNamespacedTopic(),
+                    e);
             throw new ProducerException("Interrupted while producing Avro message", e);
 
         } catch (ExecutionException e) {
+            logger.error(
+                    "Failed to produce Avro message to topic '{}': {}",
+                    getTopic().getNamespacedTopic(),
+                    e.getCause() != null ? e.getCause().getMessage() : e.getMessage());
             throw new ProducerException("Failed to produce Avro message", e.getCause());
         } catch (IOException | RestClientException e) {
+            logger.error(
+                    "Failed to read schema or payload for Avro message on topic '{}': {}",
+                    getTopic().getNamespacedTopic(),
+                    e.getMessage());
             throw new ProducerException("Failed to read schema or payload for Avro message", e);
         }
     }

src/main/java/io/github/ktestify/io/kafka/impl/RawKafkaProducer.java

@@ -58,6 +58,11 @@ protected void produce() {
         try {
             String payload = resolvePayload();
             String key = context.getRecordKey();
+            logger.debug(
+                    "Producing raw record to topic '{}' with key '{}':\n{}",
+                    getTopic().getNamespacedTopic(),
+                    key,
+                    payload);
             ProducerRecord<String, String> record = buildRecord(key, payload);
             RecordMetadata metadata = producer.send(record).get();
             logger.info(
@@ -69,8 +74,16 @@ protected void produce() {
                     metadata.timestamp());
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
+            logger.error(
+                    "Interrupted while producing raw message to topic '{}'",
+                    getTopic().getNamespacedTopic(),
+                    e);
             throw new ProducerException("Interrupted while producing raw message", e);
         } catch (ExecutionException e) {
+            logger.error(
+                    "Failed to produce raw message to topic '{}': {}",
+                    getTopic().getNamespacedTopic(),
+                    e.getCause() != null ? e.getCause().getMessage() : e.getMessage());
             throw new ProducerException("Failed to produce raw message", e.getCause());
         }
     }