Solace with testcontainers - problem with ports

masz

Hi!

I have problem with creating Solace container via testcontainers plugin. Exception which appear:

09:15:15.486 [main] ERROR 🐳 [solace:1.0.0] - Could not start container

org.testcontainers.containers.ContainerLaunchException: Timed out waiting for container port to open (localhost ports: [51600, 51601, 51492, 51493, 51494, 51495, 51596, 51597, 51599] should be listening)

Below full log:

logs.txt

Also here is my class for Solace container:

import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.utility.DockerImageName;

import java.util.Arrays;

public class PubSubPlusContainer extends GenericContainer<PubSubPlusContainer> {
    private String adminUsername;
    private String adminPassword;

    public static final DockerImageName DEFAULT_IMAGE_NAME = DockerImageName.parse("solace/solace-pubsub-standard");
    public static final String DEFAULT_IMAGE_TAG = "latest";
    private static final String DEFAULT_ADMIN_USERNAME = "admin";
    private static final String DEFAULT_ADMIN_PASSWORD = "admin";
    private static final String DEFAULT_MAX_CONNECTION_COUNT = "100";
    private static final long DEFAULT_SHM_SIZE = (long) Math.pow(1024, 3); // 1 GB

    public PubSubPlusContainer() {
        this(DEFAULT_IMAGE_NAME.withTag(DEFAULT_IMAGE_TAG));
    }

    public PubSubPlusContainer(String dockerImageName) {
        this(DockerImageName.parse(dockerImageName));
    }

    public PubSubPlusContainer(DockerImageName dockerImageName) {
        super(dockerImageName);
        dockerImageName.assertCompatibleWith(DEFAULT_IMAGE_NAME);
        this.withExposedPorts(Arrays.stream(Port.values()).map(Port::getInternalPort).toArray(Integer[]::new));
        this.withAdminUsername(DEFAULT_ADMIN_USERNAME);
        this.withAdminPassword(DEFAULT_ADMIN_PASSWORD);
        this.withMaxConnectionCount(DEFAULT_MAX_CONNECTION_COUNT);
        this.withSharedMemorySize(DEFAULT_SHM_SIZE);
        this.waitingFor(Wait.forListeningPort());
    }

    public String getAdminUsername() {
        return adminUsername;
    }

    public String getAdminPassword() {
        return adminPassword;
    }

    public String getOrigin(Port port) {
        if (port.getProtocol() == null) {
            throw new IllegalArgumentException(String.format("Getting origin of port %s is not supported", port.name()));
        }

        return String.format("%s://%s:%s", port.getProtocol(), getContainerIpAddress(),
                getMappedPort(port.getInternalPort()));
    }

    public Integer getSshPort() {
        return getMappedPort(Port.SSH.getInternalPort());
    }

    public PubSubPlusContainer withAdminUsername(String username) {
        adminUsername = username;
        return withEnv("username_admin_globalaccesslevel", adminUsername);
    }

    public PubSubPlusContainer withAdminPassword(String password) {
        adminPassword = password;
        return withEnv("username_admin_password", password);
    }

    public PubSubPlusContainer withMaxConnectionCount(String maxConnectionCount) {
        return withEnv("system_scaling_maxconnectioncount", maxConnectionCount);
    }

    public enum Port {
        AMQP(5672, "amqp"),
        MQTT(1883, "tcp"),
        MQTT_WEB(8000, "ws"),
        REST(9000, "http"),
        SEMP(8080, "http"),
        SMF(55555, "tcp"),
        SMF_TLS(55443, "tcps"),
        SMF_WEB(8008, "ws"),
        SSH(2222, null);

        private final int containerPort;
        private final String protocol;

        Port(int containerPort, String protocol) {
            this.containerPort = containerPort;
            this.protocol = protocol;
        }

        public int getInternalPort() {
            return containerPort;
        }

        private String getProtocol() {
            return protocol;
        }
    }

}

I contact testcontainers team, but they responded that I need to try contact Solace whether I'm using correctly that image "solace/solace-pubsub-standard". Could somebody help me?

masz

Topic can be closed, problem was here:

public enum Port {
        AMQP(5672, "amqp"),
        MQTT(1883, "tcp"),
        MQTT_WEB(8000, "ws"),
        REST(9000, "http"),
        SEMP(8080, "http"),
        SMF(55555, "tcp"),
        SMF_TLS(55443, "tcps"),
        SMF_WEB(8008, "ws"),
        SSH(2222, null);

        private final int containerPort;
        private final String protocol;

        Port(int containerPort, String protocol) {
            this.containerPort = containerPort;
            this.protocol = protocol;
        }

        public int getInternalPort() {
            return containerPort;
        }

        private String getProtocol() {
            return protocol;
        }
    }

with port 55443 (after removing that port container started).

amackenzie

I am not familiar with testcontainers, but at 1st glance, it seems to be that the ports are not exposed. From reading the docs, testcontainers will map the PS+ ports to randomized ports to avoid collisions with other containers. But I think these need to be exposed to the external network.

When doing a normal docker run, you will typically map the internal ports to an external port.

For example, to make PubSub+ Broker Manager visible on your network, you would add -p 8080:8090 on docker run will map the internal port 8080 to the host port of 8090. Then your users would access the Manager UI via 8090 from your network.

So, is the missing piece that the testcontainers random ports are not exposed?

I see there is something called .withExposedPorts() in the config of the container. Perhaps you need to call that with the enum you have above? Networking and communicating with containers - Testcontainers

masz

Thanks for answer, but I'm doing that 😅 here:

 public PubSubPlusContainer(DockerImageName dockerImageName) {
        super(dockerImageName);
        dockerImageName.assertCompatibleWith(DEFAULT_IMAGE_NAME);
        this.withExposedPorts(Arrays.stream(Port.values()).map(Port::getInternalPort).toArray(Integer[]::new));
        this.withAdminUsername(DEFAULT_ADMIN_USERNAME);
        this.withAdminPassword(DEFAULT_ADMIN_PASSWORD);
        this.withMaxConnectionCount(DEFAULT_MAX_CONNECTION_COUNT);
        this.withSharedMemorySize(DEFAULT_SHM_SIZE);
        this.waitingFor(Wait.forListeningPort());
    }

I compare my implementation with implementation of KafkaContainer (which works perfectly fine), but I couldn't find difference :/.

Example of class for Kafka which work:

import com.github.dockerjava.api.command.InspectContainerResponse;
import org.testcontainers.containers.Container.ExecResult;
import org.testcontainers.utility.DockerImageName;

public class KafkaContainer extends GenericContainer<KafkaContainer> {
    private static final DockerImageName DEFAULT_IMAGE_NAME = DockerImageName.parse("confluentinc/cp-kafka");
    private static final String DEFAULT_TAG = "5.4.3";
    public static final int KAFKA_PORT = 9093;
    public static final int ZOOKEEPER_PORT = 2181;
    private static final String DEFAULT_INTERNAL_TOPIC_RF = "1";
    protected String externalZookeeperConnect;

    /** @deprecated */
    @Deprecated
    public KafkaContainer() {
        this(DEFAULT_IMAGE_NAME.withTag("5.4.3"));
    }

    /** @deprecated */
    @Deprecated
    public KafkaContainer(String confluentPlatformVersion) {
        this(DEFAULT_IMAGE_NAME.withTag(confluentPlatformVersion));
    }

    public KafkaContainer(DockerImageName dockerImageName) {
        super(dockerImageName);
        this.externalZookeeperConnect = null;
        dockerImageName.assertCompatibleWith(new DockerImageName[]{DEFAULT_IMAGE_NAME});
        this.withExposedPorts(new Integer[]{9093});
        this.withEnv("KAFKA_LISTENERS", "PLAINTEXT://0.0.0.0:9093,BROKER://0.0.0.0:9092");
        this.withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT");
        this.withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "BROKER");
        this.withEnv("KAFKA_BROKER_ID", "1");
        this.withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1");
        this.withEnv("KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS", "1");
        this.withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1");
        this.withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1");
        this.withEnv("KAFKA_LOG_FLUSH_INTERVAL_MESSAGES", "9223372036854775807");
        this.withEnv("KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "0");
    }

    public KafkaContainer withEmbeddedZookeeper() {
        this.externalZookeeperConnect = null;
        return (KafkaContainer)this.self();
    }

    public KafkaContainer withExternalZookeeper(String connectString) {
        this.externalZookeeperConnect = connectString;
        return (KafkaContainer)this.self();
    }

    public String getBootstrapServers() {
        return String.format("PLAINTEXT://%s:%s", this.getHost(), this.getMappedPort(9093));
    }

    protected void configure() {
        this.withEnv("KAFKA_ADVERTISED_LISTENERS", String.format("BROKER://%s:9092", this.getNetwork() != null ? this.getNetworkAliases().get(0) : "localhost"));
        String command = "#!/bin/bash\n";
        if (this.externalZookeeperConnect != null) {
            this.withEnv("KAFKA_ZOOKEEPER_CONNECT", this.externalZookeeperConnect);
        } else {
            this.addExposedPort(2181);
            this.withEnv("KAFKA_ZOOKEEPER_CONNECT", "localhost:2181");
            command = command + "echo 'clientPort=2181' > zookeeper.properties\n";
            command = command + "echo 'dataDir=/var/lib/zookeeper/data' >> zookeeper.properties\n";
            command = command + "echo 'dataLogDir=/var/lib/zookeeper/log' >> zookeeper.properties\n";
            command = command + "zookeeper-server-start zookeeper.properties &\n";
        }

        command = command + "echo '' > /etc/confluent/docker/ensure \n";
        command = command + "/etc/confluent/docker/run \n";
        this.withCommand(new String[]{"sh", "-c", command});
    }

    protected void containerIsStarted(InspectContainerResponse containerInfo) {
        try {
            String brokerAdvertisedListener = this.brokerAdvertisedListener(containerInfo);
            ExecResult result = this.execInContainer(new String[]{"kafka-configs", "--alter", "--bootstrap-server", brokerAdvertisedListener, "--entity-type", "brokers", "--entity-name", (String)this.getEnvMap().get("KAFKA_BROKER_ID"), "--add-config", "advertised.listeners=[" + String.join(",", this.getBootstrapServers(), brokerAdvertisedListener) + "]"});
            if (result.getExitCode() != 0) {
                throw new IllegalStateException(result.toString());
            }
        } catch (Throwable var4) {
            throw var4;
        }
    }

    protected String brokerAdvertisedListener(InspectContainerResponse containerInfo) {
        return String.format("BROKER://%s:%s", containerInfo.getConfig().getHostName(), "9092");
    }
}

masz

Topic can be closed, problem was here:

public enum Port {
        AMQP(5672, "amqp"),
        MQTT(1883, "tcp"),
        MQTT_WEB(8000, "ws"),
        REST(9000, "http"),
        SEMP(8080, "http"),
        SMF(55555, "tcp"),
        SMF_TLS(55443, "tcps"),
        SMF_WEB(8008, "ws"),
        SSH(2222, null);

        private final int containerPort;
        private final String protocol;

        Port(int containerPort, String protocol) {
            this.containerPort = containerPort;
            this.protocol = protocol;
        }

        public int getInternalPort() {
            return containerPort;
        }

        private String getProtocol() {
            return protocol;
        }
    }

with port 55443 (after removing that port container started).

aditya

Hi @masz , I am getting the similar issue as yours but from the beginning I am not using the 55443 port still I am getting the same issue.

Please find the below code:

private static final int SEMP_PORT = 8080;
private static final int REST_PORT = 9000;
private static final int MQTT_WEB_PORT = 8000;
private static final int MQTT_PORT = 1883;
private static final int AMQP_PORT = 5672;
private static final int WEB_TRANSPORT = 8008;
private static final int SMF_PORT = 55555;
static {
    LOGGER.info("Starting Solace PubSub+ Docker Container");
    solace = new GenericContainer<>("solace/solace-pubsub-standard:latest")
            .withExposedPorts(SMF_PORT, WEB_TRANSPORT, AMQP_PORT, MQTT_PORT, MQTT_WEB_PORT, REST_PORT, SEMP_PORT)
            .withSharedMemorySize(1000000000L).withEnv("username_admin_globalaccesslevel", ADMIN_USER)
            .withEnv("username_admin_password", ADMIN_PASSWORD).withEnv("system_scaling_interconnection", "1200")
            .waitingFor(Wait.forListeningPort());

    solace.start();

Please can you check and help me out to resolve this issue.