Memory management in Multi and Uni

[IshlahulHanif]

I'm pretty new to mutiny and reactive programming, so pardon if the answer is obvious

I have an issue where my instance consistently get java.lang.OutOfMemoryError: Java heap space error due to processing a high count of data from database. I mitigate this by replacing the Uni with Void at the end of Uni, assuming it will allow previously created list will be garbage collected if I do so.

This is roughly my code:

public Uni<Void>processData(Request request) {
    AtomicReference<List<Item>> listItemRef = new AtomicReference<>();
    return getAllMatchingItem(request)
            .onItem().transformToMulti(result -> Multi.createFrom().iterable(ListUtils.partition(result, 10000)))
            .onItem().transformToUni(resultBatched -> Uni.createFrom().item(() -> resultBatched)
                    .chain(resultBatched -> {
                        listItemRef.setRelease(resultBatched);
                        return getOtherItemBasedOnItem(resultBatched);
                    })
                    .chain(otherItems -> {
                        List<Item> items = listItemRef.getPlain();
                        return createProcessedItems(items, otherItems);
                    })
                    .chain(processedItems -> {
                        return insertProcessedItems(processedItems);
                    })
                    .onTermination().invoke(() -> {
                        LOGGER.info("Data process success");
                    })
                    .replaceWithVoid() // to make sure no remaining reference to upstream Unis
            )
            .merge(1)
            .collect().asList()
            .replaceWithVoid()
            .emitOn(MutinyHelper.executor(Vertx.currentContext()));
}

as for the functions, roughly they are like this:

public Uni<List<Item>> getAllMatchingItem(Request request) {
    // get data to DB, basically PanacheRepository.find(...).list
    return Uni.createFrom().item(() -> repo.find(" ... ").list)
            .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
            .emitOn(MutinyHelper.executor(Vertx.currentContext()));
}

public Uni<List<OtherItem>> getOtherItemBasedOnItem(List<Item> items) {
    // get data to DB, basically PanacheRepository.find(id in items.getIdSet).list()
    return Uni.createFrom().item(() -> repo.find("id in ?1", items.getIDSet()).list)
            .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
            .emitOn(MutinyHelper.executor(Vertx.currentContext()));
}

public Uni<List<ProcessedItem>> createProcessedItems(List<Item> items, List<OtherItem> otherItems) {
    return Uni.createFrom().item(() -> doCreateProcessedItems(item, otherItems))
            .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
            .emitOn(MutinyHelper.executor(Vertx.currentContext()));
}

public List<ProcessedItem> doCreateProcessedItems(List<Item> items, List<OtherItem> otherItems) {
    // process those items
    // basically create Map, create list, etc
    // no external function call, just plain code
}

public Uni<Void> insertProcessedItems(List<ProcessedItem> processedItems) {
    // basically just insert/update
    return Uni.createFrom().voidItem()
            .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
            .invoke(() -> repo.persist(processedItems))
            .emitOn(MutinyHelper.executor(Vertx.currentContext()));
}

I'm using non-reactive hibernate due to need multiple data source (reactive does not support that yet) and the project is too far to re-write the project in imperative way. I'm also not sure if using Multi is the correct approach here, instead of Uni<Collection> because the getAllMatchingItem is already in memory anyway

[jponge]

A couple of observations.

1. If you create a Multi from already in-memory items, then it's a good sign Uni<List<T>> is a better match, because you do get a Multi of convenience, but there is no form of back-pressure.
2. collect().asList() needs to buffer lots of items before emitting a result, which consumes memory. I'm not even sure why you do that since you have a replaceWithVoid right after.

[IshlahulHanif]

@jponge

1. So, if I can make the getAllMatchingItem paginated instead, can it reduce the memory usage?
2. I'm using collect().asList() so the processData will be finished when all the items are processed, for example I want to send a metrics telling the process is completed. If not using collect(), is there a better way?

[IshlahulHanif]

maybe change getAllMatchingItem into something like this

public Multi<List<Item>> getItemsPaginated(Request request, int pageSize) {
    Multi<List<Item>> sequence = Multi.createFrom().generator(() -> 0, (index, emitter) -> {
        List<Item> res = getItemsPaginatedNonUni(request, index, pageSize);
        if (res.isEmpty()) emitter.complete();
        emitter.emit(res);
        return index + 1;
    });

    return sequence
            .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
            .emitOn(MutinyHelper.executor(Vertx.currentContext()));
}

or is it better to use Uni<List<Item>> and do a .chain() in a for loop?

[jponge]

You create a Multi from items that are already in memory, so yes a Uni<List> is more appropriate.

[IshlahulHanif]

uhm, if I do a pagination requests like the example above, the items won't be in memory yet (right??)
in this case, is it ok to use Multi?

also you mention

collect().asList() needs to buffer lots of items before emitting a result

if not using .collect(), how can I wait until all items are emitted?

as for you said you're not even sure why you do that since I have a replaceWithVoid right after, is because I just need to wait until all the process is completed without I actually care about the result after theinsertProcessedItems(processedItems)