Skip to content

Commit

Permalink
Check admin endpoint instead of metrics for Pulsar WaitStrategy. (#5514)
Browse files Browse the repository at this point in the history
Fixes #5513

Co-authored-by: Eddú Meléndez Gonzales <eddu.melendez@gmail.com>
Co-authored-by: Kevin Wittek <kiview@users.noreply.github.com>
  • Loading branch information
3 people authored Jun 28, 2022
1 parent c0eb042 commit 56d0cf1
Show file tree
Hide file tree
Showing 2 changed files with 23 additions and 1 deletion.
Original file line number Diff line number Diff line change
Expand Up @@ -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>.
*/
Expand Down Expand Up @@ -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));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down

0 comments on commit 56d0cf1

Please sign in to comment.