adutra commented on issue #2568:
URL: https://github.com/apache/polaris/issues/2568#issuecomment-4477616719
@visit2rahul thanks for pointing this out: you're right, there is a race
condition when `onNext` is called.
The Reactive Streams spec (rule 1.3) requires `onNext()` to be called
sequentially. SmallRye Mutiny's `UnicastProcessor` relies on this contract and
is not safe for concurrent `onNext()` calls: concurrent emissions can silently
drop events.
In the current code, `processEvent` is called from HTTP request handler
threads, so multiple concurrent requests for the same realm can race on
`processor.onNext(event)`. The Caffeine cache ensures only one
`UnicastProcessor` is created per realm, but it offers no protection against
concurrent use of it.
Le's repurpose this issue. As for the fix, I think we can just synchronize
on the processor:
```java
protected void processEvent(String realmId, PolarisEvent event) {
var processor = Objects.requireNonNull(processors.get(realmId));
synchronized (processor) {
processor.onNext(event);
}
}
```
A more sophisticated approach using a `MultiEmitter` would also be possible,
but I don't think it's worth the added complexity.
Are you planning to tackle this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
