IT: use COUNTING allocator and check that broker container has started

Signed-off-by: Evgeny Malygin <emalygin@bloomberg.net>

Author: 678098

bmq-sdk/src/test/docker/config/bmqbrkrcfg.json

@@ -1,6 +1,6 @@
 {
     "taskConfig": {
-        "allocatorType": "STACKTRACETEST",
+        "allocatorType": "COUNTING",
         "allocationLimit": 34359738368,
         "logController": {
             "fileName": "/var/local/bmq/logs/logs.%T.%p",

bmq-sdk/src/test/java/com/bloomberg/bmq/it/util/BmqBrokerContainer.java

@@ -53,6 +53,7 @@ public class BmqBrokerContainer implements BmqBroker {
     private static final String OUTPUT_FILENAME = "output.log";
     private static final long MAX_CONTAINER_WAIT_TIME_MS = 5000;
     private static final long CONTAINER_HEALTH_CHECK_DT_MS = 100;
+    private static final String BROKER_READY_MESSAGE = "Cluster (local) is available";
 
     private final SessionOptions sessionOptions;
     private final DockerClient client;
@@ -194,22 +195,36 @@ public void start() {
                     totalTimeMs += CONTAINER_HEALTH_CHECK_DT_MS) {
                 Thread.sleep(CONTAINER_HEALTH_CHECK_DT_MS);
 
+                // Check if container is still running
                 InspectContainerResponse resp = client.inspectContainerCmd(containerId).exec();
                 if (!resp.getState().getRunning()) {
                     logger.error(
                             "Container '{}' is not running, status = '{}'",
                             containerId,
                             resp.getState().getStatus());
                     throw new RuntimeException(
-                            String.format("Failed to start container '{}'", containerId));
+                            String.format("Failed to start container '%s'", containerId));
+                }
+
+                // Check logs for broker readiness message
+                if (isBrokerReady()) {
+                    logger.info("Container '{}' broker is ready", containerId);
+                    break;
+                }
+
+                if (totalTimeMs + CONTAINER_HEALTH_CHECK_DT_MS >= MAX_CONTAINER_WAIT_TIME_MS) {
+                    throw new RuntimeException(
+                            String.format(
+                                    "Timeout waiting for broker ready in container '%s'",
+                                    containerId));
                 }
             }
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             return;
         }
 
-        logger.info("Container '{}' is running", containerId);
+        logger.info("Container '{}' broker is ready", containerId);
 
         // For BlazingMQ broker running in container default tier should be the 'lcl-{guest
         // hostname}'
@@ -357,6 +372,29 @@ public void onNext(Frame item) {
         }
     }
 
+    private boolean isBrokerReady() {
+        final boolean[] ready = {false};
+        ResultCallback.Adapter<Frame> callback =
+                new ResultCallback.Adapter<Frame>() {
+                    @Override
+                    public void onNext(Frame item) {
+                        if (item.toString().contains(BROKER_READY_MESSAGE)) {
+                            ready[0] = true;
+                        }
+                    }
+                };
+        try {
+            client.logContainerCmd(containerId)
+                    .withStdOut(true)
+                    .withStdErr(true)
+                    .exec(callback)
+                    .awaitCompletion();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+        return ready[0];
+    }
+
     private String getHostname() {
         // Get container hostname
         logger.info("Get '{}' container hostname", containerName);