You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@polaris.apache.org by "visit2rahul (via GitHub)" <gi...@apache.org> on 2026/05/19 23:25:12 UTC
[PR] Synchronize onComplete calls in InMemoryBufferEventListener (closes #4497) [polaris]
visit2rahul opened a new pull request, #4498:
URL: https://github.com/apache/polaris/pull/4498
## Summary
`UnicastProcessor.onNext()` is declared `public synchronized void` in smallrye-mutiny, but `onComplete()` is not synchronized. The two methods can therefore interleave: a concurrent `processEvent` (onNext) and eviction-driven `onComplete` on the same processor have no mutual exclusion, which can silently drop events at the eviction boundary.
This PR wraps both `onComplete()` call sites (the Caffeine eviction listener and the `shutdown()` loop) in `synchronized (processor) { ... }` blocks so they acquire the same intrinsic monitor that `onNext()` uses. `processEvent` itself does not change.
## Background
Surfaced during review of #4487 by @nandorKollar. @adutra agreed it warranted its own PR. Race interleaving details are in #4497.
## Test plan
No new test in this PR. The race window is microseconds at the eviction boundary, controlled by Caffeine's internal cleanup thread; a deterministic test of this interleaving is not portable. The fix is a small synchronization-only diff (+20 / -3) that reviewers can verify by inspection.
Existing tests pass locally:
- `InMemoryBufferEventListenerBufferSizeTest` (4 tests)
- `InMemoryBufferEventListenerBufferTimeTest` (1 test)
- `InMemoryBufferEventListenerIntegrationTest` (1 test)
Closes #4497.
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Synchronize onComplete calls in InMemoryBufferEventListener (closes #4497) [polaris]
Posted by "nandorKollar (via GitHub)" <gi...@apache.org>.
nandorKollar commented on PR #4498:
URL: https://github.com/apache/polaris/pull/4498#issuecomment-4502672144
> OK I'm spending too much time on this 😅 – but I think I have a version that eliminates all possible race conditions:
>
> ```java
> @ApplicationScoped
> @Identifier("persistence-in-memory-buffer")
> public class InMemoryBufferEventListener extends PolarisPersistenceEventListener {
>
> private static final Logger LOGGER = LoggerFactory.getLogger(InMemoryBufferEventListener.class);
>
> @Inject MetaStoreManagerFactory metaStoreManagerFactory;
> @Inject InMemoryBufferEventListenerConfiguration configuration;
>
> private static final class EvictedException extends IllegalStateException {}
>
> private final class EventProcessor {
>
> private final UnicastProcessor<PolarisEvent> processor;
> private final ReentrantLock lock = new ReentrantLock();
> private boolean completed = false; // guarded by lock
>
> EventProcessor(String realmId) {
> processor = UnicastProcessor.create();
> processor
> .emitOn(Infrastructure.getDefaultWorkerPool())
> .group()
> .intoLists()
> .of(configuration.maxBufferSize(), configuration.bufferTime())
> .subscribe()
> .with(events -> flush(realmId, events), error -> onProcessorError(realmId, error));
> }
>
> void onNext(PolarisEvent event) {
> lock.lock();
> try {
> if (completed) {
> throw new EvictedException();
> }
> processor.onNext(event);
> } finally {
> lock.unlock();
> }
> }
>
> void onComplete() {
> lock.lock();
> try {
> if (!completed) {
> completed = true;
> processor.onComplete();
> }
> } finally {
> lock.unlock();
> }
> }
> }
>
> @VisibleForTesting
> final LoadingCache<String, Processor> processors =
> Caffeine.newBuilder()
> .expireAfterAccess(Duration.ofHours(1))
> .evictionListener(
> (String realmId, EventProcessor processor, RemovalCause cause) -> {
> processor.onComplete();
> })
> .build(EventProcessor::new);
>
> private final ReadWriteLock shutdownLock = new ReentrantReadWriteLock();
> private boolean shutdown = false; // guarded by shutdownLock
>
> @Override
> protected void processEvent(String realmId, PolarisEvent event) {
> shutdownLock.readLock().lock();
> try {
> if (shutdown) {
> return;
> }
> while (true) {
> var processor = Objects.requireNonNull(processors.get(realmId));
> try {
> processor.onNext(event);
> return;
> } catch (EvictedException ignored) {
> // processor was evicted; loop to get a fresh one
> }
> }
> } finally {
> shutdownLock.readLock().unlock();
> }
> }
>
> @PreDestroy
> public void shutdown() {
> shutdownLock.writeLock().lock();
> try {
> shutdown = true;
> processors.asMap().values().forEach(Processor::onComplete);
> processors.invalidateAll(); // doesn't call the eviction listener
> } finally {
> shutdownLock.writeLock().unlock();
> }
> }
>
> @Retry(maxRetries = 5, delay = 1000, jitter = 100)
> @Fallback(fallbackMethod = "onFlushError")
> protected void flush(String realmId, List<PolarisEvent> events) {
> RealmContext realmContext = () -> realmId;
> var metaStoreManager = metaStoreManagerFactory.getOrCreateMetaStoreManager(realmContext);
> var basePersistence = metaStoreManagerFactory.getOrCreateSession(realmContext);
> var callContext = new PolarisCallContext(realmContext, basePersistence);
> metaStoreManager.writeEvents(callContext, events);
> }
>
> @SuppressWarnings("unused")
> protected void onFlushError(String realmId, List<PolarisEvent> events, Throwable error) {
> LOGGER.error("Failed to persist {} events for realm '{}'", events.size(), realmId, error);
> }
>
> protected void onProcessorError(String realmId, Throwable error) {
> LOGGER.error(
> "Unexpected error while processing events for realm '{}'; some events may have been dropped",
> realmId,
> error);
> processors.invalidate(realmId);
> }
> }
> ```
>
> The only minor nit is that when onProcessorError is called, the processor is left in an inconsistent state (not completed) so a subsequent call to `onNext()` is possible – but since the processor is errored out, I don't think we care.
>
> Again I'm not sure we need to be that paranoid. I'll defer to your judgement.
This uses now two locks when calling `onNext`: our reentrant lock, and the monitor of the processor no? I'd expect that this could cause some - probably negligible - performance degradation.
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Synchronize onComplete calls in InMemoryBufferEventListener (closes #4497) [polaris]
Posted by "visit2rahul (via GitHub)" <gi...@apache.org>.
visit2rahul commented on code in PR #4498:
URL: https://github.com/apache/polaris/pull/4498#discussion_r3284648595
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListener.java:
##########
@@ -68,7 +75,17 @@ protected void processEvent(String realmId, PolarisEvent event) {
@PreDestroy
public void shutdown() {
- processors.asMap().values().forEach(UnicastProcessor::onComplete);
+ // Same rationale as the eviction listener: hold the processor's monitor so
+ // shutdown-driven onComplete cannot race a concurrent processEvent-driven onNext.
+ processors
+ .asMap()
+ .values()
+ .forEach(
+ p -> {
+ synchronized (p) {
Review Comment:
Thank you @flyrain. Good suggestion - on reflection the helper isn't about line count but about encoding the synchronization as an invariant: any future caller adding a `processor.onComplete()` would have to actively bypass the helper rather than silently omit the `synchronized` wrapper, which is the right safety property for a thread-correctness fix. Extracted to `completeSynchronized` in 69fdd71c9.
Could you trigger the CI workflow run on this push when you have a moment?
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Synchronize onComplete calls in InMemoryBufferEventListener (closes #4497) [polaris]
Posted by "visit2rahul (via GitHub)" <gi...@apache.org>.
visit2rahul commented on PR #4498:
URL: https://github.com/apache/polaris/pull/4498#issuecomment-4500377960
Thank you @nandorKollar for the careful review and approval - much appreciated, especially given you were the one who flagged the underlying issue on #4487 that led to this fix. Now awaiting committer LGTM + CI workflow approval to trigger the test run.
Please review as your time permits.
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Synchronize onComplete calls in InMemoryBufferEventListener (closes #4497) [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on PR #4498:
URL: https://github.com/apache/polaris/pull/4498#issuecomment-4501582601
OK I'm spending too much time on this 😅 – but I think I have a version that eliminates all possible race conditions:
```java
@ApplicationScoped
@Identifier("persistence-in-memory-buffer")
public class InMemoryBufferEventListener extends PolarisPersistenceEventListener {
private static final Logger LOGGER = LoggerFactory.getLogger(InMemoryBufferEventListener.class);
@Inject MetaStoreManagerFactory metaStoreManagerFactory;
@Inject InMemoryBufferEventListenerConfiguration configuration;
private static final class EvictedException extends IllegalStateException {}
public final class Processor {
private final UnicastProcessor<PolarisEvent> processor;
private final ReentrantLock lock = new ReentrantLock();
private boolean completed = false; // guarded by lock
public Processor(String realmId) {
processor = UnicastProcessor.create();
processor
.emitOn(Infrastructure.getDefaultWorkerPool())
.group()
.intoLists()
.of(configuration.maxBufferSize(), configuration.bufferTime())
.subscribe()
.with(events -> flush(realmId, events), error -> onProcessorError(realmId, error));
}
public void onNext(PolarisEvent event) {
lock.lock();
try {
if (completed) {
throw new EvictedException();
}
processor.onNext(event);
} finally {
lock.unlock();
}
}
public void onComplete() {
lock.lock();
try {
if (!completed) {
completed = true;
processor.onComplete();
}
} finally {
lock.unlock();
}
}
}
@VisibleForTesting
final LoadingCache<String, Processor> processors =
Caffeine.newBuilder()
.expireAfterAccess(Duration.ofHours(1))
.evictionListener(
(String realmId, Processor processor, RemovalCause cause) -> {
processor.onComplete();
})
.build(Processor::new);
private final ReadWriteLock shutdownLock = new ReentrantReadWriteLock();
private boolean shutdown = false; // guarded by shutdownLock
@Override
protected void processEvent(String realmId, PolarisEvent event) {
shutdownLock.readLock().lock();
try {
if (shutdown) {
return;
}
while (true) {
var processor = Objects.requireNonNull(processors.get(realmId));
try {
processor.onNext(event);
return;
} catch (EvictedException ignored) {
// processor was evicted; loop to get a fresh one
}
}
} finally {
shutdownLock.readLock().unlock();
}
}
@PreDestroy
public void shutdown() {
shutdownLock.writeLock().lock();
try {
shutdown = true;
processors.asMap().values().forEach(Processor::onComplete);
processors.invalidateAll(); // doesn't call the eviction listener
} finally {
shutdownLock.writeLock().unlock();
}
}
@Retry(maxRetries = 5, delay = 1000, jitter = 100)
@Fallback(fallbackMethod = "onFlushError")
protected void flush(String realmId, List<PolarisEvent> events) {
RealmContext realmContext = () -> realmId;
var metaStoreManager = metaStoreManagerFactory.getOrCreateMetaStoreManager(realmContext);
var basePersistence = metaStoreManagerFactory.getOrCreateSession(realmContext);
var callContext = new PolarisCallContext(realmContext, basePersistence);
metaStoreManager.writeEvents(callContext, events);
}
@SuppressWarnings("unused")
protected void onFlushError(String realmId, List<PolarisEvent> events, Throwable error) {
LOGGER.error("Failed to persist {} events for realm '{}'", events.size(), realmId, error);
}
protected void onProcessorError(String realmId, Throwable error) {
LOGGER.error(
"Unexpected error while processing events for realm '{}'; some events may have been dropped",
realmId,
error);
processors.invalidate(realmId);
}
}
```
The only minor nit is that when onProcessorError is called, the processor is left in an inconsistent state (not completed) so a subsequent call to `onNext()` is possible – but since the processor is errored out, I don't think we care.
Again I'm not sure we need to be that paranoid. I'll defer to your judgement.
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Synchronize onComplete calls in InMemoryBufferEventListener (closes #4497) [polaris]
Posted by "visit2rahul (via GitHub)" <gi...@apache.org>.
visit2rahul commented on PR #4498:
URL: https://github.com/apache/polaris/pull/4498#issuecomment-4546469500
@adutra @flyrain hope you both are doing good. Checking if this is good to go? or we need any more work on this PR? please let me know!
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Synchronize onComplete calls in InMemoryBufferEventListener (closes #4497) [polaris]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on code in PR #4498:
URL: https://github.com/apache/polaris/pull/4498#discussion_r3277215570
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListener.java:
##########
@@ -56,8 +56,15 @@ public class InMemoryBufferEventListener extends PolarisPersistenceEventListener
Caffeine.newBuilder()
.expireAfterAccess(Duration.ofHours(1))
.evictionListener(
- (String realmId, UnicastProcessor<?> processor, RemovalCause cause) ->
- processor.onComplete())
+ (String realmId, UnicastProcessor<?> processor, RemovalCause cause) -> {
+ // onComplete is not synchronized in smallrye-mutiny's UnicastProcessor,
+ // unlike onNext. Acquire the processor's intrinsic monitor (the same one
+ // onNext uses) so that eviction-driven onComplete and concurrent
+ // processEvent-driven onNext mutually exclude.
+ synchronized (processor) {
Review Comment:
Comment is good. Worth one extra line: this assumes smallrye-mutiny's `UnicastProcessor.onNext` locks on `this` — re-verify on mutiny version bumps, otherwise this guard becomes a silent no-op.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListener.java:
##########
@@ -68,7 +75,17 @@ protected void processEvent(String realmId, PolarisEvent event) {
@PreDestroy
public void shutdown() {
- processors.asMap().values().forEach(UnicastProcessor::onComplete);
+ // Same rationale as the eviction listener: hold the processor's monitor so
+ // shutdown-driven onComplete cannot race a concurrent processEvent-driven onNext.
+ processors
+ .asMap()
+ .values()
+ .forEach(
+ p -> {
+ synchronized (p) {
Review Comment:
Same `synchronized (p) { p.onComplete(); }` pattern as the eviction listener — could pull into a tiny private helper (`completeSynchronized(processor)`) to keep both call sites in lockstep.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListener.java:
##########
@@ -56,8 +56,15 @@ public class InMemoryBufferEventListener extends PolarisPersistenceEventListener
Caffeine.newBuilder()
.expireAfterAccess(Duration.ofHours(1))
.evictionListener(
- (String realmId, UnicastProcessor<?> processor, RemovalCause cause) ->
- processor.onComplete())
+ (String realmId, UnicastProcessor<?> processor, RemovalCause cause) -> {
+ // onComplete is not synchronized in smallrye-mutiny's UnicastProcessor,
+ // unlike onNext. Acquire the processor's intrinsic monitor (the same one
+ // onNext uses) so that eviction-driven onComplete and concurrent
+ // processEvent-driven onNext mutually exclude.
+ synchronized (processor) {
Review Comment:
Thanks for fix, @visit2rahul ! Not a blocker, but is there a way to test it?
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Synchronize onComplete calls in InMemoryBufferEventListener (closes #4497) [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra merged PR #4498:
URL: https://github.com/apache/polaris/pull/4498
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Synchronize onComplete calls in InMemoryBufferEventListener (closes #4497) [polaris]
Posted by "visit2rahul (via GitHub)" <gi...@apache.org>.
visit2rahul commented on PR #4498:
URL: https://github.com/apache/polaris/pull/4498#issuecomment-4563570853
Hi @adutra, hope all is well! Filed the follow-up as #4572 -- taking it up now, will open a PR for your review once ready.
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Synchronize onComplete calls in InMemoryBufferEventListener (closes #4497) [polaris]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on code in PR #4498:
URL: https://github.com/apache/polaris/pull/4498#discussion_r3277215570
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListener.java:
##########
@@ -56,8 +56,15 @@ public class InMemoryBufferEventListener extends PolarisPersistenceEventListener
Caffeine.newBuilder()
.expireAfterAccess(Duration.ofHours(1))
.evictionListener(
- (String realmId, UnicastProcessor<?> processor, RemovalCause cause) ->
- processor.onComplete())
+ (String realmId, UnicastProcessor<?> processor, RemovalCause cause) -> {
+ // onComplete is not synchronized in smallrye-mutiny's UnicastProcessor,
+ // unlike onNext. Acquire the processor's intrinsic monitor (the same one
+ // onNext uses) so that eviction-driven onComplete and concurrent
+ // processEvent-driven onNext mutually exclude.
+ synchronized (processor) {
Review Comment:
Comment is good. Worth one extra line: this assumes smallrye-mutiny's `UnicastProcessor.onNext` locks on `this` — re-verify on mutiny version bumps, otherwise this guard becomes a silent no-op.
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Synchronize onComplete calls in InMemoryBufferEventListener (closes #4497) [polaris]
Posted by "visit2rahul (via GitHub)" <gi...@apache.org>.
visit2rahul commented on PR #4498:
URL: https://github.com/apache/polaris/pull/4498#issuecomment-4510836016
Thank you @adutra for the careful analysis and the wrapper sketch.
On the two concerns:
1. The future-Mutiny-implementation-detail is real, and the wrapper pattern is the right shape for it.
2. The eviction-gap race - you yourself noted it's "OK-ish because we don't have strong delivery guarantees" and "unrealistic". Agreed.
Would it work to do the same split that worked on #4487? Ship this PR as the minimal `onComplete` race fix (already approved by @nandorKollar and @flyrain), and file a follow-up issue with your wrapper as the starting point. The wrapper seems to have a few details worth iterating on independently - @nandorKollar already flagged the double-lock tradeoff, and the wrapper also doesn't close the durability gap during `@PreDestroy` (the `flush(events)` subscriber keeps running asynchronously after `shutdown()` returns). Would those be better debated in their own thread? Thoughts?
On @flyrain's "is there a way to test it" - the race window is microseconds at the Caffeine eviction boundary, so a deterministic test isn't portable (same rationale as the test we dropped from #4487). Happy to add a code comment at each call site explaining the race, or file a follow-up for an integration-style probabilistic test - WDYT?
@adutra @nandorKollar @flyrain please review as your time permits.
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Synchronize onComplete calls in InMemoryBufferEventListener (closes #4497) [polaris]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on PR #4498:
URL: https://github.com/apache/polaris/pull/4498#issuecomment-4546531003
Good from my side. Hi @adutra, any further feedback or unresolved comment?
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org