confluentinc / confluentinc/parallel-consumer

Memory Leak from JStreamVertxParallelStreamProcessor

Open
#912 1 comment 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
299
Forks
172
PR merge metrics
No merged PRs in 30d

Description

Image

```
// tag::example[]
var resultStream = parallelConsumer.vertxHttpReqInfoStream(context -> {
var consumerRecord = context.getSingleConsumerRecord();
log.info("Concurrently constructing and returning RequestInfo from record: {}", consumerRecord);
Map 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.

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.