Memory Leak from JStreamVertxParallelStreamProcessor

[sharifahmad2061]

 // tag::example[]
        var resultStream = parallelConsumer.vertxHttpReqInfoStream(context -> {
            var consumerRecord = context.getSingleConsumerRecord();
            log.info("Concurrently constructing and returning RequestInfo from record: {}", consumerRecord);
            Map<String, String> params = UniMaps.of("recordKey", consumerRecord.key(), "payload", consumerRecord.value());
            return new RequestInfo("localhost", port, "/api", params); // <1>
        });
        // end::example[]

        resultStream.forEach(x -> {
            log.info("From result stream: {}", x);
        });

existing example or code written in similar fashion will get into OOM situation if stream becomes empty at some point and is populated later on again.

in our case we didn't consume the result stream at all and it cause OOM errors.

what should be the correct way to consume resultStream. or I am wrong and resultStream.forEach should work fine.