From 4f9594d5c1df72851ec174f75679726f6b892ef7 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Edd=C3=BA=20Mel=C3=A9ndez=20Gonzales?=
 <eddu.melendez@gmail.com>
Date: Mon, 21 Oct 2024 20:49:19 -0500
Subject: [PATCH] Add support for apachepulsar/pulsar-all image (#9448)

`apachepulsar/pulsar-all` brings connectors in the image.
---
 .../containers/PulsarContainer.java           |  4 +-
 .../containers/AbstractPulsar.java            | 67 +++++++++++++++++++
 .../CompatibleApachePulsarImageTest.java      | 41 ++++++++++++
 .../containers/PulsarContainerTest.java       | 62 +----------------
 4 files changed, 113 insertions(+), 61 deletions(-)
 create mode 100644 modules/pulsar/src/test/java/org/testcontainers/containers/AbstractPulsar.java
 create mode 100644 modules/pulsar/src/test/java/org/testcontainers/containers/CompatibleApachePulsarImageTest.java

diff --git a/modules/pulsar/src/main/java/org/testcontainers/containers/PulsarContainer.java b/modules/pulsar/src/main/java/org/testcontainers/containers/PulsarContainer.java
index 0181d7144d9..36485ad76a0 100644
--- a/modules/pulsar/src/main/java/org/testcontainers/containers/PulsarContainer.java
+++ b/modules/pulsar/src/main/java/org/testcontainers/containers/PulsarContainer.java
@@ -7,7 +7,7 @@
 /**
  * Testcontainers implementation for Apache Pulsar.
  * <p>
- * Supported image: {@code apachepulsar/pulsar}
+ * Supported images: {@code apachepulsar/pulsar}, {@code apachepulsar/pulsar-all}
  * <p>
  * Exposed ports:
  * <ul>
@@ -64,7 +64,7 @@ public PulsarContainer(String pulsarVersion) {
 
     public PulsarContainer(final DockerImageName dockerImageName) {
         super(dockerImageName);
-        dockerImageName.assertCompatibleWith(DockerImageName.parse("apachepulsar/pulsar"));
+        dockerImageName.assertCompatibleWith(DEFAULT_IMAGE_NAME, DockerImageName.parse("apachepulsar/pulsar-all"));
         withExposedPorts(BROKER_PORT, BROKER_HTTP_PORT);
         setWaitStrategy(waitAllStrategy);
     }
diff --git a/modules/pulsar/src/test/java/org/testcontainers/containers/AbstractPulsar.java b/modules/pulsar/src/test/java/org/testcontainers/containers/AbstractPulsar.java
new file mode 100644
index 00000000000..5246fc6b8ad
--- /dev/null
+++ b/modules/pulsar/src/test/java/org/testcontainers/containers/AbstractPulsar.java
@@ -0,0 +1,67 @@
+package org.testcontainers.containers;
+
+import org.apache.pulsar.client.admin.ListTopicsOptions;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.transaction.Transaction;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class AbstractPulsar {
+
+    public static final String TEST_TOPIC = "test_topic";
+
+    protected void testPulsarFunctionality(String pulsarBrokerUrl) throws Exception {
+        try (
+            PulsarClient client = PulsarClient.builder().serviceUrl(pulsarBrokerUrl).build();
+            Consumer consumer = client.newConsumer().topic(TEST_TOPIC).subscriptionName("test-subs").subscribe();
+            Producer<byte[]> producer = client.newProducer().topic(TEST_TOPIC).create()
+        ) {
+            producer.send("test containers".getBytes());
+            CompletableFuture<Message> future = consumer.receiveAsync();
+            Message message = future.get(5, TimeUnit.SECONDS);
+
+            assertThat(new String(message.getData())).isEqualTo("test containers");
+        }
+    }
+
+    protected void testTransactionFunctionality(String pulsarBrokerUrl) throws Exception {
+        try (
+            PulsarClient client = PulsarClient.builder().serviceUrl(pulsarBrokerUrl).enableTransaction(true).build();
+            Consumer<String> consumer = client
+                .newConsumer(Schema.STRING)
+                .topic("transaction-topic")
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                .subscriptionName("test-transaction-sub")
+                .subscribe();
+            Producer<String> producer = client
+                .newProducer(Schema.STRING)
+                .sendTimeout(0, TimeUnit.SECONDS)
+                .topic("transaction-topic")
+                .create()
+        ) {
+            final Transaction transaction = client.newTransaction().build().get();
+            producer.newMessage(transaction).value("first").send();
+            transaction.commit();
+            Message<String> message = consumer.receive();
+            assertThat(message.getValue()).isEqualTo("first");
+        }
+    }
+
+    protected void assertTransactionsTopicCreated(PulsarAdmin pulsarAdmin) throws PulsarAdminException {
+        final List<String> topics = pulsarAdmin
+            .topics()
+            .getPartitionedTopicList("pulsar/system", ListTopicsOptions.builder().includeSystemTopic(true).build());
+        assertThat(topics).contains("persistent://pulsar/system/transaction_coordinator_assign");
+    }
+}
diff --git a/modules/pulsar/src/test/java/org/testcontainers/containers/CompatibleApachePulsarImageTest.java b/modules/pulsar/src/test/java/org/testcontainers/containers/CompatibleApachePulsarImageTest.java
new file mode 100644
index 00000000000..b7a9cd3ff4c
--- /dev/null
+++ b/modules/pulsar/src/test/java/org/testcontainers/containers/CompatibleApachePulsarImageTest.java
@@ -0,0 +1,41 @@
+package org.testcontainers.containers;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.testcontainers.utility.DockerImageName;
+
+@RunWith(Parameterized.class)
+public class CompatibleApachePulsarImageTest extends AbstractPulsar {
+
+    @Parameterized.Parameters(name = "{0}")
+    public static String[] params() {
+        return new String[] { "apachepulsar/pulsar:3.0.0", "apachepulsar/pulsar-all:3.0.0" };
+    }
+
+    @Parameterized.Parameter
+    public String imageName;
+
+    @Test
+    public void testUsage() throws Exception {
+        try (PulsarContainer pulsar = new PulsarContainer(DockerImageName.parse(this.imageName));) {
+            pulsar.start();
+            final String pulsarBrokerUrl = pulsar.getPulsarBrokerUrl();
+
+            testPulsarFunctionality(pulsarBrokerUrl);
+        }
+    }
+
+    @Test
+    public void testTransactions() throws Exception {
+        try (PulsarContainer pulsar = new PulsarContainer(DockerImageName.parse(this.imageName)).withTransactions();) {
+            pulsar.start();
+
+            try (PulsarAdmin pulsarAdmin = PulsarAdmin.builder().serviceHttpUrl(pulsar.getHttpServiceUrl()).build()) {
+                assertTransactionsTopicCreated(pulsarAdmin);
+            }
+            testTransactionFunctionality(pulsar.getPulsarBrokerUrl());
+        }
+    }
+}
diff --git a/modules/pulsar/src/test/java/org/testcontainers/containers/PulsarContainerTest.java b/modules/pulsar/src/test/java/org/testcontainers/containers/PulsarContainerTest.java
index 29321afca82..779a3a6114a 100644
--- a/modules/pulsar/src/test/java/org/testcontainers/containers/PulsarContainerTest.java
+++ b/modules/pulsar/src/test/java/org/testcontainers/containers/PulsarContainerTest.java
@@ -1,29 +1,16 @@
 package org.testcontainers.containers;
 
-import org.apache.pulsar.client.admin.ListTopicsOptions;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminException;
-import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.Message;
-import org.apache.pulsar.client.api.Producer;
-import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.Schema;
-import org.apache.pulsar.client.api.SubscriptionInitialPosition;
-import org.apache.pulsar.client.api.transaction.Transaction;
 import org.junit.Test;
 import org.testcontainers.utility.DockerImageName;
 
 import java.time.Duration;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
-public class PulsarContainerTest {
-
-    public static final String TEST_TOPIC = "test_topic";
+public class PulsarContainerTest extends AbstractPulsar {
 
     private static final DockerImageName PULSAR_IMAGE = DockerImageName.parse("apachepulsar/pulsar:3.0.0");
 
@@ -84,7 +71,8 @@ public void shouldNotEnableFunctionsWorkerByDefault() throws Exception {
     public void shouldWaitForFunctionsWorkerStarted() throws Exception {
         try (
             // constructorWithFunctionsWorker {
-            PulsarContainer pulsar = new PulsarContainer(PULSAR_IMAGE).withFunctionsWorker();
+            PulsarContainer pulsar = new PulsarContainer(DockerImageName.parse("apachepulsar/pulsar:3.0.0"))
+                .withFunctionsWorker();
             // }
         ) {
             pulsar.start();
@@ -111,13 +99,6 @@ public void testTransactions() throws Exception {
         }
     }
 
-    private void assertTransactionsTopicCreated(PulsarAdmin pulsarAdmin) throws PulsarAdminException {
-        final List<String> topics = pulsarAdmin
-            .topics()
-            .getPartitionedTopicList("pulsar/system", ListTopicsOptions.builder().includeSystemTopic(true).build());
-        assertThat(topics).contains("persistent://pulsar/system/transaction_coordinator_assign");
-    }
-
     @Test
     public void testTransactionsAndFunctionsWorker() throws Exception {
         try (PulsarContainer pulsar = new PulsarContainer(PULSAR_IMAGE).withTransactions().withFunctionsWorker()) {
@@ -149,41 +130,4 @@ public void testStartupTimeoutIsHonored() {
                 .hasRootCauseMessage("Precondition failed: timeout must be greater than zero");
         }
     }
-
-    protected void testPulsarFunctionality(String pulsarBrokerUrl) throws Exception {
-        try (
-            PulsarClient client = PulsarClient.builder().serviceUrl(pulsarBrokerUrl).build();
-            Consumer consumer = client.newConsumer().topic(TEST_TOPIC).subscriptionName("test-subs").subscribe();
-            Producer<byte[]> producer = client.newProducer().topic(TEST_TOPIC).create()
-        ) {
-            producer.send("test containers".getBytes());
-            CompletableFuture<Message> future = consumer.receiveAsync();
-            Message message = future.get(5, TimeUnit.SECONDS);
-
-            assertThat(new String(message.getData())).isEqualTo("test containers");
-        }
-    }
-
-    protected void testTransactionFunctionality(String pulsarBrokerUrl) throws Exception {
-        try (
-            PulsarClient client = PulsarClient.builder().serviceUrl(pulsarBrokerUrl).enableTransaction(true).build();
-            Consumer<String> consumer = client
-                .newConsumer(Schema.STRING)
-                .topic("transaction-topic")
-                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
-                .subscriptionName("test-transaction-sub")
-                .subscribe();
-            Producer<String> producer = client
-                .newProducer(Schema.STRING)
-                .sendTimeout(0, TimeUnit.SECONDS)
-                .topic("transaction-topic")
-                .create()
-        ) {
-            final Transaction transaction = client.newTransaction().build().get();
-            producer.newMessage(transaction).value("first").send();
-            transaction.commit();
-            Message<String> message = consumer.receive();
-            assertThat(message.getValue()).isEqualTo("first");
-        }
-    }
 }