From 90f0f138ad70ee110b67fa97b324f2fd7c3afd55 Mon Sep 17 00:00:00 2001
From: E W <emmettwalsh@gmail.com>
Date: Mon, 29 Jul 2024 12:36:52 +0100
Subject: [PATCH] Add container network support (#353)

Co-authored-by: CORPORATE\ewalsh <ewalsh@marketaxess.com>
---
 .../kafka/DockerKafkaEnvironment.java         | 31 ++++++++++++++++---
 1 file changed, 27 insertions(+), 4 deletions(-)

diff --git a/kafka-test/src/main/java/io/specmesh/kafka/DockerKafkaEnvironment.java b/kafka-test/src/main/java/io/specmesh/kafka/DockerKafkaEnvironment.java
index d56b0966..2785bb85 100644
--- a/kafka-test/src/main/java/io/specmesh/kafka/DockerKafkaEnvironment.java
+++ b/kafka-test/src/main/java/io/specmesh/kafka/DockerKafkaEnvironment.java
@@ -85,6 +85,7 @@ public final class DockerKafkaEnvironment
     private final Map<String, String> srEnv;
     private final Set<AclBinding> aclBindings;
     private final Optional<Credentials> adminUser;
+    private final Optional<Network> explicitNetwork;
 
     private Network network;
     private KafkaContainer kafkaBroker;
@@ -108,7 +109,8 @@ private DockerKafkaEnvironment(
             final Optional<DockerImageName> srDockerImage,
             final Map<String, String> srEnv,
             final Set<AclBinding> aclBindings,
-            final Optional<Credentials> adminUser) {
+            final Optional<Credentials> adminUser,
+            final Optional<Network> explicitNetwork) {
         this.startUpTimeout = requireNonNull(startUpTimeout, "startUpTimeout");
         this.startUpAttempts = startUpAttempts;
         this.kafkaDockerImage = requireNonNull(kafkaDockerImage, "kafkaDockerImage");
@@ -117,6 +119,7 @@ private DockerKafkaEnvironment(
         this.srEnv = Map.copyOf(requireNonNull(srEnv, "srEnv"));
         this.aclBindings = Set.copyOf(requireNonNull(aclBindings, "aclBindings"));
         this.adminUser = requireNonNull(adminUser, "credentials");
+        this.explicitNetwork = requireNonNull(explicitNetwork, "explicitNetwork");
         tearDown();
     }
 
@@ -187,7 +190,7 @@ private void setUp() {
             return;
         }
 
-        network = Network.newNetwork();
+        network = explicitNetwork.orElseGet(Network::newNetwork);
 
         kafkaBroker =
                 new KafkaContainer(kafkaDockerImage)
@@ -232,7 +235,9 @@ private void tearDown() {
         }
 
         if (network != null) {
-            network.close();
+            if (explicitNetwork.isEmpty()) {
+                network.close();
+            }
             network = null;
         }
 
@@ -247,6 +252,17 @@ private void installAcls() {
         }
     }
 
+    public String testNetworkKafkaBootstrapServers() {
+        return "PLAINTEXT://" + kafkaBroker.getNetworkAliases().get(0) + ":9092";
+    }
+
+    public String testNetworkSchemeRegistryServer() {
+        return "http://"
+                + schemaRegistry.getNetworkAliases().get(0)
+                + ":"
+                + SchemaRegistryContainer.SCHEMA_REGISTRY_PORT;
+    }
+
     /** Builder of {@link DockerKafkaEnvironment}. */
     public static final class Builder {
 
@@ -271,6 +287,12 @@ public static final class Builder {
         private final Map<String, String> userPasswords = new LinkedHashMap<>();
         private boolean enableAcls = false;
         private final Set<AclBinding> aclBindings = new HashSet<>();
+        private Optional<Network> explicitNetwork = Optional.empty();
+
+        public Builder withNetwork(final Network network) {
+            this.explicitNetwork = Optional.of(network);
+            return this;
+        }
 
         /**
          * Customise the startup count.
@@ -436,7 +458,8 @@ public DockerKafkaEnvironment build() {
                     srImage,
                     srEnv,
                     aclBindings,
-                    adminUser());
+                    adminUser(),
+                    explicitNetwork);
         }
 
         private Optional<Credentials> adminUser() {