Stream.consumeAsFlow() finally block not getting called

[josephlbarnett]

We're using java.util.stream.Stream.consumeAsFlow() to process a stream as a flow.

our usage looked like:

val stream = jooqQuery.fetchLazy().stream().map { /* map result to data class */}
return stream.consumeAsFlow().flowOn(Dispatchers.IO)

Occasionally, it appears that the finally { stream.close() } block is not getting called (possibly on certain cancellation / error conditions?), resulting in connection leaks. However, it appears that if we add an onCompletion {} to the returned flow, that block does get called even when the finally gets skipped:

val stream = jooqQuery.fetchLazy().stream().map { /* map result to data class */}
return stream.consumeAsFlow().flowOn(Dispatchers.IO)
       .onCompletion { stream.close() } // should not need this, as the finally block in the StreamFlow should close it

Have verified this by duplicating the StreamFlow class and adding logging to the finally block, as well as having logging in the .onCompletion block above, and seen a few instances of the completion log firing without the finally log firing, but can't easily reproduce this (usually both fire).

[LouisCAD]

Do you have a self-contained reproducer, or a link to it from play.kotl.in?

[josephlbarnett]

the collecting code lives here: https://github.com/trib3/leakycauldron/blob/master/graphql/src/main/kotlin/com/trib3/graphql/websocket/GraphQLWebSocketConsumer.kt#L103-L126

Even in our real service, it's hard to reproduce on demand, but a toy example that should demonstrate the same issue lives at https://github.com/trib3/example-cauldron-service/ ( flow gets defined at https://github.com/trib3/example-cauldron-service/blob/master/server/src/main/kotlin/com/trib3/example/server/graphql/Subscription.kt ). Using apollographql's websocket transport to connect and kick off a bunch of subscription queries, then closing those connections might be the best way to try to trigger it.

[qwwdfsad]

It's not exactly a Stream.asFlow() issue, but rather flowOn(...).asPublisher().asFlow().
As a workaround, getting rid of asPublisher().asFlow() chain should help.

[qwwdfsad]

Self-contained repro (prints Done [2] and Finished):

@Test
fun testNonAtomicStart() = runBlocking<Unit> {
    try {
        coroutineScope {
            val dispatcher = wrapperDispatcher()
            val job = coroutineContext[Job]!!
            val flow = flowOf(1, 2, 3)
                .onCompletion { println("Done [1]") }
                .flowOn(dispatcher)
                .onCompletion { println("Done [2]") }

            launch {
                flow.collect()
            }
            launch {
                job.cancel()
            }
        }
    } catch (e: Throwable) {
        println("Finished")
    }
}

[josephlbarnett]

Interesting. Agree that the .asPublisher().asFlow() is not ideal, so may look into building Flow support for the graphql-java layer to avoid that. In the meantime, the post-flowOn() onCompletion {} block is a workaround that seems to work -- is it fully safe or is there still potential for the issue to manifest?

[qwwdfsad]

is it fully safe or is there still potential for the issue to manifest?

It is fully safe.

What about asPublisher().asFlow() -- it (by its nature, not by its implementation) breaks the structured concurrency and the order of onCompletion blocks: reactive-streams/reactive-streams-jvm#482 So I would recommend avoiding using this combination at all