Seeing large memory consumption in reactor.core..SinkEmptyMulticast in version 10.2.4

[chinshaw]

I don't know exactly what version started exhibiting this issue but I just rolledback to version 9.2.2 and I don't see the same memory consumption problems. We have a pretty straight forward crud app written in kotlin that uses springboot, r2dbc and dgs.

In the snapshot memory is being collected in the SinkEmptyMulticast which seems to point to an issue with the dataloader event consumption. I tried tracing down but the reactor debug code is a pain to trace down anything useful. As you can see the Sink is consuming almost 250GB for a small amount of data. Kind of a pain to get the snapshot but the r2dbc threadpool consumed a total of 7GB but when passed back through the dataloader the memory allocation blew up. After rolling back to 9.2.2 I can see the batchlaoder stack uses around 1GB of memory. I did check the docs and tried turning on dgs.graphql.spring.webmvc.asyncdispatch.enabled however it didn't seem to make a difference.

This is our pretty simple wrapper that adapts the BatchLoader into a Kotlin coroutine scoped execution. Other than that it is simply making r2dbc database calls using spring CoroutineCrudRepository implementations.

@OptIn(DelicateCoroutinesApi::class)
interface CoroutineBatchLoaderWithContext<K : Any, V : Any> : MappedBatchLoaderWithContext<K, V>, DisposableBean {

    companion object {
        // Shared coroutine scope for all batch loaders to prevent scope proliferation
        private val batchLoaderScope = CoroutineScope(
            Dispatchers.IO + SupervisorJob() + CoroutineName("DataLoaderBatch")
        )
    }

    override fun load(
        keys: Set<K>,
        environment: BatchLoaderEnvironment
    ): CompletionStage<Map<K, V>> {
        val context = DgsContext.from(environment) as ReactiveDgsContext
        val coroutineContext = context.reactorContext?.asCoroutineContext() ?: EmptyCoroutineContext

        return batchLoaderScope.future(context = coroutineContext) {
            loadAsync(keys, environment)
        }
    }

    suspend fun loadAsync(
        keys: Set<K>,
        environment: BatchLoaderEnvironment
    ): Map<K, V>

    /**
     * Shutdown the batchloader scope
     */
    override fun destroy() {
        batchLoaderScope.cancel()
    }
}

This is a sample of our common pattern for resolvers if that helps. We typically return CompletableFuture's or List's.

    @DgsData(parentType = "Well", field = "productionCount")
    override fun count(dfe: DataFetchingEnvironment): CompletableFuture<Long> {
        val batchLoader = dfe.getDataLoader<Xid, Counts>(DailyProductionCountsBatchLoader.NAME)!!
        val wellXid = checkNotNull(dfe.getSource<WellEntity>()?.xid) { "Missing well xid while retrieving daily production count" }

        return batchLoader.load(wellXid).thenApply { it?.count ?: 0L }
    }

[chinshaw]

I rolled to version 10.4.0 and the issue does not occur but just to double check rolled back to 10.2.4 and was able to reproduce the memory issue.

[paulbakker]

Thanks for the follow-up. Since it's no longer happening on 10.4.0 I expect this was a change in Spring GraphQL (or downstream dependency in Spring) that caused the issue and later fixed it, since I can't think of any changes in DGS that would affect this.

Ok with closing this since it seems to be fixed?