Check admin endpoint instead of metrics for Pulsar WaitStrategy. (#5514)

Fixes #5513

Co-authored-by: Eddú Meléndez Gonzales <[email protected]>
Co-authored-by: Kevin Wittek <[email protected]>

Author: 3 people

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));

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();