[Part 09] AsyncAPI 3.x: track what the service consumes and emits over Kafka

client-java/controller-api/src/main/java/org/evomaster/client/java/controller/api/dto/AdditionalInfoDto.java

@@ -67,4 +67,10 @@ public class AdditionalInfoDto {
     public List<ExternalServiceInfoDto> employedDefaultWM = new ArrayList<>();
 
     public List<HostnameResolutionInfoDto> hostnameResolutionInfoDtos = new ArrayList<>();
+
+    /**
+     * The messages the SUT handed to a broker client to publish (eg a Kafka producer), in order.
+     * Only recorded for clients the instrumentation tracks.
+     */
+    public List<EmittedMessageDto> emittedMessages = new ArrayList<>();
 }

client-java/controller-api/src/main/java/org/evomaster/client/java/controller/api/dto/EmittedMessageDto.java

@@ -0,0 +1,29 @@
+package org.evomaster.client.java.controller.api.dto;
+
+/**
+ * A message the SUT handed to a broker client to publish during an action, eg a record given to
+ * a Kafka producer. Only the address and the payload, the latter only when it is text, truncated.
+ */
+public class EmittedMessageDto {
+
+    /**
+     * Where the message was published to, eg a Kafka topic. Null if the client did not say.
+     */
+    public String address;
+
+    /**
+     * The payload as text: a String, or bytes read as UTF-8. Null for no payload or any other
+     * object, whose wire form depends on the client's serializer; see {@link #payloadType}.
+     */
+    public String payload;
+
+    /**
+     * Whether {@link #payload} was cut because it was too long. Null if not known.
+     */
+    public Boolean payloadTruncated;
+
+    /**
+     * The name of the class of the payload the SUT gave. Null if it gave none.
+     */
+    public String payloadType;
+}

client-java/controller-api/src/main/java/org/evomaster/client/java/controller/api/dto/problem/asyncapi/AsyncApiActionDto.java

@@ -38,7 +38,16 @@ public enum MessageCompletionStrategy {
          * objective for the first time, or improving one. A heuristic: a handler that only runs
          * code already covered earlier in the same test leaves coverage stable too.
          */
-        COVERAGE_STABLE
+        COVERAGE_STABLE,
+
+        /**
+         * Stop early once the SUT has received at least one message from {@link #address}, and
+         * every thread that received one has gone back to its client for more, which is what a
+         * consumer loop does when its handler has returned. Read off what the instrumentation
+         * records of the SUT's broker client calls, so only for clients it tracks. A heuristic:
+         * work handed off to another thread looks done as soon as it is handed off.
+         */
+        HANDLER_RETURN
     }
 
     /**

client-java/controller-api/src/main/java/org/evomaster/client/java/controller/api/dto/problem/asyncapi/AsyncApiReplyDto.java

@@ -22,7 +22,8 @@ public enum SettleEndReason {
         /**
          * The whole {@link AsyncApiActionDto#settleMs} window went by. Also what a
          * {@link AsyncApiActionDto.MessageCompletionStrategy#COVERAGE_STABLE} wait ends with when the SUT's
-         * coverage cannot be read.
+         * coverage cannot be read, and a {@link AsyncApiActionDto.MessageCompletionStrategy#HANDLER_RETURN} one
+         * when what the SUT received cannot be.
          */
         DEADLINE,
 
@@ -35,7 +36,13 @@ public enum SettleEndReason {
         /**
          * The waiting thread was interrupted.
          */
-        INTERRUPTED
+        INTERRUPTED,
+
+        /**
+         * Every message the SUT received from {@link AsyncApiActionDto#address} in this action
+         * was handled, see {@link AsyncApiActionDto.MessageCompletionStrategy#HANDLER_RETURN}.
+         */
+        HANDLER_RETURN
     }
 
     /**

client-java/controller/src/main/java/org/evomaster/client/java/controller/EmbeddedSutController.java

@@ -156,6 +156,15 @@ public final Long getObjectiveUpdateCount() {
         return InstrumentationController.getObjectiveUpdateCount();
     }
 
+    @Override
+    public final MessageHandlingState getMessageHandlingState(String address) {
+        //nothing to read without the agent, as for the objective update count
+        if (!isInstrumentationActivated()) {
+            return null;
+        }
+        return InstrumentationController.getMessageHandlingState(address);
+    }
+
     @Override
     public final String getExecutableFullPath(){
         return null; //not needed for embedded

client-java/controller/src/main/java/org/evomaster/client/java/controller/ExternalSutController.java

@@ -463,6 +463,14 @@ public final Long getObjectiveUpdateCount(long timeoutMs) {
         return serverController.getObjectiveUpdateCount(timeoutMs);
     }
 
+    @Override
+    public final MessageHandlingState getMessageHandlingState(String address) {
+        if (!isInstrumentationActivated()) {
+            return null;
+        }
+        return serverController.getMessageHandlingState(address);
+    }
+
     @Override
     public final void newActionSpecificHandler(ActionDto dto) {
         if (isInstrumentationActivated()) {

client-java/controller/src/main/java/org/evomaster/client/java/controller/internal/AsyncApiSettle.java

@@ -2,8 +2,10 @@
 
 import org.evomaster.client.java.controller.api.dto.problem.asyncapi.AsyncApiActionDto;
 import org.evomaster.client.java.controller.api.dto.problem.asyncapi.AsyncApiReplyDto;
+import org.evomaster.client.java.instrumentation.MessageHandlingState;
 
 import java.util.Objects;
+import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.function.LongFunction;
 
@@ -20,10 +22,13 @@ public class AsyncApiSettle {
     private static final String OBJECTIVE_UPDATES_FAILURE_WARNING = "Failed to read how many coverage objectives" +
             " the SUT has reached or improved, so the whole settle window is waited";
 
+    private static final String MESSAGE_HANDLING_FAILURE_WARNING = "Failed to read what the SUT received from the" +
+            " address a message was published to, so the whole settle window is waited";
+
     /**
-     * How often the SUT's coverage is read while waiting for it to become stable, in milliseconds.
-     * Short next to any useful stable window, and long enough that reading it, a socket round
-     * trip for an external SUT, costs little.
+     * How often the SUT's coverage, or what it received, is read while waiting for it to be done,
+     * in milliseconds. Short next to any useful stable window, and long enough that reading it, a
+     * socket round trip for an external SUT, costs little.
      */
     static final long POLL_INTERVAL_MS = 25;
 
@@ -75,9 +80,27 @@ public AsyncApiSettle() {
      *                          Only read for {@link AsyncApiActionDto.MessageCompletionStrategy#COVERAGE_STABLE}
      */
     public void settle(AsyncApiActionDto dto, AsyncApiReplyDto reply, LongFunction<Long> objectiveUpdates) {
+        settle(dto, reply, objectiveUpdates, address -> null);
+    }
+
+    /**
+     * As {@link #settle(AsyncApiActionDto, AsyncApiReplyDto, LongFunction)}, for every completion.
+     *
+     * @param messageHandling   reads what the SUT received from an address in the current action,
+     *                          and how much of it it handled, or null when that cannot be known.
+     *                          Only read, with {@link AsyncApiActionDto#address}, for
+     *                          {@link AsyncApiActionDto.MessageCompletionStrategy#HANDLER_RETURN}
+     */
+    public void settle(
+            AsyncApiActionDto dto,
+            AsyncApiReplyDto reply,
+            LongFunction<Long> objectiveUpdates,
+            Function<String, MessageHandlingState> messageHandling
+    ) {
         Objects.requireNonNull(dto);
         Objects.requireNonNull(reply);
         Objects.requireNonNull(objectiveUpdates);
+        Objects.requireNonNull(messageHandling);
 
         Long settleMs = dto.settleMs;
         if (settleMs == null || settleMs <= 0) {
@@ -86,13 +109,18 @@ public void settle(AsyncApiActionDto dto, AsyncApiReplyDto reply, LongFunction<L
 
         boolean coverageStable = dto.completion == AsyncApiActionDto.MessageCompletionStrategy.COVERAGE_STABLE
                 && dto.stableMs != null && dto.stableMs > 0;
+        boolean handlerReturn = dto.completion == AsyncApiActionDto.MessageCompletionStrategy.HANDLER_RETURN;
 
         long start = clock.nowMs();
         AsyncApiReplyDto.SettleEndReason end;
         try {
-            end = coverageStable
-                    ? waitForStableCoverage(start, settleMs, dto.stableMs, objectiveUpdates)
-                    : waitUntil(start + settleMs);
+            if (coverageStable) {
+                end = waitForStableCoverage(start, settleMs, dto.stableMs, objectiveUpdates);
+            } else if (handlerReturn) {
+                end = waitForHandlerReturn(start + settleMs, dto.address, messageHandling);
+            } else {
+                end = waitUntil(start + settleMs);
+            }
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             end = AsyncApiReplyDto.SettleEndReason.INTERRUPTED;
@@ -136,10 +164,42 @@ private AsyncApiReplyDto.SettleEndReason waitUntil(long deadline) throws Interru
     }
 
     /**
-     * Read the SUT's coverage every {@link #POLL_INTERVAL_MS}, and stop once it has gone
-     * {@code stableMs} without moving, or at the deadline, whichever comes first. The stable
-     * period is counted from the start of the wait, so a service that does nothing new after
-     * the driver returned is let go after {@code stableMs}.
+     * Stop once the SUT received something from the action's address in this action and is done
+     * with all of it, or at the deadline. What was received before the driver returned counts
+     * too, as does a leftover message from an earlier action consumed only now. A missing
+     * address or unreadable state means the whole window.
+     */
+    private AsyncApiReplyDto.SettleEndReason waitForHandlerReturn(
+            long deadline,
+            String address,
+            Function<String, MessageHandlingState> messageHandling
+    ) throws InterruptedException {
+
+        if (address == null) {
+            return waitUntil(deadline);
+        }
+
+        while (true) {
+            MessageHandlingState state = readOrNull(() -> messageHandling.apply(address),
+                    MESSAGE_HANDLING_FAILURE_WARNING);
+            if (state == null) {
+                return waitUntil(deadline);
+            }
+            if (state.isHandled()) {
+                return AsyncApiReplyDto.SettleEndReason.HANDLER_RETURN;
+            }
+
+            long now = clock.nowMs();
+            if (now >= deadline) {
+                return AsyncApiReplyDto.SettleEndReason.DEADLINE;
+            }
+            sleeper.sleep(Math.min(POLL_INTERVAL_MS, deadline - now));
+        }
+    }
+
+    /**
+     * Stop once the SUT's coverage has gone {@code stableMs} without moving, counted from the
+     * start of the wait, or at the deadline.
      */
     private AsyncApiReplyDto.SettleEndReason waitForStableCoverage(
             long start,

client-java/controller/src/main/java/org/evomaster/client/java/controller/internal/EMController.java

@@ -21,6 +21,7 @@
 import org.evomaster.client.java.controller.redis.RedisCommandExecutor;
 import org.evomaster.client.java.controller.redis.ReflectionBasedRedisClient;
 import org.evomaster.client.java.instrumentation.AdditionalInfo;
+import org.evomaster.client.java.instrumentation.EmittedMessage;
 import org.evomaster.client.java.instrumentation.InputProperties;
 import org.evomaster.client.java.instrumentation.TargetInfo;
 import org.evomaster.client.java.instrumentation.shared.StringSpecializationInfo;
@@ -603,6 +604,15 @@ private void prepareForTestResults(HttpServletRequest httpServletRequest) {
         assert trackRequestSource(httpServletRequest);
     }
 
+    private static EmittedMessageDto toDto(EmittedMessage message) {
+        EmittedMessageDto dto = new EmittedMessageDto();
+        dto.address = message.getAddress();
+        dto.payload = message.getPayload();
+        dto.payloadTruncated = message.isPayloadTruncated();
+        dto.payloadType = message.getPayloadType();
+        return dto;
+    }
+
     private Response computeTestResults(
             Set<Integer> ids,
             boolean killSwitch,
@@ -675,6 +685,9 @@ private Response computeTestResults(
                                         des.getRemotePort()
                                 )
                         ).collect(Collectors.toList());
+                        info.emittedMessages = a.getEmittedMessagesView().stream()
+                                .map(EMController::toDto)
+                                .collect(Collectors.toList());
                         info.stringSpecializations = new LinkedHashMap<>();
                         for (Map.Entry<String, Set<StringSpecializationInfo>> entry :
                                 a.getStringSpecializationsView().entrySet()) {
@@ -867,6 +880,12 @@ public Response newAction(
                 replyDto.index = index;
 
                 try {
+                    /*
+                        So that the driver's own broker client calls are not recorded as the SUT's,
+                        when the driver's classes are instrumented. Only this thread: the SUT's
+                        threads keep consuming meanwhile.
+                     */
+                    ExecutionTracer.setExecutingDriverAction(true);
                     sutController.executeAsyncApiAction(dto.asyncApiCall, replyDto);
                 } catch (Exception e) {
                     /*
@@ -879,6 +898,8 @@ public Response newAction(
                     replyDto.published = false;
                     replyDto.errorMessage = msg;
                     return Response.status(500).entity(WrappedResponseDto.withData(replyDto)).build();
+                } finally {
+                    ExecutionTracer.setExecutingDriverAction(false);
                 }
 
                 if (!Boolean.TRUE.equals(replyDto.published)) {
@@ -897,7 +918,8 @@ public Response newAction(
                     was already turned off for this action by newAction above, and must stay
                     off for the service's threads, not only this one.
                  */
-                asyncApiSettle.settle(dto.asyncApiCall, replyDto, sutController::getObjectiveUpdateCount);
+                asyncApiSettle.settle(dto.asyncApiCall, replyDto,
+                        sutController::getObjectiveUpdateCount, sutController::getMessageHandlingState);
 
                 return Response.status(200).entity(WrappedResponseDto.withData(replyDto)).build();
             }

client-java/controller/src/main/java/org/evomaster/client/java/controller/internal/SutController.java

@@ -60,6 +60,7 @@
 import org.evomaster.client.java.controller.redis.ReflectionBasedRedisClient;
 import org.evomaster.client.java.instrumentation.AdditionalInfo;
 import org.evomaster.client.java.instrumentation.BootTimeObjectiveInfo;
+import org.evomaster.client.java.instrumentation.MessageHandlingState;
 import org.evomaster.client.java.instrumentation.TargetInfo;
 import org.evomaster.client.java.instrumentation.staticstate.UnitsInfoRecorder;
 import org.evomaster.client.java.sql.*;
@@ -1883,6 +1884,18 @@ public Long getObjectiveUpdateCount(long timeoutMs) {
         return getObjectiveUpdateCount();
     }
 
+    /**
+     * What the SUT received, in the current action, from a broker address (eg a Kafka topic),
+     * and how much of it it is done handling, as recorded by the instrumentation. Read while an
+     * action is kept open.
+     *
+     * @param address where the SUT receives messages from
+     * @return the state, or null when unknown, eg because the SUT is not instrumented
+     */
+    public MessageHandlingState getMessageHandlingState(String address) {
+        return null;
+    }
+
     protected BootTimeInfoDto getBootTimeInfoDto(BootTimeObjectiveInfo info){
         if (info == null)
             return null;