Add auto-configuration capability to ServerAdapters

Author: aguibert

docs/features/Examples.md

@@ -14,6 +14,7 @@ Sometimes code is worth a thousand words. Here are some pointers to working exam
 - [JAX-RS and JDBC applicaiton using a PostgreSQL database](https://github.com/MicroShed/microshed-testing/tree/master/sample-apps/jdbc-app) 
 - [JAX-RS application secured with MP JWT](https://github.com/MicroShed/microshed-testing/tree/master/sample-apps/jaxrs-mpjwt)
 - [JAX-RS and MongoDB application that depends on an external REST service](https://github.com/MicroShed/microshed-testing/tree/master/sample-apps/everything-app)
+- [Application using Apache Kafka messaging](https://github.com/MicroShed/microshed-testing/tree/master/sample-apps/kafka-app)
 - [Application with no Dockerfile using OpenLiberty adapter](https://github.com/MicroShed/microshed-testing/tree/master/sample-apps/liberty-app)
 
 ## Runtime examples:

docs/features/KafkaMessaging.md

@@ -0,0 +1,91 @@
+---
+layout: post
+title: "Kafka Messaging"
+order: 22
+---
+
+MicroShed Testing provides integration with applications using [Apache Kafka](https://kafka.apache.org/) for messaging. Apache Kafka is
+a messaging engine that is commonly used with Java microservice applications, and also is commonly used with [MicroProfile Reactive Messaging(https://github.com/eclipse/microprofile-reactive-messaging).
+
+## Sending and receiving messages from tests
+
+If an application purely uses Kafka Messaging for communication, a true-to-production way of testing is to also have the test client driving requests
+on the application via message passing. To do this, MicroShed Testing offers two annotations: `@KafkaConsumerConfig` and `@KafkaProducerConfig`
+
+### Example setup
+
+To begin using Kafka with MicroShed Testing, define a `KafkaContainer` in the test environment:
+
+```java
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.Network;
+// other imports ...
+
+public class AppContainerConfig implements SharedContainerConfiguration {
+
+    private static Network network = Network.newNetwork();
+
+    @Container
+    public static KafkaContainer kafka = new KafkaContainer()
+                    .withNetwork(network);
+
+    @Container
+    public static ApplicationContainer app = new ApplicationContainer()
+                    .withNetwork(network)
+                    .dependsOn(kafka);
+}
+```
+
+Runtimes such as OpenLiberty and Quarkus will be automatically configured if a `KafkaContainer` is present
+in the test environment. For other runtimes, you can link the containers together by using `kafka.withNetworkAlias("kafka")` 
+and `app.withEnv("<runtime-specific kafka bootstrap servers property>", "kafka:9092")`.
+
+
+### Example usage
+
+```java
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.microshed.testing.kafka.KafkaConsumerConfig;
+import org.microshed.testing.kafka.KafkaProducerConfig;
+// other imports ...
+
+@MicroShedTest
+@SharedContainerConfig(AppContainerConfig.class)
+public class KitchenEndpointIT {
+
+  @KafkaProducerConfig(keySerializer = StringSerializer.class,      // (1)
+                       valueSerializer = StringSerializer.class)
+  public static KafkaProducer<String, String> producer;
+
+  @KafkaConsumerConfig(keyDeserializer = StringDeserializer.class, 
+                       valueDeserializer = StringDeserializer.class, 
+                       groupId = "update-status", 
+                       topics = "statusTopic")                      // (2)
+  public static KafkaConsumer<String, String> consumer;
+  
+  @Test
+  public void myTest() {
+    // Use the producer to send messages
+    producer.send(...);
+
+    // Use the consumer to poll for records
+    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(30));
+    // ...
+  }
+}
+```
+
+1. Each `@KafkaProducerConfig` and `@KafkaConsumerConfig` must define a set of key/value [de]serializers
+that correspond to the key/value types defined in the `KafkaProducer` and `KafkaConsumer`.
+2. For `@KafkaConsumerConfig` zero or more `topics` may be specified to automatically subscribe the 
+injected `consumer` to the specified `topics`.
+
+
+## Additional resources
+
+- [Example application using Apache Kafka messaging](https://github.com/MicroShed/microshed-testing/tree/master/sample-apps/kafka-app)
+- [OpenLiberty blog on using MicroProfile Reactive Messaging](https://openliberty.io/blog/2019/09/13/microprofile-reactive-messaging.html)
+- [Quarkus guide on using Apache Kafka with Reactive Messaging](https://quarkus.io/guides/kafka)

modules/liberty/src/main/java/org/testcontainers/containers/liberty/LibertyAdapter.java

@@ -26,12 +26,24 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
 
+import org.junit.jupiter.api.extension.ExtensionConfigurationException;
+import org.microshed.testing.ApplicationEnvironment;
+import org.microshed.testing.testcontainers.ApplicationContainer;
+import org.microshed.testing.testcontainers.config.HollowTestcontainersConfiguration;
+import org.microshed.testing.testcontainers.config.TestcontainersConfiguration;
 import org.microshed.testing.testcontainers.spi.ServerAdapter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.images.builder.ImageFromDockerfile;
 
 public class LibertyAdapter implements ServerAdapter {
 
+    static final Logger LOG = LoggerFactory.getLogger(LibertyAdapter.class);
+
     private static String BASE_DOCKER_IMAGE = "openliberty/open-liberty:full-java8-openj9-ubi";
     private static final String CONFIG_FILE_PROP = "MICROSHED_TEST_LIBERTY_CONFIG_FILE";
 
@@ -126,4 +138,52 @@ public ImageFromDockerfile getDefaultImage(File appFile) {
         return image;
     }
 
+    @Override
+    public void configure(Set<GenericContainer<?>> containers) {
+        configureKafka(containers);
+    }
+
+    private void configureKafka(Set<GenericContainer<?>> containers) {
+        Class<?> KafkaContainer = tryLoad("org.testcontainers.containers.KafkaContainer");
+        if (KafkaContainer == null)
+            return;
+        Set<GenericContainer<?>> kafkaContainers = containers.stream()
+                        .filter(c -> KafkaContainer.isAssignableFrom(c.getClass()))
+                        .collect(Collectors.toSet());
+        if (kafkaContainers.size() != 1)
+            return;
+
+        // At this point we have found exactly 1 kafka container
+        GenericContainer<?> kafka = kafkaContainers.iterator().next();
+
+        // Configure app container with bootstrap server
+        String bootstrapProperty = "MP_MESSAGING_CONNECTOR_LIBERTY_KAFKA_BOOTSTRAP_SERVERS";
+        String bootstrapServer = null;
+        if (ApplicationEnvironment.Resolver.isSelected(TestcontainersConfiguration.class)) {
+            if (kafka.getNetworkAliases().size() == 0)
+                throw new ExtensionConfigurationException("Unable to configure kafka bootstrap server because no network alias is defined");
+            bootstrapServer = kafka.getNetworkAliases().get(kafka.getNetworkAliases().size() - 1) + ":9092";
+        } else if (ApplicationEnvironment.Resolver.isSelected(HollowTestcontainersConfiguration.class)) {
+            bootstrapServer = "localhost:9093";
+        } else {
+            return;
+        }
+        String finalBootstrapServers = bootstrapServer;
+        LOG.info("Auto-configuring ApplicationContainer instances with " + bootstrapProperty + "=" + finalBootstrapServers);
+        containers.stream()
+                        .filter(c -> ApplicationContainer.class.isAssignableFrom(c.getClass()))
+                        .filter(c -> c.getNetwork().equals(kafka.getNetwork()))
+                        .filter(c -> !c.getEnvMap().containsKey(bootstrapProperty))
+                        .filter(c -> !c.getEnvMap().containsKey(bootstrapProperty.toLowerCase()))
+                        .forEach(c -> c.withEnv(bootstrapProperty, finalBootstrapServers));
+    }
+
+    private static Class<?> tryLoad(String clazz) {
+        try {
+            return Class.forName(clazz, false, LibertyAdapter.class.getClassLoader());
+        } catch (ClassNotFoundException | LinkageError e) {
+            return null;
+        }
+    }
+
 }

modules/testcontainers/src/main/java/org/microshed/testing/testcontainers/config/TestcontainersConfiguration.java

@@ -21,6 +21,7 @@
 import java.lang.reflect.Field;
 import java.lang.reflect.Modifier;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -91,6 +92,14 @@ public void applyConfiguration(Class<?> testClass) {
             unsharedContainers.forEach(c -> c.setNetwork(Network.SHARED));
         }
 
+        // Give ServerAdapters a chance to do some auto-wiring between containers
+        allContainers().stream()
+                        .filter(c -> ApplicationContainer.class.isAssignableFrom(c.getClass()))
+                        .findFirst()
+                        .ifPresent(c -> {
+                            ((ApplicationContainer) c).getServerAdapter().configure(allContainers());
+                        });
+
         if (isJwtNeeded()) {
             allContainers().stream()
                             .filter(c -> ApplicationContainer.class.isAssignableFrom(c.getClass()))
@@ -238,7 +247,7 @@ protected Set<GenericContainer<?>> discoverContainers(Class<?> clazz) {
     protected Set<GenericContainer<?>> allContainers() {
         Set<GenericContainer<?>> all = new HashSet<>(unsharedContainers);
         all.addAll(sharedContainers);
-        return all;
+        return Collections.unmodifiableSet(all);
     }
 
     private static Class<?> tryLoad(String clazz) {

modules/testcontainers/src/main/java/org/microshed/testing/testcontainers/spi/ServerAdapter.java

@@ -21,7 +21,10 @@
 import java.io.File;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 
+import org.microshed.testing.testcontainers.ApplicationContainer;
+import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.images.builder.ImageFromDockerfile;
 
 /**
@@ -103,4 +106,17 @@ default ImageFromDockerfile getDefaultImage(File appFile) {
     default Optional<String> getReadinessPath() {
         return Optional.empty();
     }
+
+    /**
+     * An optional hook that may be implemented for the purposes of auto-wiring multiple
+     * containers the the test environment together.
+     * <p>
+     * For example, the <code>LibertyAdapter</code> detects the presence of a KafkaContainer
+     * and will automatically configure the ApplicationContainer to communicate with it by
+     * calling {@link ApplicationContainer#withEnv(String, String)}
+     *
+     * @param allContainers An unmodifiable set of the containers discovered in the test environment
+     */
+    default void configure(Set<GenericContainer<?>> allContainers) {
+    }
 }

sample-apps/kafka-app/src/test/java/org/example/app/KitchenEndpointIT.java

@@ -22,9 +22,6 @@
 
 import java.time.Duration;
 
-import javax.json.bind.Jsonb;
-import javax.json.bind.JsonbBuilder;
-
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;