From 69d2b6f4b588964f50b54dc2ce67e5c0d8e6fc63 Mon Sep 17 00:00:00 2001
From: Esta Nagy <nagyesta@gmail.com>
Date: Thu, 16 Jan 2025 23:11:09 +0100
Subject: [PATCH] Add Azure Event Hubs Emulator container to Azure module - Fix
 code review findings

Signed-off-by: Esta Nagy <nagyesta@gmail.com>
---
 .../AzureEventHubsEmulatorContainer.java      |  35 ++++-
 .../AzureEventHubsEmulatorContainerTest.java  | 134 +++++++++---------
 2 files changed, 93 insertions(+), 76 deletions(-)

diff --git a/modules/azure/src/main/java/org/testcontainers/azure/AzureEventHubsEmulatorContainer.java b/modules/azure/src/main/java/org/testcontainers/azure/AzureEventHubsEmulatorContainer.java
index 7149b73281a..02bf61bff05 100644
--- a/modules/azure/src/main/java/org/testcontainers/azure/AzureEventHubsEmulatorContainer.java
+++ b/modules/azure/src/main/java/org/testcontainers/azure/AzureEventHubsEmulatorContainer.java
@@ -32,25 +32,39 @@ public class AzureEventHubsEmulatorContainer extends GenericContainer<AzureEvent
         "mcr.microsoft.com/azure-messaging/eventhubs-emulator"
     );
 
-    private final AzuriteContainer azuriteContainer;
+    private AzuriteContainer azuriteContainer;
 
     private boolean useKafka;
 
     /**
      * @param dockerImageName specified docker image name to run
      */
-    public AzureEventHubsEmulatorContainer(
-        final DockerImageName dockerImageName,
-        final AzuriteContainer azuriteContainer
-    ) {
+    public AzureEventHubsEmulatorContainer(final String dockerImageName) {
+        this(DockerImageName.parse(dockerImageName));
+    }
+
+    /**
+     * @param dockerImageName specified docker image name to run
+     */
+    public AzureEventHubsEmulatorContainer(final DockerImageName dockerImageName) {
         super(dockerImageName);
         dockerImageName.assertCompatibleWith(DEFAULT_IMAGE_NAME);
-        this.azuriteContainer = azuriteContainer;
-        dependsOn(this.azuriteContainer);
         waitingFor(Wait.forLogMessage(".*Emulator Service is Successfully Up!.*", 1));
         withExposedPorts(DEFAULT_AMQP_PORT);
     }
 
+    /**
+     * * Sets the Azurite dependency needed by the Event Hubs Container,
+     *
+     * @param azuriteContainer The Azurite container used by Event HUbs as a dependency
+     * @return this
+     */
+    public AzureEventHubsEmulatorContainer withAzuriteContainer(final AzuriteContainer azuriteContainer) {
+        this.azuriteContainer = azuriteContainer;
+        dependsOn(this.azuriteContainer);
+        return this;
+    }
+
     /**
      * Provide the broker configuration to the container.
      *
@@ -83,6 +97,13 @@ public AzureEventHubsEmulatorContainer enableKafka() {
 
     @Override
     protected void configure() {
+        if (azuriteContainer == null) {
+            throw new IllegalStateException(
+                "The image " +
+                getDockerImageName() +
+                " requires an Azurite container. Please provide one with the withAzuriteContainer method!"
+            );
+        }
         final String azuriteHost = azuriteContainer.getNetworkAliases().get(0);
         withEnv("BLOB_SERVER", azuriteHost);
         withEnv("METADATA_SERVER", azuriteHost);
diff --git a/modules/azure/src/test/java/org/testcontainers/azure/AzureEventHubsEmulatorContainerTest.java b/modules/azure/src/test/java/org/testcontainers/azure/AzureEventHubsEmulatorContainerTest.java
index fbd072881bf..8e78d287cb5 100644
--- a/modules/azure/src/test/java/org/testcontainers/azure/AzureEventHubsEmulatorContainerTest.java
+++ b/modules/azure/src/test/java/org/testcontainers/azure/AzureEventHubsEmulatorContainerTest.java
@@ -17,11 +17,9 @@
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
-import org.awaitility.Awaitility;
 import org.junit.Rule;
 import org.junit.Test;
 import org.testcontainers.containers.Network;
-import org.testcontainers.utility.DockerImageName;
 import org.testcontainers.utility.MountableFile;
 
 import java.time.Duration;
@@ -32,6 +30,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.tuple;
+import static org.awaitility.Awaitility.await;
 import static org.awaitility.Awaitility.waitAtMost;
 
 public class AzureEventHubsEmulatorContainerTest {
@@ -39,61 +38,58 @@ public class AzureEventHubsEmulatorContainerTest {
     @Rule
     // network {
     public Network network = Network.newNetwork();
-
     // }
 
     @Rule
     // azuriteContainer {
     public AzuriteContainer azuriteContainer = new AzuriteContainer("mcr.microsoft.com/azure-storage/azurite:3.33.0")
-        .withNetwork(network);
-
+            .withNetwork(network);
     // }
 
     @Rule
     // emulatorContainer {
     public AzureEventHubsEmulatorContainer emulator = new AzureEventHubsEmulatorContainer(
-        DockerImageName.parse("mcr.microsoft.com/azure-messaging/eventhubs-emulator:2.0.1"),
-        azuriteContainer
+            "mcr.microsoft.com/azure-messaging/eventhubs-emulator:2.0.1"
     )
-        .acceptLicense()
-        .enableKafka() //optional
-        .withNetwork(network)
-        .withConfig(MountableFile.forClasspathResource("/eventhubs_config.json"));
-
+            .acceptLicense()
+            .enableKafka() //optional
+            .withNetwork(network)
+            .withConfig(MountableFile.forClasspathResource("/eventhubs_config.json"))
+            .withAzuriteContainer(azuriteContainer);
     // }
 
     @Test
     public void testWithEventHubsClient() {
         try (
-            // createProducerAndConsumer {
-            EventHubProducerClient producer = new EventHubClientBuilder()
-                .connectionString(emulator.getConnectionString())
-                .fullyQualifiedNamespace("emulatorNs1")
-                .eventHubName("eh1")
-                .buildProducerClient();
-            EventHubConsumerClient consumer = new EventHubClientBuilder()
-                .connectionString(emulator.getConnectionString())
-                .fullyQualifiedNamespace("emulatorNs1")
-                .eventHubName("eh1")
-                .consumerGroup("cg1")
-                .buildConsumerClient()
-            // }
+                // createProducerAndConsumer {
+                EventHubProducerClient producer = new EventHubClientBuilder()
+                        .connectionString(emulator.getConnectionString())
+                        .fullyQualifiedNamespace("emulatorNs1")
+                        .eventHubName("eh1")
+                        .buildProducerClient();
+                EventHubConsumerClient consumer = new EventHubClientBuilder()
+                        .connectionString(emulator.getConnectionString())
+                        .fullyQualifiedNamespace("emulatorNs1")
+                        .eventHubName("eh1")
+                        .consumerGroup("cg1")
+                        .buildConsumerClient()
+                // }
         ) {
             producer.send(Collections.singletonList(new EventData("test")));
 
             waitAtMost(Duration.ofSeconds(30))
-                .pollDelay(Duration.ofSeconds(5))
-                .untilAsserted(() -> {
-                    IterableStream<PartitionEvent> events = consumer.receiveFromPartition(
-                        "0",
-                        1,
-                        EventPosition.earliest(),
-                        Duration.ofSeconds(2)
-                    );
-                    Optional<PartitionEvent> event = events.stream().findFirst();
-                    assertThat(event).isPresent();
-                    assertThat(event.get().getData().getBodyAsString()).isEqualTo("test");
-                });
+                    .pollDelay(Duration.ofSeconds(5))
+                    .untilAsserted(() -> {
+                        IterableStream<PartitionEvent> events = consumer.receiveFromPartition(
+                                "0",
+                                1,
+                                EventPosition.earliest(),
+                                Duration.ofSeconds(2)
+                        );
+                        Optional<PartitionEvent> event = events.stream().findFirst();
+                        assertThat(event).isPresent();
+                        assertThat(event.get().getData().getBodyAsString()).isEqualTo("test");
+                    });
         }
     }
 
@@ -101,18 +97,18 @@ public void testWithEventHubsClient() {
     public void testWithKafkaClient() throws Exception {
         // kafkaProperties {
         ImmutableMap<String, String> commonProperties = ImmutableMap
-            .<String, String>builder()
-            .put("bootstrap.servers", emulator.getBootstrapServers())
-            .put("sasl.mechanism", "PLAIN")
-            .put("security.protocol", "SASL_PLAINTEXT")
-            .put(
-                "sasl.jaas.config",
-                String.format(
-                    "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"$ConnectionString\" password=\"%s\";",
-                    emulator.getConnectionString()
+                .<String, String>builder()
+                .put("bootstrap.servers", emulator.getBootstrapServers())
+                .put("sasl.mechanism", "PLAIN")
+                .put("security.protocol", "SASL_PLAINTEXT")
+                .put(
+                        "sasl.jaas.config",
+                        String.format(
+                                "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"$ConnectionString\" password=\"%s\";",
+                                emulator.getConnectionString()
+                        )
                 )
-            )
-            .build();
+                .build();
         // }
 
         Properties producerProperties = new Properties();
@@ -123,34 +119,34 @@ public void testWithKafkaClient() throws Exception {
         consumerProperties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "tc-" + UUID.randomUUID());
         consumerProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
         consumerProperties.putAll(commonProperties);
+
         try (
-            KafkaProducer<String, String> producer = new KafkaProducer<>(
-                producerProperties,
-                new StringSerializer(),
-                new StringSerializer()
-            );
-            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(
-                consumerProperties,
-                new StringDeserializer(),
-                new StringDeserializer()
-            );
+                KafkaProducer<String, String> producer = new KafkaProducer<>(
+                        producerProperties,
+                        new StringSerializer(),
+                        new StringSerializer()
+                );
+                KafkaConsumer<String, String> consumer = new KafkaConsumer<>(
+                        consumerProperties,
+                        new StringDeserializer(),
+                        new StringDeserializer()
+                );
         ) {
             String topicName = "eh1";
             consumer.subscribe(Collections.singletonList(topicName));
 
             producer.send(new ProducerRecord<>(topicName, "testcontainers", "rulezzz")).get();
 
-            Awaitility
-                .await()
-                .atMost(Duration.ofSeconds(10))
-                .untilAsserted(() -> {
-                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
-
-                    assertThat(records)
-                        .hasSize(1)
-                        .extracting(ConsumerRecord::topic, ConsumerRecord::key, ConsumerRecord::value)
-                        .containsExactly(tuple(topicName, "testcontainers", "rulezzz"));
-                });
+            await()
+                    .atMost(Duration.ofSeconds(10))
+                    .untilAsserted(() -> {
+                        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
+
+                        assertThat(records)
+                                .hasSize(1)
+                                .extracting(ConsumerRecord::topic, ConsumerRecord::key, ConsumerRecord::value)
+                                .containsExactly(tuple(topicName, "testcontainers", "rulezzz"));
+                    });
 
             consumer.unsubscribe();
         }