BFT notary demo (#725)

* Rename raft-notary-demo project to notary-demo
* Refactor serialisation filtering to allow BFT SMaRt to work, it no longer relies on the jdk.serialFilter system property
* In NodeBasedTest remove whitespace in node directory names for consistency with cordform and driver

.gitignore

@@ -32,7 +32,7 @@ lib/dokka.jar
 .idea/libraries
 .idea/shelf
 .idea/dataSources
-/gradle-plugins/.idea
+/gradle-plugins/.idea/
 
 # Include the -parameters compiler option by default in IntelliJ required for serialization.
 !.idea/compiler.xml
@@ -84,8 +84,10 @@ crashlytics-build.properties
 docs/virtualenv/
 
 # bft-smart
-node/bft-smart-config/currentView
+config/currentView
-node/config/currentView
+
+# vim
+*.swp
 
 # Files you may find useful to have in your working directory.
 PLAN

.idea/compiler.xml

@@ -61,10 +61,10 @@
       <module name="node_integrationTest" target="1.8" />
       <module name="node_main" target="1.8" />
       <module name="node_test" target="1.8" />
+      <module name="notary-demo_main" target="1.8" />
+      <module name="notary-demo_test" target="1.8" />
       <module name="quasar-hook_main" target="1.8" />
       <module name="quasar-hook_test" target="1.8" />
-      <module name="raft-notary-demo_main" target="1.8" />
-      <module name="raft-notary-demo_test" target="1.8" />
       <module name="rpc_integrationTest" target="1.8" />
       <module name="rpc_main" target="1.8" />
       <module name="rpc_smokeTest" target="1.8" />

constants.properties

@@ -1,4 +1,4 @@
-gradlePluginsVersion=0.12.1
+gradlePluginsVersion=0.12.2
 kotlinVersion=1.1.2
 guavaVersion=21.0
 bouncycastleVersion=1.56

cordform-common/src/main/java/net/corda/cordform/CordformNode.java

@@ -25,7 +25,8 @@ public class CordformNode {
     public List<String> advertisedServices = emptyList();
 
     /**
-     * If running a distributed notary, a list of node addresses for joining the Raft cluster
+     * If running a Raft notary cluster, the address of at least one node in the cluster, or leave blank to start a new cluster.
+     * If running a BFT notary cluster, the addresses of all nodes in the cluster.
      */
     public List<String> notaryClusterAddresses = emptyList();
     /**
@@ -82,11 +83,18 @@ public class CordformNode {
     }
 
     /**
-     * Set the port which to bind the Copycat (Raft) node to
+     * Set the port which to bind the Copycat (Raft) node to.
      *
      * @param notaryPort The Raft port.
      */
     public void notaryNodePort(Integer notaryPort) {
         config = config.withValue("notaryNodeAddress", ConfigValueFactory.fromAnyRef(DEFAULT_HOST + ':' + notaryPort));
     }
+
+    /**
+     * @param id The (0-based) BFT replica ID.
+     */
+    public void bftReplicaId(Integer id) {
+        config = config.withValue("bftReplicaId", ConfigValueFactory.fromAnyRef(id));
+    }
 }

core/src/main/kotlin/net/corda/core/Utils.kt

@@ -110,8 +110,9 @@ infix fun <T> ListenableFuture<T>.failure(body: (Throwable) -> Unit): Listenable
 infix fun <F, T> ListenableFuture<F>.map(mapper: (F) -> T): ListenableFuture<T> = Futures.transform(this, { (mapper as (F?) -> T)(it) })
 infix fun <F, T> ListenableFuture<F>.flatMap(mapper: (F) -> ListenableFuture<T>): ListenableFuture<T> = Futures.transformAsync(this) { mapper(it!!) }
 
-inline fun <T, reified R> Collection<T>.mapToArray(transform: (T) -> R) = run {
+inline fun <T, reified R> Collection<T>.mapToArray(transform: (T) -> R) = mapToArray(transform, iterator(), size)
-    val iterator = iterator()
+inline fun <reified R> IntProgression.mapToArray(transform: (Int) -> R) = mapToArray(transform, iterator(), 1 + (last - first) / step)
+inline fun <T, reified R> mapToArray(transform: (T) -> R, iterator: Iterator<T>, size: Int) = run {
     var expected = 0
     Array(size) {
         expected++ == it || throw UnsupportedOperationException("Array constructor is non-sequential!")

core/src/main/kotlin/net/corda/core/internal/ShutdownHook.kt

@@ -0,0 +1,26 @@
+package net.corda.core.internal
+
+interface ShutdownHook {
+    /**
+     * Safe to call from the block passed into [addShutdownHook].
+     */
+    fun cancel()
+}
+
+/**
+ * The given block will run on most kinds of termination including SIGTERM, but not on SIGKILL.
+ * @return An object via which you can cancel the hook.
+ */
+fun addShutdownHook(block: () -> Unit): ShutdownHook {
+    val hook = Thread { block() }
+    val runtime = Runtime.getRuntime()
+    runtime.addShutdownHook(hook)
+    return object : ShutdownHook {
+        override fun cancel() {
+            // Allow the block to call cancel without causing IllegalStateException in the shutdown case:
+            if (Thread.currentThread() != hook) {
+                runtime.removeShutdownHook(hook)
+            }
+        }
+    }
+}

docs/source/running-the-demos.rst

@@ -107,33 +107,38 @@ To run from the command line in Windows:
 4. Run ``gradlew samples:attachment-demo:runSender`` in another terminal window to send the attachment. Now look at the other windows to
    see the output of the demo
 
-Raft Notary demo
+Notary demo
-----------------
+-----------
 
-This demo shows a party getting transactions notarised by a distributed `Raft <https://raft.github.io/>`_-based notary service.
+This demo shows a party getting transactions notarised by either a single-node or a distributed notary service.
-The demo will start three distributed notary nodes, and two counterparty nodes. One of the counterparties will generate transactions
+All versions of the demo start two counterparty nodes.
-that transfer a self-issued asset to the other party and submit them for notarisation.
+One of the counterparties will generate transactions that transfer a self-issued asset to the other party and submit them for notarisation.
+The `Raft <https://raft.github.io/>`_ version of the demo will start three distributed notary nodes.
+The `BFT SMaRt <https://bft-smart.github.io/library/>`_ version of the demo will start four distributed notary nodes.
 
 The output will display a list of notarised transaction IDs and corresponding signer public keys. In the Raft distributed notary,
 every node in the cluster can service client requests, and one signature is sufficient to satisfy the notary composite key requirement.
+In the BFT SMaRt distributed notary, three signatures are required.
 You will notice that successive transactions get signed by different members of the cluster (usually allocated in a random order).
 
-To run from the command line in Unix:
+To run the Raft version of the demo from the command line in Unix:
 
-1. Run ``./gradlew samples:raft-notary-demo:deployNodes``, which will create node directories with configs under ``samples/raft-notary-demo/build/nodes``.
+1. Run ``./gradlew samples:notary-demo:deployNodesRaft``, which will create node directories with configs under ``samples/notary-demo/build/nodes``.
-2. Run ``./samples/raft-notary-demo/build/nodes/runnodes``, which will start the nodes in separate terminal windows/tabs.
+2. Run ``./samples/notary-demo/build/nodes/runnodes``, which will start the nodes in separate terminal windows/tabs.
    Wait until a "Node started up and registered in ..." message appears on each of the terminals
-3. Run ``./gradlew samples:raft-notary-demo:notarise`` to make a call to the "Party" node to initiate notarisation requests
+3. Run ``./gradlew samples:notary-demo:notarise`` to make a call to the "Party" node to initiate notarisation requests
    In a few seconds you will see a message "Notarised 10 transactions" with a list of transaction ids and the signer public keys
 
 To run from the command line in Windows:
 
-1. Run ``gradlew samples:raft-notary-demo:deployNodes``, which will create node directories with configs under ``samples\raft-notary-demo\build\nodes``.
+1. Run ``gradlew samples:notary-demo:deployNodesRaft``, which will create node directories with configs under ``samples\notary-demo\build\nodes``.
-2. Run ``samples\raft-notary-demo\build\nodes\runnodes``, which will start the nodes in separate terminal windows/tabs.
+2. Run ``samples\notary-demo\build\nodes\runnodes``, which will start the nodes in separate terminal windows/tabs.
    Wait until a "Node started up and registered in ..." message appears on each of the terminals
-3. Run ``gradlew samples:raft-notary-demo:notarise`` to make a call to the "Party" node to initiate notarisation requests
+3. Run ``gradlew samples:notary-demo:notarise`` to make a call to the "Party" node to initiate notarisation requests
    In a few seconds you will see a message "Notarised 10 transactions" with a list of transaction ids and the signer public keys
 
+To run the BFT SMaRt notary demo, use ``deployNodesBFT`` instead of ``deployNodesRaft``. For a single notary node, use ``deployNodesSingle``.
+
 Notary nodes store consumed states in a replicated commit log, which is backed by a H2 database on each node.
 You can ascertain that the commit log is synchronised across the cluster by accessing and comparing each of the nodes' backing stores
 by using the H2 web console:

node/bft-smart-config/hosts.config

@@ -1,36 +0,0 @@
-# Copyright (c) 2007-2013 Alysson Bessani, Eduardo Alchieri, Paulo Sousa, and the authors indicated in the @author tags
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-# This file defines the replicas ids, IPs and ports.
-# It is used by the replicas and clients to find connection info
-# to the initial replicas.
-# The ports defined here are the ports used by clients to communicate
-# with the replicas. Additional connections are opened by replicas to
-# communicate with each other. This additional connection is opened in the
-# next port defined here. For an example, consider the line "0 127.0.0.1 11000".
-# That means that clients will open a communication channel to replica 0 in
-# IP 127.0.0.1 and port 11000. On startup, replicas with id different than 0
-# will open a communication channel to replica 0 in port 11001.
-# The same holds for replicas 1, 2, 3 ... N.
-
-#server id, address and port (the ids from 0 to n-1 are the service replicas)
-0 127.0.0.1 11000
-1 127.0.0.1 11010
-2 127.0.0.1 11020
-3 127.0.0.1 11030
-4 127.0.0.1 11040
-5 127.0.0.1 11050
-6 127.0.0.1 11060
-7 127.0.0.1 11070
-7001 127.0.0.1 11100

node/capsule/build.gradle

@@ -40,7 +40,6 @@ task buildCordaJAR(type: FatCapsule, dependsOn: project(':node').compileJava) {
         // javaAgents = ["quasar-core-${quasar_version}-jdk8.jar=${quasarExcludeExpression}"]
         javaAgents = ["quasar-core-${quasar_version}-jdk8.jar"]
         systemProperties['visualvm.display.name'] = 'Corda'
-        systemProperties['jdk.serialFilter'] = 'maxbytes=0'
         minJavaVersion = '1.8.0'
         minUpdateVersion['1.8'] = java8_minUpdateVersion
         caplets = ['CordaCaplet']

node/src/integration-test/kotlin/net/corda/node/BootTests.kt

@@ -53,7 +53,6 @@ class BootTests {
 class ObjectInputStreamFlow : FlowLogic<Unit>() {
     @Suspendable
     override fun call() {
-        System.clearProperty("jdk.serialFilter") // This checks that the node has already consumed the property.
         val data = ByteArrayOutputStream().apply { ObjectOutputStream(this).use { it.writeObject(object : Serializable {}) } }.toByteArray()
         ObjectInputStream(data.inputStream()).use { it.readObject() }
     }

node/src/integration-test/kotlin/net/corda/node/services/BFTNotaryServiceTests.kt

@@ -5,19 +5,19 @@ import net.corda.core.contracts.StateAndRef
 import net.corda.core.contracts.StateRef
 import net.corda.core.contracts.TransactionType
 import net.corda.core.crypto.appendToCommonName
-import net.corda.core.crypto.commonName
-import net.corda.core.div
 import net.corda.core.getOrThrow
 import net.corda.core.identity.Party
 import net.corda.core.node.services.ServiceInfo
 import net.corda.core.node.services.ServiceType
 import net.corda.core.utilities.ALICE
+import net.corda.core.utilities.DUMMY_NOTARY
 import net.corda.flows.NotaryError
 import net.corda.flows.NotaryException
 import net.corda.flows.NotaryFlow
 import net.corda.node.internal.AbstractNode
 import net.corda.node.internal.Node
 import net.corda.node.services.transactions.BFTNonValidatingNotaryService
+import net.corda.node.services.transactions.minCorrectReplicas
 import net.corda.node.utilities.ServiceIdentityGenerator
 import net.corda.node.utilities.transaction
 import net.corda.testing.node.NodeBasedTest
@@ -28,71 +28,55 @@ import kotlin.test.assertEquals
 import kotlin.test.assertFailsWith
 
 class BFTNotaryServiceTests : NodeBasedTest() {
-    private companion object {
-        val notaryCommonName = X500Name("CN=BFT Notary Server,O=R3,OU=corda,L=Zurich,C=CH")
-
-        fun buildNodeName(it: Int, notaryName: X500Name): X500Name {
-            return notaryName.appendToCommonName("-$it")
-        }
-    }
-
     @Test
     fun `detect double spend`() {
-        val masterNode = startBFTNotaryCluster(notaryCommonName, 4, BFTNonValidatingNotaryService.type).first()
+        val clusterName = X500Name("CN=BFT,O=R3,OU=corda,L=Zurich,C=CH")
+        startBFTNotaryCluster(clusterName, 4, BFTNonValidatingNotaryService.type)
         val alice = startNode(ALICE.name).getOrThrow()
-
+        val notaryParty = alice.netMapCache.getNotary(clusterName)!!
-        val notaryParty = alice.netMapCache.getNotary(notaryCommonName)!!
-
         val inputState = issueState(alice, notaryParty)
-
         val firstTxBuilder = TransactionType.General.Builder(notaryParty).withItems(inputState)
         val firstSpendTx = alice.services.signInitialTransaction(firstTxBuilder)
-
+        alice.services.startFlow(NotaryFlow.Client(firstSpendTx)).resultFuture.getOrThrow()
-        val firstSpend = alice.services.startFlow(NotaryFlow.Client(firstSpendTx))
+        val secondSpendBuilder = TransactionType.General.Builder(notaryParty).withItems(inputState).also {
-        firstSpend.resultFuture.getOrThrow()
+            it.addOutputState(DummyContract.SingleOwnerState(0, alice.info.legalIdentity))
-
-        val secondSpendBuilder = TransactionType.General.Builder(notaryParty).withItems(inputState).run {
-            val dummyState = DummyContract.SingleOwnerState(0, alice.info.legalIdentity)
-            addOutputState(dummyState)
-            this
         }
         val secondSpendTx = alice.services.signInitialTransaction(secondSpendBuilder)
         val secondSpend = alice.services.startFlow(NotaryFlow.Client(secondSpendTx))
-
+        val ex = assertFailsWith(NotaryException::class) {
-        val ex = assertFailsWith(NotaryException::class) { secondSpend.resultFuture.getOrThrow() }
+            secondSpend.resultFuture.getOrThrow()
+        }
         val error = ex.error as NotaryError.Conflict
         assertEquals(error.txId, secondSpendTx.id)
     }
 
-    private fun issueState(node: AbstractNode, notary: Party): StateAndRef<*> {
+    private fun issueState(node: AbstractNode, notary: Party) = node.run {
-        return node.database.transaction {
+        database.transaction {
-            val builder = DummyContract.generateInitial(Random().nextInt(), notary, node.info.legalIdentity.ref(0))
+            val builder = DummyContract.generateInitial(Random().nextInt(), notary, info.legalIdentity.ref(0))
-            val stx = node.services.signInitialTransaction(builder)
+            val stx = services.signInitialTransaction(builder)
-            node.services.recordTransactions(listOf(stx))
+            services.recordTransactions(listOf(stx))
             StateAndRef(builder.outputStates().first(), StateRef(stx.id, 0))
         }
     }
 
-    private fun startBFTNotaryCluster(notaryName: X500Name,
+    private fun startBFTNotaryCluster(clusterName: X500Name,
                                       clusterSize: Int,
-                                      serviceType: ServiceType): List<Node> {
+                                      serviceType: ServiceType) {
         require(clusterSize > 0)
-        val quorum = (2 * clusterSize + 1) / 3
+        val replicaNames = (0 until clusterSize).map { DUMMY_NOTARY.name.appendToCommonName(" $it") }
         ServiceIdentityGenerator.generateToDisk(
-                (0 until clusterSize).map { tempFolder.root.toPath() / "${notaryName.commonName}-$it" },
+                replicaNames.map { baseDirectory(it) },
                 serviceType.id,
-                notaryName,
+                clusterName,
-                quorum)
+                minCorrectReplicas(clusterSize))
-
+        val serviceInfo = ServiceInfo(serviceType, clusterName)
-        val serviceInfo = ServiceInfo(serviceType, notaryName)
+        val notaryClusterAddresses = (0 until clusterSize).map { "localhost:${11000 + it * 10}" }
-        val nodes = (0 until clusterSize).map {
+        (0 until clusterSize).forEach {
             startNode(
-                    buildNodeName(it, notaryName),
+                    replicaNames[it],
                     advertisedServices = setOf(serviceInfo),
-                    configOverrides = mapOf("notaryNodeId" to it)
+                    configOverrides = mapOf("bftReplicaId" to it, "notaryClusterAddresses" to notaryClusterAddresses)
             ).getOrThrow()
         }
-
-        return nodes
     }
 }

node/src/integration-test/kotlin/net/corda/services/messaging/P2PMessagingTest.kt

@@ -3,8 +3,6 @@ package net.corda.services.messaging
 import com.google.common.util.concurrent.Futures
 import com.google.common.util.concurrent.ListenableFuture
 import net.corda.core.*
-import net.corda.core.crypto.X509Utilities
-import net.corda.core.crypto.commonName
 import net.corda.core.messaging.MessageRecipients
 import net.corda.core.messaging.SingleMessageRecipient
 import net.corda.core.node.services.DEFAULT_SESSION_ID
@@ -64,10 +62,8 @@ class P2PMessagingTest : NodeBasedTest() {
     // TODO Use a dummy distributed service
     @Test
     fun `communicating with a distributed service which the network map node is part of`() {
-
-        val root = tempFolder.root.toPath()
         ServiceIdentityGenerator.generateToDisk(
-                listOf(root / DUMMY_MAP.name.commonName, root / SERVICE_2_NAME.commonName),
+                listOf(DUMMY_MAP.name, SERVICE_2_NAME).map { baseDirectory(it) },
                 RaftValidatingNotaryService.type.id,
                 DISTRIBUTED_SERVICE_NAME)
 

node/src/integration-test/kotlin/net/corda/services/messaging/P2PSecurityTest.kt

@@ -2,8 +2,6 @@ package net.corda.services.messaging
 
 import com.google.common.util.concurrent.ListenableFuture
 import net.corda.core.crypto.X509Utilities
-import net.corda.core.crypto.commonName
-import net.corda.core.div
 import net.corda.core.getOrThrow
 import net.corda.core.identity.Party
 import net.corda.core.node.NodeInfo
@@ -60,7 +58,7 @@ class P2PSecurityTest : NodeBasedTest() {
 
     private fun startSimpleNode(legalName: X500Name): SimpleNode {
         val config = TestNodeConfiguration(
-                baseDirectory = tempFolder.root.toPath() / legalName.commonName,
+                baseDirectory = baseDirectory(legalName),
                 myLegalName = legalName,
                 networkMapService = NetworkMapInfo(networkMapNode.configuration.p2pAddress, networkMapNode.info.legalIdentity.name))
         config.configureWithDevSSLCertificate() // This creates the node's TLS cert with the CN as the legal name

node/src/main/kotlin/net/corda/node/Corda.kt

@@ -10,10 +10,10 @@ import net.corda.core.crypto.commonName
 import net.corda.core.crypto.orgName
 import net.corda.core.node.VersionInfo
 import net.corda.core.utilities.Emoji
-import net.corda.core.utilities.LogHelper.withLevel
 import net.corda.node.internal.Node
 import net.corda.node.internal.enforceSingleNodeIsRunning
 import net.corda.node.services.config.FullNodeConfiguration
+import net.corda.node.services.transactions.bftSMaRtSerialFilter
 import net.corda.node.shell.InteractiveShell
 import net.corda.node.utilities.registration.HTTPNetworkRegistrationService
 import net.corda.node.utilities.registration.NetworkRegistrationHelper
@@ -21,7 +21,6 @@ import org.fusesource.jansi.Ansi
 import org.fusesource.jansi.AnsiConsole
 import org.slf4j.LoggerFactory
 import org.slf4j.bridge.SLF4JBridgeHandler
-import java.io.*
 import java.lang.management.ManagementFactory
 import java.net.InetAddress
 import java.nio.file.Paths
@@ -72,8 +71,6 @@ fun main(args: Array<String>) {
     enforceSingleNodeIsRunning(cmdlineOptions.baseDirectory)
 
     initLogging(cmdlineOptions)
-    disableJavaDeserialization() // Should be after initLogging to avoid TMI.
-
     // Manifest properties are only available if running from the corda jar
     fun manifestValue(name: String): String? = if (Manifests.exists(name)) Manifests.read(name) else null
 
@@ -107,7 +104,7 @@ fun main(args: Array<String>) {
         println("Unable to load the configuration file: ${e.rootCause.message}")
         exitProcess(2)
     }
-
+    SerialFilter.install(if (conf.bftReplicaId != null) ::bftSMaRtSerialFilter else ::defaultSerialFilter)
     if (cmdlineOptions.isRegistration) {
         println()
         println("******************************************************************")
@@ -208,29 +205,12 @@ private fun assertCanNormalizeEmptyPath() {
     }
 }
 
-private fun failStartUp(message: String): Nothing {
+internal fun failStartUp(message: String): Nothing {
     println(message)
     println("Corda will now exit...")
     exitProcess(1)
 }
 
-private fun disableJavaDeserialization() {
-    // ObjectInputFilter and friends are in java.io in Java 9 but sun.misc in backports, so we are using the system property interface for portability.
-    // This property has already been set in the Capsule. Anywhere else may be too late, but we'll repeat it here for developers.
-    System.setProperty("jdk.serialFilter", "maxbytes=0")
-    // Attempt at deserialization so that ObjectInputFilter (permanently) inits itself:
-    val data = ByteArrayOutputStream().apply { ObjectOutputStream(this).use { it.writeObject(object : Serializable {}) } }.toByteArray()
-    try {
-        withLevel("java.io.serialization", "WARN") {
-            ObjectInputStream(data.inputStream()).use { it.readObject() } // Logs REJECTED at INFO, which we don't want users to see.
-        }
-        // JDK 8u121 is the earliest JDK8 JVM that supports this functionality.
-        failStartUp("Corda forbids Java deserialisation. Please upgrade to at least JDK 8u121 and set system property 'jdk.serialFilter' to 'maxbytes=0' when booting Corda.")
-    } catch (e: InvalidClassException) {
-        // Good, our system property is honoured.
-    }
-}
-
 private fun printPluginsAndServices(node: Node) {
     node.configuration.extraAdvertisedServiceIds.let {
         if (it.isNotEmpty()) printBasicNodeInfo("Providing network services", it.joinToString())

node/src/main/kotlin/net/corda/node/SerialFilter.kt

@@ -0,0 +1,62 @@
+package net.corda.node
+
+import java.lang.reflect.Field
+import java.lang.reflect.Method
+import java.lang.reflect.Proxy
+
+internal object SerialFilter {
+    private val filterInterface: Class<*>
+    private val serialClassGetter: Method
+    private val undecided: Any
+    private val rejected: Any
+    private val serialFilterLock: Any
+    private val serialFilterField: Field
+
+    init {
+        // ObjectInputFilter and friends are in java.io in Java 9 but sun.misc in backports:
+        fun getFilterInterface(packageName: String): Class<*>? {
+            return try {
+                Class.forName("$packageName.ObjectInputFilter")
+            } catch (e: ClassNotFoundException) {
+                null
+            }
+        }
+        // JDK 8u121 is the earliest JDK8 JVM that supports this functionality.
+        filterInterface = getFilterInterface("java.io")
+                ?: getFilterInterface("sun.misc")
+                ?: failStartUp("Corda forbids Java deserialisation. Please upgrade to at least JDK 8u121.")
+        serialClassGetter = Class.forName("${filterInterface.name}\$FilterInfo").getMethod("serialClass")
+        val statusEnum = Class.forName("${filterInterface.name}\$Status")
+        undecided = statusEnum.getField("UNDECIDED").get(null)
+        rejected = statusEnum.getField("REJECTED").get(null)
+        val configClass = Class.forName("${filterInterface.name}\$Config")
+        serialFilterLock = configClass.getDeclaredField("serialFilterLock").also { it.isAccessible = true }.get(null)
+        serialFilterField = configClass.getDeclaredField("serialFilter").also { it.isAccessible = true }
+    }
+
+    internal fun install(acceptClass: (Class<*>) -> Boolean) {
+        val filter = Proxy.newProxyInstance(javaClass.classLoader, arrayOf(filterInterface)) { _, _, args ->
+            val serialClass = serialClassGetter.invoke(args[0]) as Class<*>?
+            if (applyPredicate(acceptClass, serialClass)) {
+                undecided
+            } else {
+                rejected
+            }
+        }
+        // Can't simply use the setter as in non-trampoline mode Capsule has inited the filter in premain:
+        synchronized(serialFilterLock) {
+            serialFilterField.set(null, filter)
+        }
+    }
+
+    internal fun applyPredicate(acceptClass: (Class<*>) -> Boolean, serialClass: Class<*>?): Boolean {
+        // Similar logic to jdk.serialFilter, our concern is side-effects at deserialisation time:
+        if (null == serialClass) return true
+        var componentType: Class<*> = serialClass
+        while (componentType.isArray) componentType = componentType.componentType
+        if (componentType.isPrimitive) return true
+        return acceptClass(componentType)
+    }
+}
+
+internal fun defaultSerialFilter(@Suppress("UNUSED_PARAMETER") clazz: Class<*>) = false

node/src/main/kotlin/net/corda/node/driver/Driver.kt

@@ -32,6 +32,8 @@ import net.corda.nodeapi.config.SSLConfiguration
 import net.corda.nodeapi.config.parseAs
 import net.corda.cordform.CordformNode
 import net.corda.cordform.CordformContext
+import net.corda.core.internal.ShutdownHook
+import net.corda.core.internal.addShutdownHook
 import okhttp3.OkHttpClient
 import okhttp3.Request
 import org.bouncycastle.asn1.x500.X500Name
@@ -236,22 +238,19 @@ fun <DI : DriverDSLExposedInterface, D : DriverDSLInternalInterface, A> genericD
         coerce: (D) -> DI,
         dsl: DI.() -> A
 ): A {
-    var shutdownHook: Thread? = null
+    var shutdownHook: ShutdownHook? = null
     try {
         driverDsl.start()
-        shutdownHook = Thread({
+        shutdownHook = addShutdownHook {
             driverDsl.shutdown()
-        })
+        }
-        Runtime.getRuntime().addShutdownHook(shutdownHook)
         return dsl(coerce(driverDsl))
     } catch (exception: Throwable) {
         log.error("Driver shutting down because of exception", exception)
         throw exception
     } finally {
         driverDsl.shutdown()
-        if (shutdownHook != null) {
+        shutdownHook?.cancel()
-            Runtime.getRuntime().removeShutdownHook(shutdownHook)
-        }
     }
 }
 
@@ -558,21 +557,19 @@ class DriverDSL(
             verifierType: VerifierType,
             rpcUsers: List<User>
     ): ListenableFuture<Pair<Party, List<NodeHandle>>> {
-        val nodeNames = (1..clusterSize).map { DUMMY_NOTARY.name.appendToCommonName(it.toString()) }
+        val nodeNames = (0 until clusterSize).map { DUMMY_NOTARY.name.appendToCommonName(" $it") }
         val paths = nodeNames.map { baseDirectory(it) }
         ServiceIdentityGenerator.generateToDisk(paths, type.id, notaryName)
-
+        val advertisedServices = setOf(ServiceInfo(type, notaryName))
-        val serviceInfo = ServiceInfo(type, notaryName)
-        val advertisedService = setOf(serviceInfo)
         val notaryClusterAddress = portAllocation.nextHostAndPort()
 
         // Start the first node that will bootstrap the cluster
-        val firstNotaryFuture = startNode(nodeNames.first(), advertisedService, rpcUsers, verifierType, mapOf("notaryNodeAddress" to notaryClusterAddress.toString()))
+        val firstNotaryFuture = startNode(nodeNames.first(), advertisedServices, rpcUsers, verifierType, mapOf("notaryNodeAddress" to notaryClusterAddress.toString()))
         // All other nodes will join the cluster
         val restNotaryFutures = nodeNames.drop(1).map {
             val nodeAddress = portAllocation.nextHostAndPort()
             val configOverride = mapOf("notaryNodeAddress" to nodeAddress.toString(), "notaryClusterAddresses" to listOf(notaryClusterAddress.toString()))
-            startNode(it, advertisedService, rpcUsers, verifierType, configOverride)
+            startNode(it, advertisedServices, rpcUsers, verifierType, configOverride)
         }
 
         return firstNotaryFuture.flatMap { firstNotary ->

node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt

@@ -61,6 +61,7 @@ import org.jetbrains.exposed.sql.Database
 import org.slf4j.Logger
 import java.io.IOException
 import java.lang.reflect.Modifier.*
+import java.net.InetAddress
 import java.net.URL
 import java.nio.file.FileAlreadyExistsException
 import java.nio.file.Path
@@ -518,10 +519,11 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
             RaftNonValidatingNotaryService.type -> RaftNonValidatingNotaryService(timestampChecker, uniquenessProvider as RaftUniquenessProvider)
             RaftValidatingNotaryService.type -> RaftValidatingNotaryService(timestampChecker, uniquenessProvider as RaftUniquenessProvider)
             BFTNonValidatingNotaryService.type -> with(configuration as FullNodeConfiguration) {
-                val nodeId = notaryNodeId ?: throw IllegalArgumentException("notaryNodeId value must be specified in the configuration")
+                val replicaId = bftReplicaId ?: throw IllegalArgumentException("bftReplicaId value must be specified in the configuration")
-                val client = BFTSMaRt.Client(nodeId)
+                BFTSMaRtConfig(notaryClusterAddresses).use { config ->
-                tokenizableServices += client
+                    val client = BFTSMaRt.Client(config, replicaId).also { tokenizableServices += it } // (Ab)use replicaId for clientId.
-                BFTNonValidatingNotaryService(services, timestampChecker, nodeId, database, client)
+                    BFTNonValidatingNotaryService(config, services, timestampChecker, replicaId, database, client)
+                }
             }
             else -> {
                 throw IllegalArgumentException("Notary type ${type.id} is not handled by makeNotaryService.")

node/src/main/kotlin/net/corda/node/internal/EnforceSingleNodeIsRunning.kt

@@ -1,7 +1,7 @@
 package net.corda.node.internal
 
+import net.corda.core.internal.addShutdownHook
 import net.corda.core.div
-import net.corda.core.utilities.loggerFor
 import java.io.RandomAccessFile
 import java.lang.management.ManagementFactory
 import java.nio.file.Path
@@ -26,9 +26,9 @@ fun enforceSingleNodeIsRunning(baseDirectory: Path) {
     }
     // Avoid the lock being garbage collected. We don't really need to release it as the OS will do so for us
     // when our process shuts down, but we try in stop() anyway just to be nice.
-    Runtime.getRuntime().addShutdownHook(Thread {
+    addShutdownHook {
         pidFileLock.release()
-    })
+    }
     val ourProcessID: String = ManagementFactory.getRuntimeMXBean().name.split("@")[0]
     pidFileRw.setLength(0)
     pidFileRw.write(ourProcessID.toByteArray())

node/src/main/kotlin/net/corda/node/internal/Node.kt

@@ -5,16 +5,15 @@ import com.google.common.net.HostAndPort
 import com.google.common.util.concurrent.Futures
 import com.google.common.util.concurrent.ListenableFuture
 import com.google.common.util.concurrent.SettableFuture
-import net.corda.core.flatMap
+import net.corda.core.*
+import net.corda.core.internal.ShutdownHook
+import net.corda.core.internal.addShutdownHook
 import net.corda.core.messaging.RPCOps
-import net.corda.core.minutes
 import net.corda.core.node.ServiceHub
 import net.corda.core.node.VersionInfo
 import net.corda.core.node.services.ServiceInfo
 import net.corda.core.node.services.ServiceType
 import net.corda.core.node.services.UniquenessProvider
-import net.corda.core.seconds
-import net.corda.core.success
 import net.corda.core.utilities.loggerFor
 import net.corda.core.utilities.trace
 import net.corda.node.printBasicNodeInfo
@@ -47,7 +46,6 @@ import java.io.IOException
 import java.time.Clock
 import java.util.*
 import javax.management.ObjectName
-import kotlin.concurrent.thread
 
 /**
  * A Node manages a standalone server that takes part in the P2P network. It creates the services found in [ServiceHub],
@@ -112,7 +110,7 @@ class Node(override val configuration: FullNodeConfiguration,
 
     var messageBroker: ArtemisMessagingServer? = null
 
-    private var shutdownThread: Thread? = null
+    private var shutdownHook: ShutdownHook? = null
 
     private lateinit var userService: RPCUserService
 
@@ -295,12 +293,9 @@ class Node(override val configuration: FullNodeConfiguration,
 
             (startupComplete as SettableFuture<Unit>).set(Unit)
         }
-
+        shutdownHook = addShutdownHook {
-        shutdownThread = thread(start = false) {
             stop()
         }
-        Runtime.getRuntime().addShutdownHook(shutdownThread)
-
         return this
     }
 
@@ -322,12 +317,9 @@ class Node(override val configuration: FullNodeConfiguration,
         synchronized(this) {
             if (shutdown) return
             shutdown = true
-
             // Unregister shutdown hook to prevent any unnecessary second calls to stop
-            if ((shutdownThread != null) && (Thread.currentThread() != shutdownThread)) {
+            shutdownHook?.cancel()
-                Runtime.getRuntime().removeShutdownHook(shutdownThread)
+            shutdownHook = null
-                shutdownThread = null
-            }
         }
         printBasicNodeInfo("Shutting down ...")
 

node/src/main/kotlin/net/corda/node/services/config/NodeConfiguration.kt

@@ -62,7 +62,7 @@ data class FullNodeConfiguration(
         // Instead this should be a Boolean indicating whether that broker is an internal one started by the node or an external one
         val messagingServerAddress: HostAndPort?,
         val extraAdvertisedServiceIds: List<String>,
-        val notaryNodeId: Int?,
+        val bftReplicaId: Int?,
         val notaryNodeAddress: HostAndPort?,
         val notaryClusterAddresses: List<HostAndPort>,
         override val certificateChainCheckPolicies: List<CertChainPolicyConfig>,

node/src/main/kotlin/net/corda/node/services/transactions/BFTNonValidatingNotaryService.kt

@@ -14,6 +14,7 @@ import net.corda.core.utilities.unwrap
 import net.corda.flows.NotaryException
 import net.corda.node.services.api.ServiceHubInternal
 import org.jetbrains.exposed.sql.Database
+import java.nio.file.Path
 import kotlin.concurrent.thread
 
 /**
@@ -21,14 +22,18 @@ import kotlin.concurrent.thread
  *
  * A transaction is notarised when the consensus is reached by the cluster on its uniqueness, and timestamp validity.
  */
-class BFTNonValidatingNotaryService(services: ServiceHubInternal,
+class BFTNonValidatingNotaryService(config: BFTSMaRtConfig,
+                                    services: ServiceHubInternal,
                                     timestampChecker: TimestampChecker,
                                     serverId: Int,
                                     db: Database,
-                                    val client: BFTSMaRt.Client) : NotaryService {
+                                    private val client: BFTSMaRt.Client) : NotaryService {
     init {
+        val configHandle = config.handle()
         thread(name = "BFTSmartServer-$serverId", isDaemon = true) {
-            Server(serverId, db, "bft_smart_notary_committed_states", services, timestampChecker)
+            configHandle.use {
+                Server(configHandle.path, serverId, db, "bft_smart_notary_committed_states", services, timestampChecker)
+            }
         }
     }
 
@@ -62,11 +67,12 @@ class BFTNonValidatingNotaryService(services: ServiceHubInternal,
         }
     }
 
-    private class Server(id: Int,
+    private class Server(configHome: Path,
+                         id: Int,
                          db: Database,
                          tableName: String,
                          services: ServiceHubInternal,
-                         timestampChecker: TimestampChecker) : BFTSMaRt.Server(id, db, tableName, services, timestampChecker) {
+                         timestampChecker: TimestampChecker) : BFTSMaRt.Server(configHome, id, db, tableName, services, timestampChecker) {
 
         override fun executeCommand(command: ByteArray): ByteArray {
             val request = command.deserialize<BFTSMaRt.CommitRequest>()

node/src/main/kotlin/net/corda/node/services/transactions/BFTSMaRt.kt

@@ -32,6 +32,7 @@ import net.corda.node.services.transactions.BFTSMaRt.Server
 import net.corda.node.utilities.JDBCHashMap
 import net.corda.node.utilities.transaction
 import org.jetbrains.exposed.sql.Database
+import java.nio.file.Path
 import java.util.*
 
 /**
@@ -66,13 +67,17 @@ object BFTSMaRt {
         data class Signatures(val txSignatures: List<DigitalSignature>) : ClusterResponse()
     }
 
-    class Client(val id: Int) : SingletonSerializeAsToken() {
+    class Client(config: BFTSMaRtConfig, private val clientId: Int) : SingletonSerializeAsToken() {
+        private val configHandle = config.handle()
+
         companion object {
             private val log = loggerFor<Client>()
         }
 
         /** A proxy for communicating with the BFT cluster */
-        private val proxy: ServiceProxy by lazy { buildProxy() }
+        private val proxy: ServiceProxy by lazy {
+            configHandle.use { buildProxy(it.path) }
+        }
 
         /**
          * Sends a transaction commit request to the BFT cluster. The [proxy] will deliver the request to every
@@ -86,10 +91,10 @@ object BFTSMaRt {
             return response
         }
 
-        private fun buildProxy(): ServiceProxy {
+        private fun buildProxy(configHome: Path): ServiceProxy {
             val comparator = buildResponseComparator()
             val extractor = buildExtractor()
-            return ServiceProxy(id, "bft-smart-config", comparator, extractor)
+            return ServiceProxy(clientId, configHome.toString(), comparator, extractor)
         }
 
         /** A comparator to check if replies from two replicas are the same. */
@@ -111,7 +116,7 @@ object BFTSMaRt {
                 val accepted = responses.filterIsInstance<ReplicaResponse.Signature>()
                 val rejected = responses.filterIsInstance<ReplicaResponse.Error>()
 
-                log.debug { "BFT Client $id: number of replicas accepted the commit: ${accepted.size}, rejected: ${rejected.size}" }
+                log.debug { "BFT Client $clientId: number of replicas accepted the commit: ${accepted.size}, rejected: ${rejected.size}" }
 
                 // TODO: only return an aggregate if the majority of signatures are replies
                 // TODO: return an error reported by the majority and not just the first one
@@ -137,7 +142,8 @@ object BFTSMaRt {
      * The validation logic can be specified by implementing the [executeCommand] method.
      */
     @Suppress("LeakingThis")
-    abstract class Server(val id: Int,
+    abstract class Server(configHome: Path,
+                          val replicaId: Int,
                           val db: Database,
                           tableName: String,
                           val services: ServiceHubInternal,
@@ -152,7 +158,7 @@ object BFTSMaRt {
 
         init {
             // TODO: Looks like this statement is blocking. Investigate the bft-smart node startup.
-            ServiceReplica(id, "bft-smart-config", this, this, null, DefaultReplier())
+            ServiceReplica(replicaId, configHome.toString(), this, this, null, DefaultReplier())
         }
 
         override fun appExecuteUnordered(command: ByteArray, msgCtx: MessageContext): ByteArray? {

node/src/main/kotlin/net/corda/node/services/transactions/BFTSMaRtConfig.kt

@@ -0,0 +1,61 @@
+package net.corda.node.services.transactions
+
+import com.google.common.net.HostAndPort
+import net.corda.core.div
+import java.io.FileWriter
+import java.io.PrintWriter
+import java.net.InetAddress
+import java.nio.file.Files
+
+/**
+ * BFT SMaRt can only be configured via files in a configHome directory.
+ * Each instance of this class creates such a configHome, accessible via [path].
+ * The files are deleted on [close] typically via [use], see [PathManager] for details.
+ */
+class BFTSMaRtConfig(replicaAddresses: List<HostAndPort>) : PathManager(Files.createTempDirectory("bft-smart-config")) {
+    companion object {
+        internal val portIsClaimedFormat = "Port %s is claimed by another replica: %s"
+    }
+
+    init {
+        val claimedPorts = mutableSetOf<Int>()
+        replicaAddresses.map { it.port }.forEach { base ->
+            // Each replica claims the configured port and the next one:
+            (0..1).map { base + it }.forEach { port ->
+                claimedPorts.add(port) || throw IllegalArgumentException(portIsClaimedFormat.format(port, claimedPorts))
+            }
+        }
+        configWriter("hosts.config") {
+            replicaAddresses.forEachIndexed { index, address ->
+                // The documentation strongly recommends IP addresses:
+                println("${index} ${InetAddress.getByName(address.host).hostAddress} ${address.port}")
+            }
+        }
+        val n = replicaAddresses.size
+        val systemConfig = String.format(javaClass.getResource("system.config.printf").readText(), n, maxFaultyReplicas(n))
+        configWriter("system.config") {
+            print(systemConfig)
+        }
+    }
+
+    private fun configWriter(name: String, block: PrintWriter.() -> Unit) {
+        // Default charset, consistent with loaders:
+        FileWriter((path / name).toFile()).use {
+            PrintWriter(it).use {
+                it.run(block)
+            }
+        }
+    }
+}
+
+fun maxFaultyReplicas(clusterSize: Int) = (clusterSize - 1) / 3
+fun minCorrectReplicas(clusterSize: Int) = (2 * clusterSize + 3) / 3
+fun minClusterSize(maxFaultyReplicas: Int) = maxFaultyReplicas * 3 + 1
+
+fun bftSMaRtSerialFilter(clazz: Class<*>): Boolean = clazz.name.let {
+    it.startsWith("bftsmart.")
+            || it.startsWith("java.security.")
+            || it.startsWith("java.util.")
+            || it.startsWith("java.lang.")
+            || it.startsWith("java.net.")
+}

node/src/main/kotlin/net/corda/node/services/transactions/PathManager.kt

@@ -0,0 +1,43 @@
+package net.corda.node.services.transactions
+
+import net.corda.core.internal.addShutdownHook
+import java.io.Closeable
+import java.nio.file.Path
+import java.util.concurrent.atomic.AtomicInteger
+
+internal class DeleteOnExitPath(internal val path: Path) {
+    private val shutdownHook = addShutdownHook { dispose() }
+    internal fun dispose() {
+        path.toFile().deleteRecursively()
+        shutdownHook.cancel()
+    }
+}
+
+open class PathHandle internal constructor(private val deleteOnExitPath: DeleteOnExitPath, private val handleCounter: AtomicInteger) : Closeable {
+    val path
+        get(): Path {
+            val path = deleteOnExitPath.path
+            check(handleCounter.get() != 0) { "Defunct path: $path" }
+            return path
+        }
+
+    init {
+        handleCounter.incrementAndGet()
+    }
+
+    fun handle() = PathHandle(deleteOnExitPath, handleCounter)
+
+    override fun close() {
+        if (handleCounter.decrementAndGet() == 0) {
+            deleteOnExitPath.dispose()
+        }
+    }
+}
+
+/**
+ * An instance of this class is a handle on a temporary [path].
+ * If necessary, additional handles on the same path can be created using the [handle] method.
+ * The path is (recursively) deleted when [close] is called on the last handle, typically at the end of a [use] expression.
+ * The value of eager cleanup of temporary files is that there are cases when shutdown hooks don't run e.g. SIGKILL.
+ */
+open class PathManager(path: Path) : PathHandle(DeleteOnExitPath(path), AtomicInteger())

node/src/main/resources/net/corda/node/services/transactions/system.config.printf

@@ -32,10 +32,10 @@ system.communication.defaultkeys = true
 ############################################
 
 #Number of servers in the group
-system.servers.num = 4
+system.servers.num = %s
 
 #Maximum number of faulty replicas
-system.servers.f = 1
+system.servers.f = %s
 
 #Timeout to asking for a client request
 system.totalordermulticast.timeout = 2000

node/src/test/kotlin/net/corda/node/SerialFilterTests.kt

@@ -0,0 +1,31 @@
+package net.corda.node
+
+import org.junit.Test
+import java.io.IOException
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+import kotlin.test.fail
+
+class SerialFilterTests {
+    @Test
+    fun `null and primitives are accepted and arrays are unwrapped`() {
+        val acceptClass = { _: Class<*> -> fail("Should not be invoked.") }
+        listOf(null, Byte::class.javaPrimitiveType, IntArray::class.java, Array<CharArray>::class.java).forEach {
+            assertTrue(SerialFilter.applyPredicate(acceptClass, it))
+        }
+    }
+
+    @Test
+    fun `the predicate is applied to the componentType`() {
+        val classes = mutableListOf<Class<*>>()
+        val acceptClass = { clazz: Class<*> ->
+            classes.add(clazz)
+            false
+        }
+        listOf(String::class.java, Array<Unit>::class.java, Array<Array<IOException>>::class.java).forEach {
+            assertFalse(SerialFilter.applyPredicate(acceptClass, it))
+        }
+        assertEquals(listOf<Class<*>>(String::class.java, Unit::class.java, IOException::class.java), classes)
+    }
+}

node/src/test/kotlin/net/corda/node/internal/NodeTest.kt

@@ -5,6 +5,7 @@ import net.corda.core.crypto.commonName
 import net.corda.core.div
 import net.corda.core.getOrThrow
 import net.corda.core.utilities.ALICE
+import net.corda.core.utilities.WHITESPACE
 import net.corda.testing.node.NodeBasedTest
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.Test
@@ -12,11 +13,8 @@ import org.junit.Test
 class NodeTest : NodeBasedTest() {
     @Test
     fun `empty plugins directory`() {
-        val baseDirectory = tempFolder.root.toPath() / ALICE.name.commonName
+        val baseDirectory = baseDirectory(ALICE.name)
         (baseDirectory / "plugins").createDirectories()
-        val node = startNode(ALICE.name).getOrThrow()
+        startNode(ALICE.name).getOrThrow()
-        // Make sure we created the plugins dir in the correct place
-        assertThat(baseDirectory).isEqualTo(node.configuration.baseDirectory)
-
     }
 }

node/src/test/kotlin/net/corda/node/services/transactions/BFTSMaRtConfigTests.kt

@@ -0,0 +1,40 @@
+package net.corda.node.services.transactions
+
+import com.google.common.net.HostAndPort
+import net.corda.node.services.transactions.BFTSMaRtConfig.Companion.portIsClaimedFormat
+import org.junit.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+
+class BFTSMaRtConfigTests {
+    @Test
+    fun `replica arithmetic`() {
+        (1..20).forEach { n ->
+            assertEquals(n, maxFaultyReplicas(n) + minCorrectReplicas(n))
+        }
+        (1..3).forEach { n -> assertEquals(0, maxFaultyReplicas(n)) }
+        (4..6).forEach { n -> assertEquals(1, maxFaultyReplicas(n)) }
+        (7..9).forEach { n -> assertEquals(2, maxFaultyReplicas(n)) }
+        10.let { n -> assertEquals(3, maxFaultyReplicas(n)) }
+    }
+
+    @Test
+    fun `min cluster size`() {
+        assertEquals(1, minClusterSize(0))
+        assertEquals(4, minClusterSize(1))
+        assertEquals(7, minClusterSize(2))
+        assertEquals(10, minClusterSize(3))
+    }
+
+    @Test
+    fun `overlapping port ranges are rejected`() {
+        fun addresses(vararg ports: Int) = ports.map { HostAndPort.fromParts("localhost", it) }
+        assertFailsWith(IllegalArgumentException::class, portIsClaimedFormat.format(11001, setOf(11000, 11001))) {
+            BFTSMaRtConfig(addresses(11000, 11001)).use {}
+        }
+        assertFailsWith(IllegalArgumentException::class, portIsClaimedFormat.format(11001, setOf(11001, 11002))) {
+            BFTSMaRtConfig(addresses(11001, 11000)).use {}
+        }
+        BFTSMaRtConfig(addresses(11000, 11002)).use {} // Non-overlapping.
+    }
+}

node/src/test/kotlin/net/corda/node/services/transactions/PathManagerTests.kt

@@ -0,0 +1,32 @@
+package net.corda.node.services.transactions
+
+import net.corda.core.exists
+import org.junit.Test
+import java.nio.file.Files
+import kotlin.test.assertFailsWith
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+class PathManagerTests {
+    @Test
+    fun `path deleted when manager closed`() {
+        val manager = PathManager(Files.createTempFile(javaClass.simpleName, null))
+        val leakedPath = manager.use {
+            it.path.also { assertTrue(it.exists()) }
+        }
+        assertFalse(leakedPath.exists())
+        assertFailsWith(IllegalStateException::class) { manager.path }
+    }
+
+    @Test
+    fun `path deleted when handle closed`() {
+        val handle = PathManager(Files.createTempFile(javaClass.simpleName, null)).use {
+            it.handle()
+        }
+        val leakedPath = handle.use {
+            it.path.also { assertTrue(it.exists()) }
+        }
+        assertFalse(leakedPath.exists())
+        assertFailsWith(IllegalStateException::class) { handle.path }
+    }
+}

samples/README.md

@@ -7,5 +7,5 @@ Please refer to `README.md` in the individual project folders.  There are the fo
 * **trader-demo** A simple driver for exercising the two party trading flow. In this scenario, a buyer wants to purchase some commercial paper by swapping his cash for commercial paper. The seller learns that the buyer exists, and sends them a message to kick off the trade. The seller, having obtained his CP, then quits and the buyer goes back to waiting. The buyer will sell as much CP as he can! **We recommend starting with this demo.**
 * **Network-visualiser** A tool that uses a simulation to visualise the interaction and messages between nodes on the Corda network. Currently only works for the IRS demo.
 * **simm-valudation-demo** A demo showing two nodes reaching agreement on the valuation of a derivatives portfolio.
-* **raft-notary-demo** A simple demonstration of a node getting multiple transactions notarised by a distributed (Raft-based) notary.
+* **notary-demo** A simple demonstration of a node getting multiple transactions notarised by a single or distributed (Raft or BFT SMaRt) notary.
 * **bank-of-corda-demo** A demo showing a node acting as an issuer of fungible assets (initially Cash)

samples/notary-demo/README.md

@@ -1,5 +1,5 @@
-# Distributed Notary (Raft) Demo 
+# Distributed Notary Demo
 
-This program is a simple demonstration of a node getting multiple transactions notarised by a distributed (Raft-based) notary.
+This program is a simple demonstration of a node getting multiple transactions notarised by a distributed (Raft or BFT SMaRt) notary.
 
 Please see docs/build/html/running-the-demos.html to learn how to use this demo.

samples/notary-demo/build.gradle

@@ -41,7 +41,7 @@ publishing {
     publications {
         jarAndSources(MavenPublication) {
             from components.java
-            artifactId 'raftnotarydemo'
+            artifactId 'notarydemo'
 
             artifact sourceJar
             artifact javadocJar
@@ -57,6 +57,10 @@ task deployNodesRaft(type: Cordform, dependsOn: 'jar') {
     definitionClass = 'net.corda.notarydemo.RaftNotaryCordform'
 }
 
+task deployNodesBFT(type: Cordform, dependsOn: 'jar') {
+    definitionClass = 'net.corda.notarydemo.BFTNotaryCordform'
+}
+
 task notarise(type: JavaExec) {
     classpath = sourceSets.main.runtimeClasspath
     main = 'net.corda.notarydemo.NotariseKt'

samples/notary-demo/src/main/kotlin/net/corda/demorun/DemoRunner.kt

@@ -6,8 +6,6 @@ import net.corda.node.driver.driver
 import net.corda.cordform.CordformDefinition
 import net.corda.cordform.CordformNode
 
-fun CordformDefinition.node(configure: CordformNode.() -> Unit) = addNode { cordformNode -> cordformNode.configure() }
-
 fun CordformDefinition.clean() {
     System.err.println("Deleting: $driverDirectory")
     driverDirectory.toFile().deleteRecursively()

samples/notary-demo/src/main/kotlin/net/corda/demorun/util/DemoUtils.kt

@@ -0,0 +1,26 @@
+package net.corda.demorun.util
+
+import com.google.common.net.HostAndPort
+import net.corda.cordform.CordformDefinition
+import net.corda.cordform.CordformNode
+import net.corda.core.node.services.ServiceInfo
+import net.corda.nodeapi.User
+import org.bouncycastle.asn1.x500.X500Name
+
+fun CordformDefinition.node(configure: CordformNode.() -> Unit) {
+    addNode { cordformNode -> cordformNode.configure() }
+}
+
+fun CordformNode.name(name: X500Name) = name(name.toString())
+
+fun CordformNode.rpcUsers(vararg users: User) {
+    rpcUsers = users.map { it.toMap() }
+}
+
+fun CordformNode.advertisedServices(vararg services: ServiceInfo) {
+    advertisedServices = services.map { it.toString() }
+}
+
+fun CordformNode.notaryClusterAddresses(vararg addresses: HostAndPort) {
+    notaryClusterAddresses = addresses.map { it.toString() }
+}

samples/notary-demo/src/main/kotlin/net/corda/notarydemo/BFTNotaryCordform.kt

@@ -0,0 +1,69 @@
+package net.corda.notarydemo
+
+import com.google.common.net.HostAndPort
+import net.corda.core.div
+import net.corda.core.node.services.ServiceInfo
+import net.corda.core.utilities.ALICE
+import net.corda.core.utilities.BOB
+import net.corda.demorun.util.*
+import net.corda.demorun.runNodes
+import net.corda.node.services.transactions.BFTNonValidatingNotaryService
+import net.corda.node.utilities.ServiceIdentityGenerator
+import net.corda.cordform.CordformDefinition
+import net.corda.cordform.CordformContext
+import net.corda.cordform.CordformNode
+import net.corda.core.mapToArray
+import net.corda.node.services.transactions.minCorrectReplicas
+import org.bouncycastle.asn1.x500.X500Name
+
+fun main(args: Array<String>) = BFTNotaryCordform.runNodes()
+
+private val clusterSize = 4 // Minimum size thats tolerates a faulty replica.
+private val notaryNames = createNotaryNames(clusterSize)
+
+object BFTNotaryCordform : CordformDefinition("build" / "notary-demo-nodes", notaryNames[0]) {
+    private val clusterName = X500Name("CN=BFT,O=R3,OU=corda,L=Zurich,C=CH")
+    private val advertisedService = ServiceInfo(BFTNonValidatingNotaryService.type, clusterName)
+
+    init {
+        node {
+            name(ALICE.name)
+            p2pPort(10002)
+            rpcPort(10003)
+            rpcUsers(notaryDemoUser)
+        }
+        node {
+            name(BOB.name)
+            p2pPort(10005)
+            rpcPort(10006)
+        }
+        val clusterAddresses = (0 until clusterSize).mapToArray { HostAndPort.fromParts("localhost", 11000 + it * 10) }
+        fun notaryNode(replicaId: Int, configure: CordformNode.() -> Unit) = node {
+            name(notaryNames[replicaId])
+            advertisedServices(advertisedService)
+            notaryClusterAddresses(*clusterAddresses)
+            bftReplicaId(replicaId)
+            configure()
+        }
+        notaryNode(0) {
+            p2pPort(10009)
+            rpcPort(10010)
+        }
+        notaryNode(1) {
+            p2pPort(10013)
+            rpcPort(10014)
+        }
+        notaryNode(2) {
+            p2pPort(10017)
+            rpcPort(10018)
+        }
+        notaryNode(3) {
+            p2pPort(10021)
+            rpcPort(10022)
+        }
+    }
+
+    override fun setup(context: CordformContext) {
+        ServiceIdentityGenerator.generateToDisk(notaryNames.map { context.baseDirectory(it) }, advertisedService.type.id, clusterName, minCorrectReplicas(clusterSize))
+    }
+}

samples/notary-demo/src/main/kotlin/net/corda/notarydemo/Clean.kt

@@ -3,7 +3,7 @@ package net.corda.notarydemo
 import net.corda.demorun.clean
 
 fun main(args: Array<String>) {
-    listOf(SingleNotaryCordform, RaftNotaryCordform).forEach {
+    listOf(SingleNotaryCordform, RaftNotaryCordform, BFTNotaryCordform).forEach {
         it.clean()
     }
 }

samples/notary-demo/src/main/kotlin/net/corda/notarydemo/Notarise.kt

@@ -2,10 +2,12 @@ package net.corda.notarydemo
 
 import com.google.common.net.HostAndPort
 import com.google.common.util.concurrent.Futures
+import com.google.common.util.concurrent.ListenableFuture
 import net.corda.client.rpc.CordaRPCClient
 import net.corda.client.rpc.notUsed
 import net.corda.core.crypto.toStringShort
 import net.corda.core.getOrThrow
+import net.corda.core.map
 import net.corda.core.messaging.CordaRPCOps
 import net.corda.core.messaging.startFlow
 import net.corda.core.transactions.SignedTransaction
@@ -14,11 +16,10 @@ import net.corda.notarydemo.flows.DummyIssueAndMove
 import net.corda.notarydemo.flows.RPCStartableNotaryFlowClient
 
 fun main(args: Array<String>) {
-    val host = HostAndPort.fromString("localhost:10003")
+    val address = HostAndPort.fromParts("localhost", 10003)
-    println("Connecting to the recipient node ($host)")
+    println("Connecting to the recipient node ($address)")
-    CordaRPCClient(host).start("demo", "demo").use {
+    CordaRPCClient(address).start(notaryDemoUser.username, notaryDemoUser.password).use {
-        val api = NotaryDemoClientApi(it.proxy)
+        NotaryDemoClientApi(it.proxy).notarise(10)
-        api.startNotarisation()
     }
 }
 
@@ -27,34 +28,23 @@ private class NotaryDemoClientApi(val rpc: CordaRPCOps) {
     private val notary by lazy {
         val (parties, partyUpdates) = rpc.networkMapUpdates()
         partyUpdates.notUsed()
-        parties.first { it.advertisedServices.any { it.info.type.isNotary() } }.notaryIdentity
+        parties.filter { it.advertisedServices.any { it.info.type.isNotary() } }.map { it.notaryIdentity }.distinct().single()
     }
 
     private val counterpartyNode by lazy {
         val (parties, partyUpdates) = rpc.networkMapUpdates()
         partyUpdates.notUsed()
-        parties.first { it.legalIdentity.name == BOB.name }
+        parties.single { it.legalIdentity.name == BOB.name }
-    }
-
-    private companion object {
-        private val TRANSACTION_COUNT = 10
     }
 
     /** Makes calls to the node rpc to start transaction notarisation. */
-    fun startNotarisation() {
-        notarise(TRANSACTION_COUNT)
-    }
-
     fun notarise(count: Int) {
+        println("Notary: \"${notary.name}\", with composite key: ${notary.owningKey.toStringShort()}")
         val transactions = buildTransactions(count)
-        val signers = notariseTransactions(transactions)
+        println("Notarised ${transactions.size} transactions:")
-        val transactionSigners = transactions.zip(signers).map {
+        transactions.zip(notariseTransactions(transactions)).forEach { (tx, signersFuture) ->
-            val (tx, signer) = it
+            println("Tx [${tx.tx.id.prefixChars()}..] signed by ${signersFuture.getOrThrow().joinToString()}")
-            "Tx [${tx.tx.id.prefixChars()}..] signed by $signer"
+        }
-        }.joinToString("\n")
-
-        println("Notary: \"${notary.name}\", with composite key: ${notary.owningKey.toStringShort()}\n" +
-                "Notarised ${transactions.size} transactions:\n" + transactionSigners)
     }
 
     /**
@@ -63,10 +53,9 @@ private class NotaryDemoClientApi(val rpc: CordaRPCOps) {
      * as it consumes the original asset and creates a copy with the new owner as its output.
      */
     private fun buildTransactions(count: Int): List<SignedTransaction> {
-        val moveTransactions = (1..count).map {
+        return Futures.allAsList((1..count).map {
             rpc.startFlow(::DummyIssueAndMove, notary, counterpartyNode.legalIdentity).returnValue
-        }
+        }).getOrThrow()
-        return Futures.allAsList(moveTransactions).getOrThrow()
     }
 
     /**
@@ -75,10 +64,9 @@ private class NotaryDemoClientApi(val rpc: CordaRPCOps) {
      *
      * @return a list of encoded signer public keys - one for every transaction
      */
-    private fun notariseTransactions(transactions: List<SignedTransaction>): List<String> {
+    private fun notariseTransactions(transactions: List<SignedTransaction>): List<ListenableFuture<List<String>>> {
-        // TODO: Remove this suppress when we upgrade to kotlin 1.1 or when JetBrain fixes the bug.
+        return transactions.map {
-        @Suppress("UNSUPPORTED_FEATURE")
+            rpc.startFlow(::RPCStartableNotaryFlowClient, it).returnValue.map { it.map { it.by.toStringShort() } }
-        val signatureFutures = transactions.map { rpc.startFlow(::RPCStartableNotaryFlowClient, it).returnValue }
+        }
-        return Futures.allAsList(signatureFutures).getOrThrow().map { it.map { it.by.toStringShort() }.joinToString() }
     }
 }

samples/notary-demo/src/main/kotlin/net/corda/notarydemo/RaftNotaryCordform.kt

@@ -0,0 +1,70 @@
+package net.corda.notarydemo
+
+import com.google.common.net.HostAndPort
+import net.corda.core.crypto.appendToCommonName
+import net.corda.core.div
+import net.corda.core.node.services.ServiceInfo
+import net.corda.core.utilities.ALICE
+import net.corda.core.utilities.BOB
+import net.corda.core.utilities.DUMMY_NOTARY
+import net.corda.demorun.util.*
+import net.corda.node.services.transactions.RaftValidatingNotaryService
+import net.corda.node.utilities.ServiceIdentityGenerator
+import net.corda.cordform.CordformDefinition
+import net.corda.cordform.CordformContext
+import net.corda.cordform.CordformNode
+import net.corda.demorun.runNodes
+import net.corda.demorun.util.node
+import org.bouncycastle.asn1.x500.X500Name
+
+fun main(args: Array<String>) = RaftNotaryCordform.runNodes()
+
+internal fun createNotaryNames(clusterSize: Int) = (0 until clusterSize).map { DUMMY_NOTARY.name.appendToCommonName(" $it") }
+
+private val notaryNames = createNotaryNames(3)
+
+object RaftNotaryCordform : CordformDefinition("build" / "notary-demo-nodes", notaryNames[0]) {
+    private val clusterName = X500Name("CN=Raft,O=R3,OU=corda,L=Zurich,C=CH")
+    private val advertisedService = ServiceInfo(RaftValidatingNotaryService.type, clusterName)
+
+    init {
+        node {
+            name(ALICE.name)
+            p2pPort(10002)
+            rpcPort(10003)
+            rpcUsers(notaryDemoUser)
+        }
+        node {
+            name(BOB.name)
+            p2pPort(10005)
+            rpcPort(10006)
+        }
+        fun notaryNode(index: Int, configure: CordformNode.() -> Unit) = node {
+            name(notaryNames[index])
+            advertisedServices(advertisedService)
+            configure()
+        }
+        notaryNode(0) {
+            notaryNodePort(10008)
+            p2pPort(10009)
+            rpcPort(10010)
+        }
+        val clusterAddress = HostAndPort.fromParts("localhost", 10008) // Otherwise each notary forms its own cluster.
+        notaryNode(1) {
+            notaryNodePort(10012)
+            p2pPort(10013)
+            rpcPort(10014)
+            notaryClusterAddresses(clusterAddress)
+        }
+        notaryNode(2) {
+            notaryNodePort(10016)
+            p2pPort(10017)
+            rpcPort(10018)
+            notaryClusterAddresses(clusterAddress)
+        }
+    }
+
+    override fun setup(context: CordformContext) {
+        ServiceIdentityGenerator.generateToDisk(notaryNames.map { context.baseDirectory(it) }, advertisedService.type.id, clusterName)
+    }
+}

samples/notary-demo/src/main/kotlin/net/corda/notarydemo/SingleNotaryCordform.kt

@@ -5,7 +5,6 @@ import net.corda.core.node.services.ServiceInfo
 import net.corda.core.utilities.ALICE
 import net.corda.core.utilities.BOB
 import net.corda.core.utilities.DUMMY_NOTARY
-import net.corda.demorun.node
 import net.corda.demorun.runNodes
 import net.corda.node.services.startFlowPermission
 import net.corda.node.services.transactions.ValidatingNotaryService
@@ -14,31 +13,30 @@ import net.corda.notarydemo.flows.DummyIssueAndMove
 import net.corda.notarydemo.flows.RPCStartableNotaryFlowClient
 import net.corda.cordform.CordformDefinition
 import net.corda.cordform.CordformContext
+import net.corda.demorun.util.*
 
 fun main(args: Array<String>) = SingleNotaryCordform.runNodes()
 
+val notaryDemoUser = User("demou", "demop", setOf(startFlowPermission<DummyIssueAndMove>(), startFlowPermission<RPCStartableNotaryFlowClient>()))
+
 object SingleNotaryCordform : CordformDefinition("build" / "notary-demo-nodes", DUMMY_NOTARY.name) {
     init {
         node {
-            name(ALICE.name.toString())
+            name(ALICE.name)
-            nearestCity("London")
             p2pPort(10002)
             rpcPort(10003)
-            rpcUsers = listOf(User("demo", "demo", setOf(startFlowPermission<DummyIssueAndMove>(), startFlowPermission<RPCStartableNotaryFlowClient>())).toMap())
+            rpcUsers(notaryDemoUser)
         }
         node {
-            name(BOB.name.toString())
+            name(BOB.name)
-            nearestCity("New York")
             p2pPort(10005)
             rpcPort(10006)
         }
         node {
-            name(DUMMY_NOTARY.name.toString())
+            name(DUMMY_NOTARY.name)
-            nearestCity("London")
-            advertisedServices = listOf(ServiceInfo(ValidatingNotaryService.type).toString())
             p2pPort(10009)
             rpcPort(10010)
-            notaryNodePort(10008)
+            advertisedServices(ServiceInfo(ValidatingNotaryService.type))
         }
     }
 

samples/raft-notary-demo/src/main/kotlin/net/corda/notarydemo/RaftNotaryCordform.kt

@@ -1,73 +0,0 @@
-package net.corda.notarydemo
-
-import net.corda.core.crypto.appendToCommonName
-import net.corda.core.div
-import net.corda.core.node.services.ServiceInfo
-import net.corda.core.utilities.ALICE
-import net.corda.core.utilities.BOB
-import net.corda.core.utilities.DUMMY_NOTARY
-import net.corda.demorun.node
-import net.corda.demorun.runNodes
-import net.corda.node.services.startFlowPermission
-import net.corda.node.services.transactions.RaftValidatingNotaryService
-import net.corda.node.utilities.ServiceIdentityGenerator
-import net.corda.nodeapi.User
-import net.corda.notarydemo.flows.DummyIssueAndMove
-import net.corda.notarydemo.flows.RPCStartableNotaryFlowClient
-import net.corda.cordform.CordformDefinition
-import net.corda.cordform.CordformContext
-import org.bouncycastle.asn1.x500.X500Name
-
-fun main(args: Array<String>) = RaftNotaryCordform.runNodes()
-
-private val notaryNames = (1..3).map { DUMMY_NOTARY.name.appendToCommonName(" $it") }
-
-object RaftNotaryCordform : CordformDefinition("build" / "notary-demo-nodes", notaryNames[0]) {
-    private val advertisedNotary = ServiceInfo(RaftValidatingNotaryService.type, X500Name("CN=Raft,O=R3,OU=corda,L=Zurich,C=CH"))
-
-    init {
-        node {
-            name(ALICE.name.toString())
-            nearestCity("London")
-            p2pPort(10002)
-            rpcPort(10003)
-            rpcUsers = listOf(User("demo", "demo", setOf(startFlowPermission<DummyIssueAndMove>(), startFlowPermission<RPCStartableNotaryFlowClient>())).toMap())
-        }
-        node {
-            name(BOB.name.toString())
-            nearestCity("New York")
-            p2pPort(10005)
-            rpcPort(10006)
-        }
-        node {
-            name(notaryNames[0].toString())
-            nearestCity("London")
-            advertisedServices = listOf(advertisedNotary.toString())
-            p2pPort(10009)
-            rpcPort(10010)
-            notaryNodePort(10008)
-        }
-        node {
-            name(notaryNames[1].toString())
-            nearestCity("London")
-            advertisedServices = listOf(advertisedNotary.toString())
-            p2pPort(10013)
-            rpcPort(10014)
-            notaryNodePort(10012)
-            notaryClusterAddresses = listOf("localhost:10008")
-        }
-        node {
-            name(notaryNames[2].toString())
-            nearestCity("London")
-            advertisedServices = listOf(advertisedNotary.toString())
-            p2pPort(10017)
-            rpcPort(10018)
-            notaryNodePort(10016)
-            notaryClusterAddresses = listOf("localhost:10008")
-        }
-    }
-
-    override fun setup(context: CordformContext) {
-        ServiceIdentityGenerator.generateToDisk(notaryNames.map { context.baseDirectory(it) }, advertisedNotary.type.id, advertisedNotary.name!!)
-    }
-}

settings.gradle

@@ -30,6 +30,6 @@ include 'samples:trader-demo'
 include 'samples:irs-demo'
 include 'samples:network-visualiser'
 include 'samples:simm-valuation-demo'
-include 'samples:raft-notary-demo'
+include 'samples:notary-demo'
 include 'samples:bank-of-corda-demo'
 include 'cordform-common'

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

@@ -189,7 +189,7 @@ fun testConfiguration(baseDirectory: Path, legalName: X500Name, basePort: Int):
             rpcAddress = HostAndPort.fromParts("localhost", basePort + 1),
             messagingServerAddress = null,
             extraAdvertisedServiceIds = emptyList(),
-            notaryNodeId = null,
+            bftReplicaId = null,
             notaryNodeAddress = null,
             notaryClusterAddresses = emptyList(),
             certificateChainCheckPolicies = emptyList(),

test-utils/src/main/kotlin/net/corda/testing/node/NodeBasedTest.kt

@@ -4,10 +4,12 @@ import com.google.common.util.concurrent.Futures
 import com.google.common.util.concurrent.ListenableFuture
 import net.corda.core.*
 import net.corda.core.crypto.X509Utilities
+import net.corda.core.crypto.appendToCommonName
 import net.corda.core.crypto.commonName
 import net.corda.core.node.services.ServiceInfo
 import net.corda.core.node.services.ServiceType
 import net.corda.core.utilities.DUMMY_MAP
+import net.corda.core.utilities.WHITESPACE
 import net.corda.node.driver.addressMustNotBeBound
 import net.corda.node.internal.Node
 import net.corda.node.services.config.ConfigHelper
@@ -107,7 +109,7 @@ abstract class NodeBasedTest {
                            clusterSize: Int,
                            serviceType: ServiceType = RaftValidatingNotaryService.type): ListenableFuture<List<Node>> {
         ServiceIdentityGenerator.generateToDisk(
-                (0 until clusterSize).map { tempFolder.root.toPath() / "${notaryName.commonName}-$it" },
+                (0 until clusterSize).map { baseDirectory(notaryName.appendToCommonName("-$it")) },
                 serviceType.id,
                 notaryName)
 
@@ -133,12 +135,14 @@ abstract class NodeBasedTest {
         }
     }
 
+    protected fun baseDirectory(legalName: X500Name) = tempFolder.root.toPath() / legalName.commonName.replace(WHITESPACE, "")
+
     private fun startNodeInternal(legalName: X500Name,
                                   platformVersion: Int,
                                   advertisedServices: Set<ServiceInfo>,
                                   rpcUsers: List<User>,
                                   configOverrides: Map<String, Any>): Node {
-        val baseDirectory = (tempFolder.root.toPath() / legalName.commonName).createDirectories()
+        val baseDirectory = baseDirectory(legalName).createDirectories()
         val localPort = getFreeLocalPorts("localhost", 2)
         val config = ConfigHelper.loadConfig(
                 baseDirectory = baseDirectory,

verifier/src/main/kotlin/net/corda/verifier/Verifier.kt

@@ -5,6 +5,7 @@ import com.typesafe.config.Config
 import com.typesafe.config.ConfigFactory
 import com.typesafe.config.ConfigParseOptions
 import net.corda.core.ErrorOr
+import net.corda.core.internal.addShutdownHook
 import net.corda.core.div
 import net.corda.core.utilities.debug
 import net.corda.core.utilities.loggerFor
@@ -51,11 +52,11 @@ class Verifier {
             val session = sessionFactory.createSession(
                     VerifierApi.VERIFIER_USERNAME, VerifierApi.VERIFIER_USERNAME, false, true, true, locator.isPreAcknowledge, locator.ackBatchSize
             )
-            Runtime.getRuntime().addShutdownHook(Thread {
+            addShutdownHook {
                 log.info("Shutting down")
                 session.close()
                 sessionFactory.close()
-            })
+            }
             val consumer = session.createConsumer(VERIFICATION_REQUESTS_QUEUE_NAME)
             val replyProducer = session.createProducer()
             consumer.setMessageHandler {