(pulsar-connectors) branch master updated: [fix][test] Bound MqttSinkIntegrationTest container startup so it cannot hang CI (#91)

[davekj]

This is an automated email from the ASF dual-hosted git repository.

david-streamlio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-connectors.git


The following commit(s) were added to refs/heads/master by this push:
     new 8528e369 [fix][test] Bound MqttSinkIntegrationTest container startup 
so it cannot hang CI (#91)
8528e369 is described below

commit 8528e369bb50adc7e9baa229ea1450cdd224ecd3
Author: David Kjerrumgaard <35466513+david-stream...@users.noreply.github.com>
AuthorDate: Fri Jul 10 09:25:56 2026 -0700

    [fix][test] Bound MqttSinkIntegrationTest container startup so it cannot 
hang CI (#91)
    
    The Mosquitto GenericContainer had no wait strategy and no startup
    timeout, so a stalled image pull hung @BeforeClass with no bound and
    took the whole Tests - Connectors job to its 45-minute limit — a
    cancellation rather than a failure, with no diagnostics. Observed on
    run 29100595779.
    
    Add Wait.forListeningPort() with a 3-minute startup timeout, plus a
    test-level timeOut backstop. An unsatisfiable startup now fails in
    about 20s with ContainerLaunchException instead of hanging.
    
    This test only began running in CI when #74 added integrationTest to
    the matrix, so the hazard was newly live.
---
 .../org/apache/pulsar/io/mqtt/MqttSinkIntegrationTest.java     | 10 ++++++++--
 1 file changed, 8 insertions(+), 2 deletions(-)

diff --git 
a/mqtt/src/integrationTest/java/org/apache/pulsar/io/mqtt/MqttSinkIntegrationTest.java
 
b/mqtt/src/integrationTest/java/org/apache/pulsar/io/mqtt/MqttSinkIntegrationTest.java
index 85b34c4e..101d3776 100644
--- 
a/mqtt/src/integrationTest/java/org/apache/pulsar/io/mqtt/MqttSinkIntegrationTest.java
+++ 
b/mqtt/src/integrationTest/java/org/apache/pulsar/io/mqtt/MqttSinkIntegrationTest.java
@@ -25,6 +25,7 @@ import static org.testng.Assert.assertTrue;
 import com.hivemq.client.mqtt.MqttClient;
 import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.BlockingQueue;
@@ -37,6 +38,7 @@ import org.apache.pulsar.io.core.SinkContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
 import org.testcontainers.utility.DockerImageName;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
@@ -51,7 +53,11 @@ public class MqttSinkIntegrationTest {
     private static final DockerImageName MOSQUITTO_IMAGE = 
DockerImageName.parse("eclipse-mosquitto:2");
 
     private final GenericContainer<?> mqttContainer = new 
GenericContainer<>(MOSQUITTO_IMAGE)
-            .withExposedPorts(MQTT_PORT);
+            .withExposedPorts(MQTT_PORT)
+            // Without an explicit wait strategy and startup timeout, a 
stalled image pull hangs
+            // @BeforeClass with no bound, taking the whole CI job to its 
45-minute limit.
+            .waitingFor(Wait.forListeningPort())
+            .withStartupTimeout(Duration.ofMinutes(3));
 
     @BeforeClass(alwaysRun = true)
     public void beforeClass() {
@@ -63,7 +69,7 @@ public class MqttSinkIntegrationTest {
         mqttContainer.stop();
     }
 
-    @Test
+    @Test(timeOut = 120_000)
     public void testWriteE2EWithMosquitto() throws Exception {
         BlockingQueue<String> receivedPayloads = new LinkedBlockingQueue<>();
         CountDownLatch ackLatch = new CountDownLatch(3);