INFRA-417 Improve driver DSL test stability (#6415)

* Move log messages that are not useful in typical usage from info to debug level to reduce log spam.
* Add node startup check before attempting to connect.

node/src/integration-test-slow/kotlin/net/corda/node/services/distributed/DistributedServiceTests.kt

@@ -29,6 +29,8 @@ import org.junit.Ignore
 import org.junit.Test
 import rx.Observable
 import java.util.*
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
 
 class DistributedServiceTests {
     private lateinit var alice: NodeHandle
@@ -157,9 +159,9 @@ class DistributedServiceTests {
 
         // The distribution of requests should be very close to sg like 16/17/17 as by default artemis does round robin
         println("Notarisation distribution: $notarisationsPerNotary")
-        require(notarisationsPerNotary.size == 3)
+        assertEquals(3, notarisationsPerNotary.size)
         // We allow some leeway for artemis as it doesn't always produce perfect distribution
-        require(notarisationsPerNotary.values.all { it > 10 })
+        assertTrue { notarisationsPerNotary.values.all { it > 10 } }
     }
 
     private fun issueCash(amount: Amount<Currency>) {

testing/core-test-utils/src/main/kotlin/net/corda/testing/core/Expect.kt

@@ -164,14 +164,14 @@ fun <S, E : Any> S.genericExpectEvents(
         }
         val next = state.nextState(event)
         val expectedStates = state.getExpectedEvents()
-        log.info("$event :: ${expectedStates.map { it.simpleName }} -> ${next?.second?.getExpectedEvents()?.map { it.simpleName }}")
+        log.debug("$event :: ${expectedStates.map { it.simpleName }} -> ${next?.second?.getExpectedEvents()?.map { it.simpleName }}")
         if (next == null) {
             val message = "Got $event, did not match any expectations of type ${expectedStates.map { it.simpleName }}"
             if (isStrict) {
                 finishFuture.setException(Exception(message))
                 state = ExpectComposeState.Finished()
             } else {
-                log.info("$message, discarding event as isStrict=false")
+                log.debug("$message, discarding event as isStrict=false")
             }
         } else {
             state = next.second

testing/node-driver/src/main/kotlin/net/corda/testing/node/internal/DriverDSLImpl.kt

@@ -9,6 +9,7 @@ import com.typesafe.config.ConfigRenderOptions
 import com.typesafe.config.ConfigValue
 import com.typesafe.config.ConfigValueFactory
 import net.corda.client.rpc.CordaRPCClient
+import net.corda.client.rpc.RPCException
 import net.corda.cliutils.CommonCliConstants.BASE_DIR
 import net.corda.core.concurrent.CordaFuture
 import net.corda.core.concurrent.firstOf
@@ -218,16 +219,24 @@ class DriverDSLImpl(
         }
     }
 
-    private fun establishRpc(config: NodeConfig, processDeathFuture: CordaFuture<out Process>): CordaFuture<CordaRPCOps> {
+    /**
+     * @param pollInterval the interval to wait between attempting to connect, if
+     * a connection attempt fails.
+     */
+    private fun establishRpc(config: NodeConfig,
+                             processDeathFuture: CordaFuture<out Process>): CordaFuture<CordaRPCOps> {
         val rpcAddress = config.corda.rpcOptions.address
         val clientRpcSslOptions = clientSslOptionsCompatibleWith(config.corda.rpcOptions)
         val client = CordaRPCClient(rpcAddress, sslConfiguration = clientRpcSslOptions)
-        val connectionFuture = poll(executorService, "RPC connection") {
+        val connectionFuture = poll(
+                executorService = executorService,
+                pollName = "RPC connection",
+                pollInterval = RPC_CONNECT_POLL_INTERVAL) {
             try {
                 config.corda.rpcUsers[0].run { client.start(username, password) }
-            } catch (e: Exception) {
+            } catch (e: RPCException) {
                 if (processDeathFuture.isDone) throw e
-                log.info("Exception while connecting to RPC, retrying to connect at $rpcAddress", e)
+                log.info("Failed to connect to RPC at $rpcAddress")
                 null
             }
         }
@@ -673,6 +682,7 @@ class DriverDSLImpl(
                     }
             )
             val nodeFuture: CordaFuture<NodeHandle> = nodeAndThreadFuture.flatMap { (node, thread) ->
+                node.node.nodeReadyFuture.get() // Wait for the node to be ready before we connect to the node
                 establishRpc(config, openFuture()).flatMap { rpc ->
                     visibilityHandle.listen(rpc).map {
                         InProcessImpl(rpc.nodeInfo(), rpc, config.corda, webAddress, useHTTPS, thread, onNodeExit, node)
@@ -779,6 +789,7 @@ class DriverDSLImpl(
     }
 
     companion object {
+        private val RPC_CONNECT_POLL_INTERVAL: Duration = 100.millis
         internal val log = contextLogger()
 
         // While starting with inProcess mode, we need to have different names to avoid clashes

testing/node-driver/src/main/kotlin/net/corda/testing/node/internal/InternalTestUtils.kt

@@ -210,6 +210,10 @@ fun addressMustNotBeBoundFuture(executorService: ScheduledExecutorService, hostA
     }
 }
 
+/**
+ * @param pollInterval the interval running the background task.
+ * @param warnCount number of iterations to poll before printing a warning message.
+ */
 fun <A> poll(
         executorService: ScheduledExecutorService,
         pollName: String,