address comment; adjust timeout

apache · Sep 18, 2024 · 3437ace · 3437ace
1 parent b247d06
commit 3437ace
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 3 deletions.
diff --git a/sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java b/sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java
@@ -599,7 +599,7 @@ static BlockingConnection createConnection(MQTT client) throws Exception {
       try {
         connecting.await(1, TimeUnit.MINUTES);
       } catch (TimeoutException e) {
-        LOG.warn("Connection to MQTT broker pending after waiting for 1 minute");
+        LOG.warn("Connection to {} pending after waiting for 1 minute", client.getHost());
         continue;
       }
       break;

diff --git a/sdks/java/io/mqtt/src/test/java/org/apache/beam/sdk/io/mqtt/MqttIOTest.java b/sdks/java/io/mqtt/src/test/java/org/apache/beam/sdk/io/mqtt/MqttIOTest.java
@@ -150,7 +150,7 @@ public void testRead() throws Exception {
                 .withConnectionConfiguration(
                     MqttIO.ConnectionConfiguration.create("tcp://localhost:" + port, "READ_TOPIC")
                         .withClientId("READ_PIPELINE"))
-                .withMaxReadTime(Duration.standardSeconds(3)));
+                .withMaxReadTime(Duration.standardSeconds(5)));
     PAssert.that(output)
         .containsInAnyOrder(
             "This is test 0".getBytes(StandardCharsets.UTF_8),
@@ -184,7 +184,7 @@ public void testRead() throws Exception {
                       pipelineConnected = true;
                     }
                   }
-                  Thread.sleep(2000);
+                  Thread.sleep(1000);
                 }
                 for (int i = 0; i < 10; i++) {
                   publishConnection.publish(