From 56d0cf1941ad5faf77d1223766a4a0c495b29f8c Mon Sep 17 00:00:00 2001
From: Dave Maughan <davidamaughan@gmail.com>
Date: Tue, 28 Jun 2022 02:16:44 +0100
Subject: [PATCH] Check admin endpoint instead of metrics for Pulsar
 WaitStrategy.  (#5514)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Fixes #5513

Co-authored-by: Eddú Meléndez Gonzales <eddu.melendez@gmail.com>
Co-authored-by: Kevin Wittek <kiview@users.noreply.github.com>
---
 .../testcontainers/containers/PulsarContainer.java  | 13 ++++++++++++-
 .../containers/PulsarContainerTest.java             | 11 +++++++++++
 2 files changed, 23 insertions(+), 1 deletion(-)

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 069fccaae65..30c36db01d8 100644
--- a/modules/pulsar/src/main/java/org/testcontainers/containers/PulsarContainer.java
+++ b/modules/pulsar/src/main/java/org/testcontainers/containers/PulsarContainer.java
@@ -17,8 +17,14 @@ public class PulsarContainer extends GenericContainer<PulsarContainer> {
 
     public static final int BROKER_HTTP_PORT = 8080;
 
+    /**
+     * @deprecated The metrics endpoint is no longer being used for the WaitStrategy.
+     */
+    @Deprecated
     public static final String METRICS_ENDPOINT = "/metrics";
 
+    private static final String ADMIN_CLUSTERS_ENDPOINT = "/admin/v2/clusters";
+
     /**
      * See <a href="https://github.com/apache/pulsar/blob/master/pulsar-common/src/main/java/org/apache/pulsar/common/naming/SystemTopicNames.java">SystemTopicNames</a>.
      */
@@ -92,7 +98,12 @@ protected void setupCommandAndEnv() {
 
         List<WaitStrategy> waitStrategies = new ArrayList<>();
         waitStrategies.add(Wait.defaultWaitStrategy());
-        waitStrategies.add(Wait.forHttp(METRICS_ENDPOINT).forStatusCode(200).forPort(BROKER_HTTP_PORT));
+        waitStrategies.add(
+            Wait
+                .forHttp(ADMIN_CLUSTERS_ENDPOINT)
+                .forPort(BROKER_HTTP_PORT)
+                .forResponsePredicate("[\"standalone\"]"::equals)
+        );
         if (transactionsEnabled) {
             withEnv("PULSAR_PREFIX_transactionCoordinatorEnabled", "true");
             waitStrategies.add(Wait.forHttp(TRANSACTION_TOPIC_ENDPOINT).forStatusCode(200).forPort(BROKER_HTTP_PORT));
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 91397d00c40..29de6c4c872 100644
--- a/modules/pulsar/src/test/java/org/testcontainers/containers/PulsarContainerTest.java
+++ b/modules/pulsar/src/test/java/org/testcontainers/containers/PulsarContainerTest.java
@@ -122,6 +122,17 @@ public void testTransactionsAndFunctionsWorker() throws Exception {
         }
     }
 
+    @Test
+    public void testClusterFullyInitialized() throws Exception {
+        try (PulsarContainer pulsar = new PulsarContainer(PULSAR_IMAGE)) {
+            pulsar.start();
+
+            try (PulsarAdmin pulsarAdmin = PulsarAdmin.builder().serviceHttpUrl(pulsar.getHttpServiceUrl()).build()) {
+                assertThat(pulsarAdmin.clusters().getClusters()).hasSize(1).contains("standalone");
+            }
+        }
+    }
+
     protected void testPulsarFunctionality(String pulsarBrokerUrl) throws Exception {
         try (
             PulsarClient client = PulsarClient.builder().serviceUrl(pulsarBrokerUrl).build();