Fix announcement delay for TestingTrinoServer

client/trino-jdbc/src/test/java/io/trino/jdbc/TestJdbcExternalAuthentication.java

@@ -55,7 +55,6 @@
 import java.sql.DriverManager;
 import java.sql.SQLException;
 import java.sql.Statement;
-import java.time.Duration;
 import java.util.ConcurrentModificationException;
 import java.util.List;
 import java.util.Map;
@@ -114,7 +113,6 @@ public void setup()
                 .build();
         server.installPlugin(new TpchPlugin());
         server.createCatalog(TEST_CATALOG, "tpch");
-        server.waitForNodeRefresh(Duration.ofSeconds(10));
     }
 
     @AfterAll

client/trino-jdbc/src/test/java/io/trino/jdbc/TestJdbcPreparedStatement.java

@@ -44,7 +44,6 @@
 import java.sql.Time;
 import java.sql.Timestamp;
 import java.sql.Types;
-import java.time.Duration;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.LocalTime;
@@ -101,7 +100,6 @@ public void setup()
         server.installPlugin(new MemoryPlugin());
         server.createCatalog("blackhole", "blackhole");
         server.createCatalog("memory", "memory");
-        server.waitForNodeRefresh(Duration.ofSeconds(10));
 
         try (Connection connection = createConnection(false);
                 Statement statement = connection.createStatement()) {

client/trino-jdbc/src/test/java/io/trino/jdbc/TestJdbcWarnings.java

@@ -36,7 +36,6 @@
 import java.sql.SQLException;
 import java.sql.SQLWarning;
 import java.sql.Statement;
-import java.time.Duration;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
@@ -80,7 +79,6 @@ public void setupServer()
                 .build();
         server.installPlugin(new BlackHolePlugin());
         server.createCatalog("blackhole", "blackhole");
-        server.waitForNodeRefresh(Duration.ofSeconds(10));
 
         try (Connection connection = createConnection();
                 Statement statement = connection.createStatement()) {

client/trino-jdbc/src/test/java/io/trino/jdbc/TestTrinoDatabaseMetaData.java

@@ -55,7 +55,6 @@
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.sql.Types;
-import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -128,7 +127,6 @@ public void setupServer()
         countingMockConnector = new CountingMockConnector();
         server.installPlugin(countingMockConnector.getPlugin());
         server.createCatalog(COUNTING_CATALOG, "mock", ImmutableMap.of());
-        server.waitForNodeRefresh(Duration.ofSeconds(10));
 
         try (Connection connection = createConnection();
                 Statement statement = connection.createStatement()) {

client/trino-jdbc/src/test/java/io/trino/jdbc/TestTrinoDriver.java

@@ -112,7 +112,6 @@ public void setup()
         server.createCatalog(TEST_CATALOG, "tpch");
         server.installPlugin(new BlackHolePlugin());
         server.createCatalog("blackhole", "blackhole");
-        server.waitForNodeRefresh(java.time.Duration.ofSeconds(10));
         setupTestTables();
         executorService = newCachedThreadPool(daemonThreadsNamed(getClass().getSimpleName() + "-%s"));
     }

client/trino-jdbc/src/test/java/io/trino/jdbc/TestTrinoDriverAuth.java

@@ -36,7 +36,6 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
-import java.time.Duration;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
@@ -91,7 +90,6 @@ public void setup()
                 .build();
         server.installPlugin(new TpchPlugin());
         server.createCatalog(TEST_CATALOG, "tpch");
-        server.waitForNodeRefresh(Duration.ofSeconds(10));
     }
 
     @AfterAll

core/trino-main/src/main/java/io/trino/server/testing/TestingTrinoServer.java

@@ -118,20 +118,18 @@
 import java.io.UncheckedIOException;
 import java.net.URI;
 import java.nio.file.Path;
-import java.time.Duration;
-import java.time.Instant;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeoutException;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.io.MoreFiles.deleteRecursively;
 import static com.google.common.io.RecursiveDeleteOption.ALLOW_INSECURE;
 import static com.google.inject.util.Modules.EMPTY_MODULE;
+import static io.airlift.concurrent.MoreFutures.getFutureValue;
 import static java.lang.Integer.parseInt;
 import static java.nio.file.Files.createTempDirectory;
 import static java.nio.file.Files.isDirectory;
@@ -262,6 +260,7 @@ private TestingTrinoServer(
 
         if (coordinator) {
             serverProperties.put("catalog.store", "memory");
+            serverProperties.put("failure-detector.enabled", "false");
 
             // Reduce memory footprint in tests
             serverProperties.put("query.min-expire-age", "5s");
@@ -403,7 +402,7 @@ private TestingTrinoServer(
         EventListenerManager eventListenerManager = injector.getInstance(EventListenerManager.class);
         eventListeners.forEach(eventListenerManager::addEventListener);
 
-        injector.getInstance(Announcer.class).forceAnnounce();
+        getFutureValue(injector.getInstance(Announcer.class).forceAnnounce());
 
         refreshNodes();
     }
@@ -677,18 +676,6 @@ public final AllNodes refreshNodes()
         return nodeManager.getAllNodes();
     }
 
-    public void waitForNodeRefresh(Duration timeout)
-            throws InterruptedException, TimeoutException
-    {
-        Instant start = Instant.now();
-        while (refreshNodes().getActiveNodes().size() < 1) {
-            if (Duration.between(start, Instant.now()).compareTo(timeout) > 0) {
-                throw new TimeoutException("Timed out while waiting for the node to refresh");
-            }
-            MILLISECONDS.sleep(10);
-        }
-    }
-
     public <T> T getInstance(Key<T> key)
     {
         return injector.getInstance(key);

core/trino-main/src/test/java/io/trino/server/security/oauth2/BaseOAuth2WebUiAuthenticationFilterTest.java

@@ -120,7 +120,6 @@ public void setup()
                 .setProperties(getOAuth2Config(idpUrl))
                 .build();
         server.getInstance(Key.get(OAuth2Client.class)).load();
-        server.waitForNodeRefresh(Duration.ofSeconds(10));
         serverUri = server.getHttpsBaseUrl();
         uiUri = serverUri.resolve("/ui/");
 

core/trino-main/src/test/java/io/trino/server/security/oauth2/TestOAuth2WebUiAuthenticationFilterWithRefreshTokens.java

@@ -110,7 +110,6 @@ public void setup()
                         .buildOrThrow())
                 .build();
         server.getInstance(Key.get(OAuth2Client.class)).load();
-        server.waitForNodeRefresh(Duration.ofSeconds(10));
         serverUri = server.getHttpsBaseUrl();
         uiUri = serverUri.resolve("/ui/");
 

testing/trino-testing/src/main/java/io/trino/testing/DistributedQueryRunner.java

@@ -22,8 +22,6 @@
 import io.airlift.discovery.server.testing.TestingDiscoveryServer;
 import io.airlift.log.Logger;
 import io.airlift.log.Logging;
-import io.airlift.testing.Assertions;
-import io.airlift.units.Duration;
 import io.trino.Session;
 import io.trino.Session.SessionBuilder;
 import io.trino.cost.StatsCalculator;
@@ -78,6 +76,7 @@
 import static com.google.common.base.MoreObjects.firstNonNull;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Throwables.throwIfUnchecked;
+import static com.google.common.base.Verify.verify;
 import static com.google.inject.util.Modules.EMPTY_MODULE;
 import static io.airlift.log.Level.DEBUG;
 import static io.airlift.log.Level.ERROR;
@@ -88,8 +87,6 @@
 import static java.lang.Boolean.parseBoolean;
 import static java.lang.System.getenv;
 import static java.util.Objects.requireNonNull;
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import static java.util.concurrent.TimeUnit.SECONDS;
 
 public class DistributedQueryRunner
         implements QueryRunner
@@ -136,14 +133,15 @@ private DistributedQueryRunner(
     {
         requireNonNull(defaultSession, "defaultSession is null");
 
+        long start = System.nanoTime();
         setupLogging();
 
         try {
-            long start = System.nanoTime();
+            long discoveryStart = System.nanoTime();
             discoveryServer = new TestingDiscoveryServer(environment);
             closer.register(() -> closeUnchecked(discoveryServer));
             closer.register(() -> extraCloseables.forEach(DistributedQueryRunner::closeUnchecked));
-            log.info("Created TestingDiscoveryServer in %s", nanosSince(start).convertToMostSuccinctTimeUnit());
+            log.info("Created TestingDiscoveryServer in %s", nanosSince(discoveryStart));
 
             registerNewWorker = () -> createServer(false, extraProperties, environment, additionalModule, baseDataDir, Optional.empty(), Optional.of(ImmutableList.of()), ImmutableList.of());
 
@@ -196,7 +194,8 @@ private DistributedQueryRunner(
         defaultSession = defaultSession.toSessionRepresentation().toSession(coordinator.getSessionPropertyManager(), defaultSession.getIdentity().getExtraCredentials(), defaultSession.getExchangeEncryptionKey());
         this.trinoClient = closer.register(testingTrinoClientFactory.create(coordinator, defaultSession));
 
-        waitForAllNodesGloballyVisible();
+        ensureNodesGloballyVisible();
+        log.info("Created DistributedQueryRunner in %s", nanosSince(start));
     }
 
     private TestingTrinoServer createServer(
@@ -287,36 +286,20 @@ private static TestingTrinoServer createTestingTrinoServer(
     }
 
     public void addServers(int nodeCount)
-            throws Exception
     {
         for (int i = 0; i < nodeCount; i++) {
             registerNewWorker.run();
         }
-        waitForAllNodesGloballyVisible();
-    }
-
-    private void waitForAllNodesGloballyVisible()
-            throws InterruptedException
-    {
-        long start = System.nanoTime();
-        while (!allNodesGloballyVisible()) {
-            // TODO node announcement should be propagated faster when new node starts
-            Assertions.assertLessThan(nanosSince(start), new Duration(30, SECONDS));
-            MILLISECONDS.sleep(10);
-        }
-        log.info("Announced servers in %s", nanosSince(start).convertToMostSuccinctTimeUnit());
+        ensureNodesGloballyVisible();
     }
 
-    private boolean allNodesGloballyVisible()
+    private void ensureNodesGloballyVisible()
     {
         for (TestingTrinoServer server : servers) {
-            AllNodes allNodes = server.refreshNodes();
-            if (!allNodes.getInactiveNodes().isEmpty() ||
-                    (allNodes.getActiveNodes().size() != servers.size())) {
-                return false;
-            }
+            AllNodes nodes = server.refreshNodes();
+            verify(nodes.getInactiveNodes().isEmpty(), "Node manager has inactive nodes");
+            verify(nodes.getActiveNodes().size() == servers.size(), "Node manager has wrong active node count");
         }
-        return true;
     }
 
     public TestingTrinoClient getClient()