[Part 08] AsyncAPI 3.x: a fire-and-forget Spring Kafka E2E

.gitignore

@@ -173,6 +173,7 @@ Migrations/
 /core-tests/e2e-tests/spring/spring-rest-rsa/target/
 /core-tests/e2e-tests/spring/spring-rest-dynamodb/target/
 /core-tests/e2e-tests/spring/spring-asyncapi-kafka/target/
+/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/target/
 /core-tests/e2e-tests/spring/spring-rest-h2-v1/em.yaml
 /core-tests/integration-tests/core-it/target/
 /core-tests/integration-tests/core-it/em.yaml

client-java/pom.xml

@@ -45,8 +45,6 @@
         <jersey.version>2.48</jersey.version>
         <!--  THOSE ARE ONLY FOR TESTING -->
         <springboot.version>2.5.4</springboot.version>
-        <!-- The version that Spring Boot 2.5.4 manages. 3.x needs Spring 6 and JDK 17 -->
-        <spring-kafka.version>2.7.6</spring-kafka.version>
     </properties>
 
 

core-tests/e2e-tests/e2e-tests-utils/src/test/java/org/evomaster/e2etests/utils/AsyncApiTestBase.java

@@ -1,17 +1,22 @@
 package org.evomaster.e2etests.utils;
 
+import com.google.inject.Injector;
 import com.webfuzzing.commons.faults.FaultCategory;
+import kotlin.Pair;
 import org.evomaster.client.java.controller.api.dto.problem.asyncapi.AsyncApiReplyDto;
+import org.evomaster.client.java.instrumentation.shared.ObjectiveNaming;
 import org.evomaster.core.Main;
 import org.evomaster.core.problem.asyncapi.data.AsyncApiCallResult;
 import org.evomaster.core.problem.asyncapi.data.AsyncApiIndividual;
 import org.evomaster.core.problem.asyncapi.data.AsyncApiOutcome;
 import org.evomaster.core.problem.enterprise.DetectedFault;
 import org.evomaster.core.search.Solution;
+import org.evomaster.core.search.service.IdMapper;
 
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
+import java.util.TreeSet;
 import java.util.stream.Collectors;
 
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -26,6 +31,34 @@ protected Solution<AsyncApiIndividual> initAndRun(List<String> args) {
         return (Solution<AsyncApiIndividual>) Main.initAndRun(args.toArray(new String[0]));
     }
 
+    /**
+     * Run the search, and keep the injector too, to read the names of the targets afterwards.
+     */
+    protected Pair<Injector, Solution<AsyncApiIndividual>> initAndDebug(List<String> args) {
+        Pair<Injector, Solution<?>> run = Main.initAndDebug(args.toArray(new String[0]));
+        return new Pair<>(run.getFirst(), (Solution<AsyncApiIndividual>) run.getSecond());
+    }
+
+    /**
+     * That some test in the solution ran code of {@code className}, a class of the SUT.
+     *
+     * A white-box assertion, for messages nothing answers. A whole class rather than a line keeps
+     * the test independent of how the SUT's sources are laid out.
+     *
+     * @param run       what {@link #initAndDebug} returned
+     * @param className the class, with dots, e.g. {@code com.foo.Bar}
+     */
+    protected void assertClassCovered(Pair<Injector, Solution<AsyncApiIndividual>> run, String className) {
+        IdMapper idMapper = run.getFirst().getInstance(IdMapper.class);
+        String target = ObjectiveNaming.classObjectiveName(className);
+        Set<String> coveredClasses = run.getSecond().getOverall().coveredTargets().stream()
+                .map(idMapper::getDescriptiveId)
+                .filter(id -> id.startsWith(ObjectiveNaming.CLASS))
+                .collect(Collectors.toCollection(TreeSet::new));
+        assertTrue(coveredClasses.contains(target),
+                "With seed " + defaultSeed + ": no test ran any code of " + className + ", only of " + coveredClasses);
+    }
+
     /**
      * The result of every message published, across the whole solution.
      */

core-tests/e2e-tests/spring/pom.xml

@@ -38,6 +38,7 @@
         <module>spring-rpc-thrift</module>
         <module>spring-mcp-bb</module>
         <module>spring-asyncapi-kafka</module>
+        <module>spring-asyncapi-kafka-listener</module>
     </modules>
 
     <dependencyManagement>

core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/pom.xml

@@ -0,0 +1,131 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>evomaster-e2e-tests-spring</artifactId>
+        <groupId>org.evomaster</groupId>
+        <version>6.2.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <!--
+        A shop back end that only consumes Kafka messages and never answers them: fire-and-forget
+        operations, consumed with Spring Kafka's @KafkaListener and JsonDeserializer, described by
+        an AsyncAPI document written for the test.
+
+        Not part of spring-asyncapi-kafka: with spring-kafka on the classpath, Spring Boot's Kafka
+        auto-configuration would apply to NCS too, which consumes with a plain KafkaConsumer of its
+        own.
+    -->
+    <artifactId>evomaster-e2e-tests-spring-asyncapi-kafka-listener</artifactId>
+    <packaging>jar</packaging>
+
+    <properties>
+        <skipTests>${skipAsyncApi}</skipTests>
+    </properties>
+
+    <dependencies>
+        <!--
+            Issue with Jersey (coming from client-java) using old version
+        -->
+        <dependency>
+            <groupId>javax.validation</groupId>
+            <artifactId>validation-api</artifactId>
+            <version>2.0.1.Final</version>
+        </dependency>
+        <dependency>
+            <groupId>org.evomaster</groupId>
+            <artifactId>evomaster-e2e-tests-utils</artifactId>
+            <type>test-jar</type>
+        </dependency>
+        <dependency>
+            <groupId>org.evomaster</groupId>
+            <artifactId>evomaster-client-java-controller</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.evomaster</groupId>
+            <artifactId>evomaster-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.evomaster</groupId>
+            <artifactId>evomaster-client-java-instrumentation</artifactId>
+            <type>test-jar</type>
+        </dependency>
+        <!--
+            The three below are versioned here rather than in the parent's dependencyManagement:
+            managed versions reach a sibling's transitive dependencies too, and spring-mcp-bb
+            resolves Spring Boot starters through a much newer Spring Boot than
+            ${springboot.version}.
+        -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+            <version>${springboot.version}</version>
+        </dependency>
+        <dependency>
+            <!-- the in-memory database one of the handlers writes to -->
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-jdbc</artifactId>
+            <version>${springboot.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+            <version>${spring-kafka.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.h2database</groupId>
+            <artifactId>h2</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+        <dependency>
+            <!--
+                spring-kafka 2.7 is built against kafka-clients 2.7/2.8, but only calls methods
+                that are still there in the version managed in the root pom, which is also the
+                one the driver and the generated suites use.
+            -->
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>kafka</artifactId>
+        </dependency>
+        <dependency>
+            <!--
+                Not used by any source in this module, but the test suites EvoMaster generates
+                import org.hamcrest.Matchers, and they are compiled against this classpath.
+            -->
+            <groupId>org.hamcrest</groupId>
+            <artifactId>hamcrest-all</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <!-- a broker in a container per test class, as the other container-backed E2Es do -->
+                    <reuseForks>false</reuseForks>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/ExpressDispatch.java

@@ -0,0 +1,21 @@
+package com.foo.asyncapi.shop;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Books a courier for an order shipped express. Only ever reached from
+ * {@link OrderListener#onOrderPlaced}, when the shipping is {@link OrderListener#EXPRESS}, so
+ * whether any of its code is covered says whether a search got there.
+ */
+public final class ExpressDispatch {
+
+    private static final Logger log = LoggerFactory.getLogger(ExpressDispatch.class);
+
+    private ExpressDispatch() {
+    }
+
+    static void book(OrderPlaced order) {
+        log.debug("Courier booked for order {}", order.getOrderId());
+    }
+}

core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/InvoiceListener.java

@@ -0,0 +1,46 @@
+package com.foo.asyncapi.shop;
+
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.ExecutorService;
+
+/**
+ * Consumes {@link InvoiceRequested}, and leaves the rendering to another thread: the listener
+ * returns at once, and the invoice is rendered a while later. Neither the listener returning nor
+ * the record being committed says the service is done; only waiting long enough does.
+ */
+@Component
+public class InvoiceListener {
+
+    /**
+     * How long rendering an invoice takes, in milliseconds. Long enough for the listener to have
+     * returned, and the offset to have been committed, well before the invoice is rendered.
+     */
+    static final long RENDERING_MS = 150;
+
+    private static final String VALUE_TYPE =
+            JsonDeserializer.VALUE_DEFAULT_TYPE + "=com.foo.asyncapi.shop.InvoiceRequested";
+
+    private final ExecutorService invoiceExecutor;
+
+    public InvoiceListener(ExecutorService invoiceExecutor) {
+        this.invoiceExecutor = invoiceExecutor;
+    }
+
+    @KafkaListener(id = ShopTopics.INVOICES_LISTENER, topics = ShopTopics.INVOICES_REQUESTED, properties = VALUE_TYPE)
+    public void onInvoiceRequested(InvoiceRequested request) {
+
+        invoiceExecutor.submit(() -> {
+            try {
+                Thread.sleep(RENDERING_MS);
+            } catch (InterruptedException e) {
+                //the service is shutting down
+                Thread.currentThread().interrupt();
+                return;
+            }
+            InvoiceRenderer.render(request);
+        });
+    }
+}

core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/InvoiceRenderer.java

@@ -0,0 +1,23 @@
+package com.foo.asyncapi.shop;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Renders an invoice. Only ever run on the thread {@link InvoiceListener} hands its work to, after
+ * the listener has returned, so whether any of its code is covered says whether the work done
+ * after the handler was still credited to the message.
+ */
+public final class InvoiceRenderer {
+
+    private static final Logger log = LoggerFactory.getLogger(InvoiceRenderer.class);
+
+    private InvoiceRenderer() {
+    }
+
+    static void render(InvoiceRequested request) {
+        Integer cents = request.getAmountCents();
+        String total = cents == null ? "?" : String.valueOf(cents);
+        log.debug("Invoice for order {} rendered, total {}", request.getOrderId(), total);
+    }
+}

core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/InvoiceRequested.java

@@ -0,0 +1,28 @@
+package com.foo.asyncapi.shop;
+
+/**
+ * A request to render the invoice of an order, as published on
+ * {@link ShopTopics#INVOICES_REQUESTED}. Every field is null when the message left it out.
+ */
+public class InvoiceRequested {
+
+    private String orderId;
+
+    private Integer amountCents;
+
+    public String getOrderId() {
+        return orderId;
+    }
+
+    public void setOrderId(String orderId) {
+        this.orderId = orderId;
+    }
+
+    public Integer getAmountCents() {
+        return amountCents;
+    }
+
+    public void setAmountCents(Integer amountCents) {
+        this.amountCents = amountCents;
+    }
+}

core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/OrderListener.java

@@ -0,0 +1,62 @@
+package com.foo.asyncapi.shop;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+import org.springframework.stereotype.Component;
+
+/**
+ * Consumes {@link OrderPlaced}. Two things it does are behind a string value that the contract
+ * does not list, so that a search drawing random strings does not get there, but one that sees
+ * what the service compares the value with does: express shipping, and an urgent priority.
+ */
+@Component
+public class OrderListener {
+
+    private static final Logger log = LoggerFactory.getLogger(OrderListener.class);
+
+    /**
+     * Compared with {@code equals}, the way a string-keyed rule usually is.
+     */
+    static final String EXPRESS = "EXPRESS";
+
+    /**
+     * Which class the record value is read into, for this listener only: each topic carries a
+     * type of its own, and the publisher sends no type header.
+     */
+    private static final String VALUE_TYPE =
+            JsonDeserializer.VALUE_DEFAULT_TYPE + "=com.foo.asyncapi.shop.OrderPlaced";
+
+    @KafkaListener(id = ShopTopics.ORDERS_LISTENER, topics = ShopTopics.ORDERS_PLACED, properties = VALUE_TYPE)
+    public void onOrderPlaced(OrderPlaced order) {
+
+        if (EXPRESS.equals(order.getShipping())) {
+            ExpressDispatch.book(order);
+        }
+
+        Priority priority = priorityOf(order);
+        if (priority == Priority.URGENT) {
+            UrgentEscalation.escalate(order);
+        }
+
+        log.debug("Order {} placed", order.getOrderId());
+    }
+
+    /**
+     * The order's priority, or null when it names none of them: an order is still taken when its
+     * priority is not understood.
+     */
+    private static Priority priorityOf(OrderPlaced order) {
+
+        if (order.getPriority() == null) {
+            return null;
+        }
+        try {
+            return Priority.valueOf(order.getPriority());
+        } catch (IllegalArgumentException e) {
+            log.debug("Unknown priority {}", order.getPriority());
+            return null;
+        }
+    }
+}