-
|
I have a strange problem - code works with merge, but not with concatenate. Does anyone knows what could I do to locate the problem (using profilers, or adding some manual debugging)? The problem that I have is that the code below is just hanging and if I enable When I tried switching between those two methods and log the debug info: With I'm posting here parts of the code that I think are relevant for the issue I want to describe: when it fails with timeout this is part of the exception that I get: |
Beta Was this translation helpful? Give feedback.
Replies: 7 comments 19 replies
-
|
I can even find a group that is entering the processing, but it never emits the |
Beta Was this translation helpful? Give feedback.
-
|
After a lot more debugging and whatnot, I pinned the problem. first few markets have "void_reason", so that TRUE.equals(group.key()) and only that group is emitted (for each "void_reason":"REASON" I can see in the logs |
Beta Was this translation helpful? Give feedback.
-
|
It's running on |
Beta Was this translation helpful? Give feedback.
-
|
(my bad, I may be confusing with another issue that I've seen today, but still, it's never a good idea when reactive code is on the main thread) |
Beta Was this translation helpful? Give feedback.
-
|
It's a strange problem. I managed to reproduce on simple example, so I'll build sample code and share the repo. |
Beta Was this translation helpful? Give feedback.
-
|
Here is the link to smallest reproducible I created repo And it happens even if I change the running thread: |
Beta Was this translation helpful? Give feedback.
-
|
@jponge seems that somehow other group cannot have more than 130 elements? Otherwise it hangs. |
Beta Was this translation helpful? Give feedback.
-
|
Here is the smallest reproducable I created so far - change elements to 119 and it will work!? |
Beta Was this translation helpful? Give feedback.
-
|
And this is the reason: Why is it 128? |
Beta Was this translation helpful? Give feedback.
-
|
I don't think that's the root cause, 128 is for batching (we could make it configurable TBH, a hard coded value is never optimal). The problem you are observing are likely due to:
|
Beta Was this translation helpful? Give feedback.
-
|
It probably is not the root cause, but if I increase this value - it completes. Probably because then the first batch pulls all of the elements in. This definitelly seems like some error in mutiny to me (since it works for N elements, but not for N+1)? Otherwise if users need to know that their app will hang in production if element number is >128... The original code is taking values from redis, and that is executing on separate thread in original code. |
Beta Was this translation helpful? Give feedback.
-
|
To me the problem is that you get stuck on a group until it completes, because you are using a concatenation. It's got nothing to do with requesting batches of 128 elements. |
Beta Was this translation helpful? Give feedback.
-
|
@jponge Can you then please tell me what change to do to my last code I posted above, so that it works for larger json arrays? |
Beta Was this translation helpful? Give feedback.
-
|
Also note that your emitter base code (the loop that creates |
Beta Was this translation helpful? Give feedback.
-
|
There are also weird things like having a I got a timeout with something like this: public Multi<JsonNode> transform() {
List<JsonNode> str = createProblematicJson();
List<JsonNode> subs = str.stream().toList();
return Multi.createFrom().iterable(subs)
.onItem().transformToMultiAndConcatenate(
item -> {
// Create a new Multi for each item
return Multi.createFrom().iterable(item);
})
.group().by(market -> market.has("grouping_reason"))
.onItem().transformToMultiAndConcatenate(
group -> {
return TRUE.equals(group.key())
? createSeparateMessageForEachVoidedMarket(group)
: updateMessageToHaveOnlySettledMarkets(group);
})
.ifNoItem().after(Duration.ofSeconds(10)).fail().onFailure().invoke(throwable -> {
System.out.println("❌ Failure: " + throwable);
throwable.printStackTrace();
});
}That being said I have limited time right now, and I think you should really step back and simplify the code, because there are too many places where implicit subscriptions happen (collect), single element streams are made, etc. I have also no idea what the larger context is, but if you are struggling with reactive code in your Quarkus app, perhaps it's better to start with imperative programming instead. |
Beta Was this translation helpful? Give feedback.
-
|
The example I created was only for demonstration. Even though it has a lot of possible optimizations - it should still work. I guess this code that also fails cannot be optimized further? And regarding imperative programming - I agree with you. |
Beta Was this translation helpful? Give feedback.
-
|
I'll try to have a look again when I have time, this is a more approachable reproducer indeed |
Beta Was this translation helpful? Give feedback.
-
|
Ok I've found the issue in your code, I will post an explanation 😉 |
Beta Was this translation helpful? Give feedback.
-
|
Here is a more minimal reproducer with working code based on the logic you had: List<Integer> result = Multi.createFrom().range(0, 300)
.log()
.group().by(x -> x > 20)
.onItem().transformToMultiAndMerge(
group -> group.key()
? group.collect().asList().replaceWith(1).toMulti()
: group.collect().asList().replaceWith(2).toMulti()
)
.onCompletion().invoke(() -> System.out.println("Transformation completed"))
.ifNoItem().after(Duration.ofSeconds(2)).fail()
.runSubscriptionOn(Executors.newSingleThreadExecutor())
.collect().asList().await().indefinitely();
System.out.println(result);The problem in your code was that the If you extrapolate that to the rest of your code, you should see why there was a problem. |
Beta Was this translation helpful? Give feedback.
-
|
OK. Thanks for an answer. Just one more question - why if I use |
Beta Was this translation helpful? Give feedback.
-
|
And also - why my example works with 120 items, but not with more? |
Beta Was this translation helpful? Give feedback.
-
|
You are right, there's a something to investigate then: List<Integer> result = Multi.createFrom().range(0, 300)
.group().by(x -> x % 2 == 0)
.onItem().transformToMultiAndConcatenate(
group -> group.key()
? group.collect().asList().replaceWith(1).toMulti()
: group.collect().asList().replaceWith(2).toMulti()
)
.ifNoItem().after(Duration.ofSeconds(2)).fail()
.collect().asList().await().indefinitely();
System.out.println(result); |
Beta Was this translation helpful? Give feedback.
-
|
Captured in #1856 |
Beta Was this translation helpful? Give feedback.
Here is a more minimal reproducer with working code based on the logic you had: