test(e2e-mqtt): fix race condition in test subscribing
@@ -130,6 +130,7 @@ class MqttTest extends BaseE2ETestCase {
try (AutoCloseable l = createCloseableLogListener(logListener)) {
int numberOfTopics = 100;
CountDownLatch messagesCdl = new CountDownLatch(numberOfTopics);
+ CountDownLatch allSubscribed = new CountDownLatch(numberOfTopics);
kernel = new Kernel().parseArgs("-r", tempRootDir.toAbsolutePath().toString());
setDefaultRunWithUser(kernel);
deviceProvisioningHelper.updateKernelConfigWithIotConfiguration(kernel, thingInfo, TEST_REGION.toString(),
@@ -148,10 +149,12 @@ class MqttTest extends BaseE2ETestCase {
} catch (ExecutionException | InterruptedException | TimeoutException e) {
logger.atError().kv("topic", topic).cause(e).log("Error subscribing");
}
+ allSubscribed.countDown();
}
});
assertTrue(interruptCdl.await(40, TimeUnit.SECONDS), "Connection should be interrupted for client 2");
subscribeFuture.cancel(true);
+ assertTrue(allSubscribed.await(30, TimeUnit.SECONDS), "All topics must be subscribed to");
for (int i = 0; i < numberOfTopics; i++) {
client.publish(PublishRequest.builder().topic("B/" + i).payload("What's up".getBytes(StandardCharsets.UTF_8))
.build()).get(5, TimeUnit.SECONDS);
