Quasar NPE when passing a Kotlin Lambda to a Corda Flow

Question

When passing a lambda to the CollectSignaturesFlow Quasar crashes silently and is unable to continue processing. The issue is with others.map { initiateFlow(it as Party) }.toSet(). Moving this outside and establishing the set as a local variable resolves the issue.

Is there a set of known issues around what can and cannot be serialized with Quasar?

@InitiatingFlow
@StartableByRPC
class IOUIssueFlow(val state: IOUState) : FlowLogic<SignedTransaction>() {
   @Suspendable
   override fun call(): SignedTransaction {
       val txCommand = Command(
               IOUContract.Commands.Issue(),
               state.participants.map { it.owningKey })
       val builder = TransactionBuilder(serviceHub.networkMapCache.notaryIdentities.first())
               .withItems(
                       StateAndContract(
                               state,
                               IOUContract.IOU_CONTRACT_ID),
                       txCommand)
       builder.verify(serviceHub)
       val others = state.participants - ourIdentity
       return subFlow(
               FinalityFlow(
                       subFlow(
                               CollectSignaturesFlow(
                                       serviceHub.signInitialTransaction(builder),
                                       others.map { initiateFlow(it as Party) }.toSet()))))
   }
}

The code will run indefinitely without end and produces this warning:

[WARN ] 16:38:13,655 [Mock network] (FlowStateMachineImpl.kt:127) flow.[49f2ebbd-be62-462d-af87-24f5ae40c7d6].run - Terminated by unexpected exception
java.lang.NullPointerException: null
    at net.corda.node.services.statemachine.FlowStateMachineImpl.suspend(FlowStateMachineImpl.kt:468) ~[corda-node-1.0.0.jar:?]
    at net.corda.node.services.statemachine.FlowStateMachineImpl.sendInternal(FlowStateMachineImpl.kt:332) ~[corda-node-1.0.0.jar:?]
    at net.corda.node.services.statemachine.FlowStateMachineImpl.initiateSession(FlowStateMachineImpl.kt:396) ~[corda-node-1.0.0.jar:?]
    at net.corda.node.services.statemachine.FlowStateMachineImpl.sendAndReceive(FlowStateMachineImpl.kt:200) ~[corda-node-1.0.0.jar:?]
    at net.corda.core.internal.FlowStateMachine$DefaultImpls.sendAndReceive$default(FlowStateMachine.kt:27) ~[corda-core-1.0.0.jar:?]
    at net.corda.node.services.statemachine.FlowSessionImpl.sendAndReceive(FlowSessionImpl.kt:25) ~[corda-node-1.0.0.jar:?]
    at net.corda.core.flows.DataVendingFlow.sendPayloadAndReceiveDataRequest(SendTransactionFlow.kt:70) ~[corda-core-1.0.0.jar:?]
    at net.corda.core.flows.DataVendingFlow.call(SendTransactionFlow.kt:48) ~[corda-core-1.0.0.jar:?]
    at net.corda.core.flows.DataVendingFlow.call(SendTransactionFlow.kt:31) ~[corda-core-1.0.0.jar:?]
    at net.corda.core.flows.FlowLogic.subFlow(FlowLogic.kt:212) ~[corda-core-1.0.0.jar:?]
    at net.corda.core.flows.CollectSignatureFlow.call(CollectSignaturesFlow.kt:140) ~[corda-core-1.0.0.jar:?]
    at net.corda.core.flows.CollectSignatureFlow.call(CollectSignaturesFlow.kt:134) ~[corda-core-1.0.0.jar:?]
    at net.corda.core.flows.FlowLogic.subFlow(FlowLogic.kt:212) ~[corda-core-1.0.0.jar:?]
    at net.corda.core.flows.CollectSignaturesFlow.call(CollectSignaturesFlow.kt:113) ~[corda-core-1.0.0.jar:?]
    at net.corda.core.flows.CollectSignaturesFlow.call(CollectSignaturesFlow.kt:64) ~[corda-core-1.0.0.jar:?]
    at net.corda.core.flows.FlowLogic.subFlow(FlowLogic.kt:212) ~[corda-core-1.0.0.jar:?]
    at net.corda.training.flow.IOUIssueFlow.call(IOUIssueFlow.kt:39) ~[classes/:?]
    at net.corda.training.flow.IOUIssueFlow.call(IOUIssueFlow.kt:22) ~[classes/:?]
    at net.corda.node.services.statemachine.FlowStateMachineImpl.run(FlowStateMachineImpl.kt:112) [corda-node-1.0.0.jar:?]
    at net.corda.node.services.statemachine.FlowStateMachineImpl.run(FlowStateMachineImpl.kt:40) [corda-node-1.0.0.jar:?]
    at co.paralleluniverse.fibers.Fiber.run1(Fiber.java:1092) [quasar-core-0.7.9-jdk8.jar:0.7.9]
    at co.paralleluniverse.fibers.Fiber.exec(Fiber.java:788) [quasar-core-0.7.9-jdk8.jar:0.7.9]
    at co.paralleluniverse.fibers.RunnableFiberTask.doExec(RunnableFiberTask.java:100) [quasar-core-0.7.9-jdk8.jar:0.7.9]
    at co.paralleluniverse.fibers.RunnableFiberTask.run(RunnableFiberTask.java:91) [quasar-core-0.7.9-jdk8.jar:0.7.9]
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_151]
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_151]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) [?:1.8.0_151]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) [?:1.8.0_151]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_151]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_151]
    at net.corda.node.utilities.AffinityExecutor$ServiceAffinityExecutor$1$thread$1.run(AffinityExecutor.kt:69) [corda-node-1.0.0.jar:?]
[WARN ] 16:38:13,670 [Mock network] (StateMachineManager.kt:93) flow.[49f2ebbd-be62-462d-af87-24f5ae40c7d6].uncaughtException - Caught exception from flow
java.lang.IllegalStateException: No transaction in context

.

Answer 1

For me the same issue was caused by missed @Suspendable annotation on a function in a call chain.

Guess the same might be true for lambda functions.

Answer 2

The issue is that the lambda captures a reference to the entire stack. If you try to suspend the flow at this point, it will attempt to serialise the entire stack via the lambda, and the suspension of the flow will fail.

The solution is to create a local variable, or to use the lambda inside a function. The lambda will then be discarded from the stack when the function returns, so there will be no issue at suspension time.