package org.testcontainers.containers;

import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.utility.TestcontainersConfiguration;

/**
 * This container wraps Apache pulsar running in stanalone mode
 */
public class PulsarContainer extends GenericContainer<PulsarContainer> {

    public static final int PULSAR_PORT = 6850;

    public PulsarContainer() {
        this("2.1.0-incubating");
    }

    public PulsarContainer(String pulsarVersion) {
        super(TestcontainersConfiguration.getInstance().getPulsarImage() + ":" + pulsarVersion);
        withExposedPorts(PULSAR_PORT);
        withCommand("/bin/bash", "-c", "" +
            "servicePort=6850 webServicePort=8280 webServicePortTls=8643 bin/apply-config-from-env.py conf/proxy.conf && " +
            "bin/pulsar standalone & " +
            "bin/pulsar proxy --zookeeper-servers localhost:2181 --global-zookeeper-servers localhost:2181"
        );

        waitingFor(Wait.forLogMessage(".*messaging service is ready.*\\s", 1));
    }

    public String getPulsarBrokerUrl() {
        return String.format("pulsar://%s:%s", this.getContainerIpAddress(), this.getFirstMappedPort());
    }

}

package org.testcontainers.containers;

import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.utility.TestcontainersConfiguration;

/**
 * This container wraps Apache Pulsar running in standalone mode
 */
public class PulsarContainer extends GenericContainer<PulsarContainer> {

    public static final int BROKER_PORT = 6650;
    public static final int BROKER_HTTP_PORT = 8080;
    public static final String METRICS_ENDPOINT = "/metrics";

    private static final String PULSAR_VERSION = "2.1.0-incubating";

    public PulsarContainer() {
        this(PULSAR_VERSION);
    }

    public PulsarContainer(String pulsarVersion) {
        super(TestcontainersConfiguration.getInstance().getPulsarImage() + ":" + pulsarVersion);
        withExposedPorts(BROKER_PORT, BROKER_HTTP_PORT);
        withCommand("/pulsar/bin/pulsar", "standalone", "--no-functions-worker", "-nss");
        waitingFor(Wait.forHttp(METRICS_ENDPOINT).forStatusCode(200).forPort(BROKER_HTTP_PORT));
    }

    public String getPulsarBrokerUrl() {
        return String.format("pulsar://%s:%s", getContainerIpAddress(), getMappedPort(BROKER_PORT));
    }

    public String getHttpServiceUrl() {
        return String.format("http://%s:%s", getContainerIpAddress(), getMappedPort(BROKER_HTTP_PORT));
    }
}


Update Pulsar Container (#858)