You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@polaris.apache.org by "adutra (via GitHub)" <gi...@apache.org> on 2026/06/08 14:05:37 UTC
[PR] Enforce ordered per-listener event delivery [polaris]
adutra opened a new pull request, #4648:
URL: https://github.com/apache/polaris/pull/4648
This change re-enables ordered per-listener event delivery, that #4616 accidentally broke.
<!--
๐ Describe what changes you're proposing, especially breaking or user-facing changes.
๐ See https://github.com/apache/polaris/blob/main/CONTRIBUTING.md for more.
-->
## Checklist
- [ ] ๐ก๏ธ Don't disclose security issues! (contact security@apache.org)
- [ ] ๐ Clearly explained why the changes are needed, or linked related issues: Fixes #
- [ ] ๐งช Added/updated tests with good coverage, or manually tested (and explained how)
- [ ] ๐ก Added comments for complex logic
- [ ] ๐งพ Updated `CHANGELOG.md` (if needed)
- [ ] ๐ Updated documentation in `site/content/in-dev/unreleased` (if needed)
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3404904645
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListenerConfiguration.java:
##########
@@ -84,12 +84,30 @@ interface Executor {
/**
* The thread pool size. The default is -1, which is interpreted as the number of available
- * cores.
+ * listeners, capped by the number of cores
*/
@WithDefault("-1")
int poolSize();
- /** The queue size. The default is -1, which is interpreted as unbounded. */
+ /**
+ * The queue size. The default is -1, which is interpreted as unbounded. This limit is applied
+ * globally and affects all listeners. To configure the per-listener queue size, use {@code
+ * polaris.event-listener.listener-backlog.queue-size} instead.
+ */
+ @WithDefault("-1")
+ int queueSize();
+ }
+
+ /** Configuration for per-listener event backlogs. */
+ ListenerBacklog listenerBacklog();
+
+ interface ListenerBacklog {
+
+ /**
+ * The per-listener backlog queue size. The default is -1, which is interpreted as unbounded. A
+ * bounded backlog means event delivery is best-effort and events will be dropped/logged when
+ * the backlog is full.
+ */
@WithDefault("-1")
int queueSize();
Review Comment:
Heads up: I decided to use different values for the executor queue and the listener backlog queue.
My reasoning:
The shared executor only receives one drain task per active listener, so bursts mostly pressure the per-listener queue, not the shared executor queue. A low value for the executor queue is therefore OK, but listener backlog queues should probably be set to larger bounds, if any.
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3389636827
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
How about using a Vert.x timer for re-executing drain tasks that disappear due to submit/schedule races?
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3384108334
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +163,54 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Exception e) {
+ draining.set(false);
Review Comment:
Maybe nitpicky, but this still feels a bit risky... `OutOfMemoryError`, for example, is not an `Exception` ๐ค
`catch (Throwable)` works in this context in newer java versions... WDYT? ๐ค
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +163,54 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Exception e) {
+ draining.set(false);
+ throw e;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
+ draining.set(false);
+ // tasks remain in queue; next submission will trigger a new drain
Review Comment:
Could we log `e` as a WARN maybe?.. just for posterity / trouble-shooting ๐ค
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3405369855
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +166,58 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ @SuppressWarnings("FutureReturnValueIgnored")
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ maybeScheduleDrain();
Review Comment:
Should we do `.exceptionally(this::onError)` here too? Line 200 may return a failed `CompletableFuture` without logging its exception, I guess ๐ค
In this case, we probably do not have to reset `draining` on line 199, WDYT?
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3383190521
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -116,4 +143,43 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType polarisEventType) {
return enabledEventTypes.contains(polarisEventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ delegate.execute(this::drain);
Review Comment:
I think you meant Mutiny instead. I'm not sure there's any value in using a reactive framework (which is by essence non-blocking) to handle blocking operations.
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3384092469
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -116,4 +143,43 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType polarisEventType) {
return enabledEventTypes.contains(polarisEventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ delegate.execute(this::drain);
Review Comment:
Mutiny works too... I just found Reactor Core first in my IJ :)
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3383157521
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -116,4 +143,43 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType polarisEventType) {
return enabledEventTypes.contains(polarisEventType);
}
+
+ private static class ListenerExecutor implements Executor {
Review Comment:
Indeed, but I wanted to keep the abstraction clean. Also, we may need to keep this class as an `Executor` if we introduce a feature flag for 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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3389338212
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
That would open a can of worms... we are in a finally block, so throwing here potentially masks another throwable already being thrown. There is no simple way I'm afraid.
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3388819929
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
`catch (Throwable)` here too... for symmetry?
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3395623965
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
> Ideally we should add the re-submission exception to the task's exception as "suppressed"
Yes that's possible; but as you guessed, it's quite convoluted:
```java
private void drain() {
Throwable toThrow = null;
try {
Runnable task;
while ((task = queue.poll()) != null) {
task.run();
}
} catch (RuntimeException | Error t1) {
toThrow = t1;
} finally {
// A producer may have enqueued a task between the last poll() and the set(false) above,
// and seen draining=true so it did not schedule a drain. Catch that here.
if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
try {
// re-schedule for fairness
delegate.execute(this::drain);
} catch (RuntimeException | Error t2) {
// tasks remain in queue; next submission will trigger a new drain
draining.set(false);
if (toThrow == null) {
toThrow = t2;
} else {
toThrow.addSuppressed(t2);
}
}
}
}
if (toThrow != null) {
LOGGER.warn("Error while draining event queue", toThrow);
if (toThrow instanceof RuntimeException) {
throw (RuntimeException) toThrow;
}
throw (Error) toThrow;
}
}
```
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra merged PR #4648:
URL: https://github.com/apache/polaris/pull/4648
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3387530844
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +163,54 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Exception e) {
+ draining.set(false);
Review Comment:
I tend to think that if an `Error` is thrown, we're in a bigger trouble than just having `draining` out of syc. But OK to change.
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3389609645
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
Ah, sorry, I missed that we're not re-throwing here :thinking:
However, if an `Error` occurs, it will mask the original exception anyway, whether we re-throw it or just ignore :thinking:
Where do task execution exception go in this case? :thinking:
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3383004503
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -116,4 +143,43 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType polarisEventType) {
return enabledEventTypes.contains(polarisEventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ delegate.execute(this::drain);
Review Comment:
WDYT about using something from `reactor.core` to build the ordered delivery pipeline?
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3389636827
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
How about using a Vert.x timer for re-executing drain tasks that disappear due to submit/drain races?
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
How about using a Vert.x timer for re-executing tasks that disappear due to submit/drain races?
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3389643509
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
or maybe `VertxAsyncExec`?
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3395714674
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
`VertxAsyncExec` has the nice advantage of handling retries internally in case of execution rejected. But I'm also a little worried that it keeps retrying indefinitely, apparently.
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3384095126
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -116,4 +143,43 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType polarisEventType) {
return enabledEventTypes.contains(polarisEventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ delegate.execute(this::drain);
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
Review Comment:
Fair enough.
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3404635082
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
Yes correct. That's why you have the `.exceptionally(this::onError)` call.
I'm going to use `CompletableFuture` then, thanks!
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3395637278
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
And now `VertxAsyncExec`. Switching to this component comes with a few caveats:
- It's shared with other parts of the application (NoSQL, OPA). Unless we create a separate bean, we'd have to share the underlying executor. But event delivery work may be quite intense depending on how much activity is received and how many listeners there are (and how slow they can be).
- Using `VertxAsyncExce.schedulePeriodic()` is tempting, but would move event delivery from a push model (deliver as soon as an event arrives) to a pure polling mechanism. I think we should stick with `VertxAsyncExce.schedule()`, but with a delay of zero.
Here is a version with `VertxAsyncExec`:
```java
@Override
public void execute(@NonNull Runnable task) {
if (!queue.offer(task)) {
throw new RejectedExecutionException("Event listener queue is full");
}
if (draining.compareAndSet(false, true)) {
asyncExec.schedule(this::drain, Duration.ZERO); // doesn't throw
}
}
private void drain() {
try {
Runnable task;
while ((task = queue.poll()) != null) {
task.run();
}
} finally {
draining.set(false);
if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
asyncExec.schedule(this::drain, Duration.ZERO); // doesn't throw
}
}
}
```
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3389621403
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
Ideally we should add the re-submission exception to the task's exception, I think... but it's becoming too involved, I agree.... This is why I was wondering about reusing Mutiny of some other framework here :sweat_smile:
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
Ideally we should add the re-submission exception to the task's exception as "suppressed", I think... but it's becoming too involved, I agree.... This is why I was wondering about reusing Mutiny of some other framework here :sweat_smile:
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3397266115
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
I like the version with `CompletableFuture` :wink:
Thanks for the thorough analysis and alternatives :+1:
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3382868783
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -116,4 +143,43 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType polarisEventType) {
return enabledEventTypes.contains(polarisEventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ delegate.execute(this::drain);
Review Comment:
This may throw (e.g. `RejectedExecutionException`) and cause the `draining` flag to be out-of-sync with actual task execution.
##########
runtime/service/src/test/java/org/apache/polaris/service/events/PolarisEventListenersOrderedDeliveryTest.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.polaris.service.events;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+import com.google.common.collect.ImmutableMap;
+import io.quarkus.runtime.BlockingOperationControl;
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.QuarkusTestProfile;
+import io.quarkus.test.junit.TestProfile;
+import io.smallrye.common.annotation.Identifier;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.IntStream;
+import org.apache.polaris.service.events.listeners.PolarisEventListener;
+import org.junit.jupiter.api.Test;
+
+@QuarkusTest
+@TestProfile(
+ PolarisEventListenersOrderedDeliveryTest.PolarisEventListenersOrderedDeliveryTestProfile.class)
+public class PolarisEventListenersOrderedDeliveryTest {
+
+ static final int EVENTS_COUNT = 500;
+
+ public static class PolarisEventListenersOrderedDeliveryTestProfile
+ implements QuarkusTestProfile {
+ @Override
+ public Map<String, String> getConfigOverrides() {
+ return ImmutableMap.<String, String>builder()
+ .put("polaris.event-listener.executor.queue-size", String.valueOf(EVENTS_COUNT))
+ .put(
+ "polaris.event-listener.types",
+ "ordered-listener1,ordered-listener2,ordered-listener3")
+ .put(
+ "polaris.event-listener.ordered-listener1.enabled-event-types",
+ "AFTER_SEND_NOTIFICATION")
+ .put(
+ "polaris.event-listener.ordered-listener2.enabled-event-types",
+ "AFTER_SEND_NOTIFICATION")
+ .put(
+ "polaris.event-listener.ordered-listener3.enabled-event-types",
+ "AFTER_SEND_NOTIFICATION")
+ .build();
+ }
+
+ @Singleton
+ @Identifier("ordered-listener1")
+ OrderedEventListener orderedEventListener1() {
+ return new OrderedEventListener();
+ }
+
+ @Singleton
+ @Identifier("ordered-listener2")
+ OrderedEventListener orderedEventListener2() {
+ return new OrderedEventListener();
+ }
+
+ @Singleton
+ @Identifier("ordered-listener3")
+ OrderedEventListener orderedEventListener3() {
+ return new OrderedEventListener();
+ }
+ }
+
+ static class OrderedEventListener implements PolarisEventListener {
+
+ static final AttributeKey<Integer> SEQ = new AttributeKey<>("seq", Integer.class);
+
+ final List<Integer> receivedSequences = new CopyOnWriteArrayList<>();
+ final AtomicInteger receivedEventCount = new AtomicInteger();
+
+ final CountDownLatch pauseLatch = new CountDownLatch(1);
+
+ @Override
+ public void onEvent(PolarisEvent event) {
+ assertThat(BlockingOperationControl.isBlockingAllowed()).isTrue();
+ if (!event.attributes().contains(SEQ)) {
+ return;
+ }
+ try {
+ pauseLatch.await();
Review Comment:
nit: wait timeout
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -116,4 +143,43 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType polarisEventType) {
return enabledEventTypes.contains(polarisEventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ delegate.execute(this::drain);
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
Review Comment:
When we have more listeners that executor threads, this can cause starvation for some listeners, I guess ๐ค
Should we cap the iteration size?
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -116,4 +143,43 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType polarisEventType) {
return enabledEventTypes.contains(polarisEventType);
}
+
+ private static class ListenerExecutor implements Executor {
Review Comment:
nit: this is a purely local class, I guess it does not really have to implement `Executor`
##########
runtime/service/src/test/java/org/apache/polaris/service/events/PolarisEventListenersOrderedDeliveryTest.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.polaris.service.events;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+import com.google.common.collect.ImmutableMap;
+import io.quarkus.runtime.BlockingOperationControl;
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.QuarkusTestProfile;
+import io.quarkus.test.junit.TestProfile;
+import io.smallrye.common.annotation.Identifier;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.IntStream;
+import org.apache.polaris.service.events.listeners.PolarisEventListener;
+import org.junit.jupiter.api.Test;
+
+@QuarkusTest
+@TestProfile(
+ PolarisEventListenersOrderedDeliveryTest.PolarisEventListenersOrderedDeliveryTestProfile.class)
+public class PolarisEventListenersOrderedDeliveryTest {
+
+ static final int EVENTS_COUNT = 500;
+
+ public static class PolarisEventListenersOrderedDeliveryTestProfile
+ implements QuarkusTestProfile {
+ @Override
+ public Map<String, String> getConfigOverrides() {
+ return ImmutableMap.<String, String>builder()
+ .put("polaris.event-listener.executor.queue-size", String.valueOf(EVENTS_COUNT))
+ .put(
+ "polaris.event-listener.types",
+ "ordered-listener1,ordered-listener2,ordered-listener3")
+ .put(
+ "polaris.event-listener.ordered-listener1.enabled-event-types",
+ "AFTER_SEND_NOTIFICATION")
+ .put(
+ "polaris.event-listener.ordered-listener2.enabled-event-types",
+ "AFTER_SEND_NOTIFICATION")
+ .put(
+ "polaris.event-listener.ordered-listener3.enabled-event-types",
+ "AFTER_SEND_NOTIFICATION")
+ .build();
+ }
+
+ @Singleton
+ @Identifier("ordered-listener1")
+ OrderedEventListener orderedEventListener1() {
+ return new OrderedEventListener();
+ }
+
+ @Singleton
+ @Identifier("ordered-listener2")
+ OrderedEventListener orderedEventListener2() {
+ return new OrderedEventListener();
+ }
+
+ @Singleton
+ @Identifier("ordered-listener3")
+ OrderedEventListener orderedEventListener3() {
+ return new OrderedEventListener();
+ }
+ }
+
+ static class OrderedEventListener implements PolarisEventListener {
+
+ static final AttributeKey<Integer> SEQ = new AttributeKey<>("seq", Integer.class);
+
+ final List<Integer> receivedSequences = new CopyOnWriteArrayList<>();
+ final AtomicInteger receivedEventCount = new AtomicInteger();
+
+ final CountDownLatch pauseLatch = new CountDownLatch(1);
+
+ @Override
+ public void onEvent(PolarisEvent event) {
+ assertThat(BlockingOperationControl.isBlockingAllowed()).isTrue();
+ if (!event.attributes().contains(SEQ)) {
+ return;
+ }
+ try {
+ pauseLatch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ receivedSequences.add(event.attributes().getRequired(SEQ));
+ receivedEventCount.incrementAndGet();
+ }
+ }
+
+ @Inject PolarisEventDispatcher eventDispatcher;
+
+ @Inject
+ @Identifier("ordered-listener1")
+ OrderedEventListener orderedEventListener1;
+
+ @Inject
+ @Identifier("ordered-listener2")
+ OrderedEventListener orderedEventListener2;
+
+ @Inject
+ @Identifier("ordered-listener3")
+ OrderedEventListener orderedEventListener3;
+
+ @Test
+ public void testOrderedEventDelivery() {
+
+ for (int i = 0; i < EVENTS_COUNT; i++) {
+ var attrs = new EventAttributeMap().put(OrderedEventListener.SEQ, i);
+ eventDispatcher.dispatch(
+ new PolarisEvent(PolarisEventType.AFTER_SEND_NOTIFICATION, null, attrs));
+ }
+
+ orderedEventListener1.pauseLatch.countDown();
+ orderedEventListener2.pauseLatch.countDown();
+ orderedEventListener3.pauseLatch.countDown();
Review Comment:
WDYT about unblocking one of the listeners right away, one in the middle of the event submission process and one at the very end?
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3383199456
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -116,4 +143,43 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType polarisEventType) {
return enabledEventTypes.contains(polarisEventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ delegate.execute(this::drain);
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
Review Comment:
I am not sure it's worth the extra configuration option. The goal is to have at least as many threads as listeners, which is the default setup โ unless you have more listeners than CPU cores which seems unlikely with modern hardware.
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3405312146
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListenerConfiguration.java:
##########
@@ -84,12 +84,30 @@ interface Executor {
/**
* The thread pool size. The default is -1, which is interpreted as the number of available
- * cores.
+ * listeners, capped by the number of cores
*/
@WithDefault("-1")
int poolSize();
- /** The queue size. The default is -1, which is interpreted as unbounded. */
+ /**
+ * The queue size. The default is -1, which is interpreted as unbounded. This limit is applied
+ * globally and affects all listeners. To configure the per-listener queue size, use {@code
+ * polaris.event-listener.listener-backlog.queue-size} instead.
+ */
+ @WithDefault("-1")
+ int queueSize();
+ }
+
+ /** Configuration for per-listener event backlogs. */
+ ListenerBacklog listenerBacklog();
+
+ interface ListenerBacklog {
+
+ /**
+ * The per-listener backlog queue size. The default is -1, which is interpreted as unbounded. A
+ * bounded backlog means event delivery is best-effort and events will be dropped/logged when
+ * the backlog is full.
+ */
@WithDefault("-1")
int queueSize();
Review Comment:
good point :+1:
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3395627890
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
> This is why I was wondering about reusing Mutiny of some other framework here ๐
Any async framework could be leveraged to coordinate failure handling in a more composable way, for sure. The only caveat is that `Error`s are not propagated, since the error handling is self-contained.
Here is a version with `CompletableFuture`:
```java
@Override
@SuppressWarnings("FutureReturnValueIgnored")
public void execute(@NonNull Runnable task) {
if (!queue.offer(task)) {
throw new RejectedExecutionException("Event listener queue is full");
}
if (draining.compareAndSet(false, true)) {
try {
scheduleDrain();
} catch (Throwable t) {
draining.set(false);
throw t;
}
}
}
private CompletableFuture<Void> scheduleDrain() {
return CompletableFuture.runAsync(this::drain, delegate)
.thenCompose(this::maybeReschedule)
.exceptionally(this::onError);
}
private void drain() {
try {
Runnable task;
while ((task = queue.poll()) != null) {
task.run();
}
} finally {
draining.set(false);
}
}
private CompletableFuture<Void> maybeReschedule(Void unused) {
if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
// re-schedule for fairness
return scheduleDrain(); // this may throw; it would fail the stage it's running in
}
return CompletableFuture.completedFuture(null);
}
private Void onError(Throwable error) {
draining.set(false);
LOGGER.warn("Failed to schedule listener executor drain; tasks may be delayed", error);
return null;
}
```
And here is the version with Mutiny:
```java
Override
public void execute(@NonNull Runnable task) {
if (!queue.offer(task)) {
throw new RejectedExecutionException("Event listener queue is full");
}
if (draining.compareAndSet(false, true)) {
scheduleDrain().subscribe().with(unused -> {}); // doesn't throw synchronously
}
}
private Uni<Void> scheduleDrain() {
return Uni.createFrom()
.item(this::drain)
.runSubscriptionOn(delegate)
.call(this::maybeReschedule)
.onFailure()
.recoverWithItem(this::onError);
}
private Void drain() {
try {
Runnable task;
while ((task = queue.poll()) != null) {
task.run();
}
} finally {
draining.set(false);
}
return null;
}
private Uni<Void> maybeReschedule(Object unused) {
if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
// re-schedule for fairness
return scheduleDrain(); // doesn't throw synchronously
}
return Uni.createFrom().voidItem();
}
private Void onError(Throwable error) {
draining.set(false);
LOGGER.error(
"Failed to schedule listener executor drain; event delivery may be delayed", error);
return null;
}
```
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3395621784
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
Ok that's a lot - let's go step by step ๐
> Where do task execution exceptions go in this case? ๐ค
It goes to the uncaught exception handler, which by default is a `org.jboss.threads.LoggingUncaughtExceptionHandler`. It logs a message like this:
```
2026-06-11 11:52:36,707 ERROR [org.jbo.thr.errors] [,] [,,,] (executor-thread-2) Thread Thread[#126,executor-thread-2,5,main] threw an uncaught exception: java.util.concurrent.RejectedExecutionException: blah blah
at org.apache.polaris.service.events.PolarisEventListeners$ListenerExecutor.drain(PolarisEventListeners.java:206)
at io.smallrye.context.impl.wrappers.SlowContextualRunnable.run(SlowContextualRunnable.java:19)
at org.jboss.threads.EnhancedViewExecutor$EnhancedViewExecutorRunnable.run(EnhancedViewExecutor.java:496)
at io.quarkus.vertx.core.runtime.VertxCoreRecorder$15.runWith(VertxCoreRecorder.java:695)
at org.jboss.threads.EnhancedQueueExecutor$Task.doRunWith(EnhancedQueueExecutor.java:2651)
at org.jboss.threads.EnhancedQueueExecutor$Task.run(EnhancedQueueExecutor.java:2630)
at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1586)
at org.jboss.threads.DelegatingRunnable.run(DelegatingRunnable.java:11)
```
It does this for any throwable, including `Error`. Errors do not propagate up the stack.
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3395623965
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
> Ideally we should add the re-submission exception to the task's exception as "suppressed"
Yes that's possible; but as you guessed, it's quite convoluted:
```java
private void drain() {
Throwable toThrow = null;
try {
Runnable task;
while ((task = queue.poll()) != null) {
task.run();
}
} catch (RuntimeException | Error t1) {
toThrow = t1;
} finally {
if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
try {
delegate.execute(this::drain);
} catch (RuntimeException | Error t2) {
draining.set(false);
if (toThrow == null) {
toThrow = t2;
} else {
toThrow.addSuppressed(t2);
}
}
}
}
if (toThrow != null) {
LOGGER.warn("Error while draining event queue", toThrow);
if (toThrow instanceof RuntimeException) {
throw (RuntimeException) toThrow;
}
throw (Error) toThrow;
}
}
```
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3389346724
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
Oh just catch and not re-throw you mean? That would swallow `Error`s, not sure it's better.
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3389609645
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
Ah, sorry, I missed that we're not re-throwing here :thinking:
However, if an `Error` occurs, it will mask the original exception anyway, whether we re-throw it or just ignore :thinking:
Where do task execution exceptions go in this case? :thinking:
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on PR #4648:
URL: https://github.com/apache/polaris/pull/4648#issuecomment-4663728069
Interesting: `InMemoryBufferEventListenerBufferSizeTest#testProcessorFailureRecovery` just failed in CI. Probably unrelated to the changes in this branch. I'm investigating.
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on PR #4648:
URL: https://github.com/apache/polaris/pull/4648#issuecomment-4649913862
Relevant ML discussion:
https://lists.apache.org/thread/yhs40z7r90mdpqbfzpwhqgxdrd8pln96
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3395627890
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
> This is why I was wondering about reusing Mutiny of some other framework here ๐
Any async framework could be leveraged to coordinate failure handling in a more composable way, for sure. The only caveat is that `Error`s are not propagated, since the error handling is self-contained.
Here is a version with `CompletableFuture`:
```java
@Override
@SuppressWarnings("FutureReturnValueIgnored")
public void execute(@NonNull Runnable task) {
if (!queue.offer(task)) {
throw new RejectedExecutionException("Event listener queue is full");
}
if (draining.compareAndSet(false, true)) {
try {
scheduleDrain();
} catch (Throwable t) {
draining.set(false);
throw t;
}
}
}
private CompletableFuture<Void> scheduleDrain() {
return CompletableFuture.runAsync(this::drain, delegate)
.thenCompose(this::maybeReschedule)
.exceptionally(this::onError);
}
private void drain() {
try {
Runnable task;
while ((task = queue.poll()) != null) {
task.run();
}
} finally {
draining.set(false);
}
}
private CompletableFuture<Void> maybeReschedule(Void unused) {
// A producer may have enqueued a task between the last poll() and draining.set(false) above,
// and seen draining=true so it did not schedule a drain. Catch that here.
if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
// re-schedule for fairness
return scheduleDrain(); // this may throw; it would fail the stage it's running in
}
return CompletableFuture.completedFuture(null);
}
private Void onError(Throwable error) {
// tasks remain in queue; next submission will trigger a new drain
draining.set(false);
LOGGER.warn("Failed to schedule listener executor drain; tasks may be delayed", error);
return null;
}
```
And here is the version with Mutiny:
```java
@Override
public void execute(@NonNull Runnable task) {
if (!queue.offer(task)) {
throw new RejectedExecutionException("Event listener queue is full");
}
if (draining.compareAndSet(false, true)) {
try {
scheduleDrain().subscribe().with(unused -> {});
} catch (Throwable t) {
draining.set(false);
throw t;
}
}
}
private Uni<Void> scheduleDrain() {
return Uni.createFrom()
.item(this::drain)
.runSubscriptionOn(delegate)
.call(this::maybeReschedule)
.onFailure()
.recoverWithItem(this::onError);
}
private Void drain() {
try {
Runnable task;
while ((task = queue.poll()) != null) {
task.run();
}
} finally {
draining.set(false);
}
return null;
}
private Uni<Void> maybeReschedule(Object unused) {
// A producer may have enqueued a task between the last poll() and draining.set(false) above,
// and seen draining=true so it did not schedule a drain. Catch that here.
if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
// re-schedule for fairness
return scheduleDrain();
}
return Uni.createFrom().voidItem();
}
private Void onError(Throwable error) {
// tasks remain in queue; next submission will trigger a new drain
draining.set(false);
LOGGER.error(
"Failed to schedule listener executor drain; event delivery may be delayed", error);
return null;
}
```
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3397285127
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
With `CompletableFuture` I guess we have to log task exceptions in our code, otherwise they will remain in the `CompletableFuture` undiscovered, because nobody waits on it, right? :thinking:
--
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] Enforce ordered per-listener event delivery [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4648:
URL: https://github.com/apache/polaris/pull/4648#discussion_r3395637278
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -136,4 +165,55 @@ private void deliverEvent(
public boolean hasListeners(PolarisEventType eventType) {
return eventTypesWithListeners.contains(eventType);
}
+
+ private static class ListenerExecutor implements Executor {
+
+ private final Executor delegate;
+ private final Queue<Runnable> queue;
+ private final AtomicBoolean draining = new AtomicBoolean(false);
+
+ ListenerExecutor(Executor delegate, int capacity) {
+ this.delegate = delegate;
+ this.queue =
+ capacity == -1 ? new ConcurrentLinkedQueue<>() : new ArrayBlockingQueue<>(capacity);
+ }
+
+ @Override
+ public void execute(@NonNull Runnable task) {
+ if (!queue.offer(task)) {
+ throw new RejectedExecutionException("Event listener queue is full");
+ }
+ if (draining.compareAndSet(false, true)) {
+ try {
+ delegate.execute(this::drain);
+ } catch (Throwable t) {
+ draining.set(false);
+ throw t;
+ }
+ }
+ }
+
+ private void drain() {
+ try {
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ draining.set(false);
+ // A producer may have enqueued a task between the last poll() and the set(false) above,
+ // and seen draining=true so it did not schedule a drain. Catch that here.
+ if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
+ try {
+ // re-schedule for fairness
+ delegate.execute(this::drain);
+ } catch (Exception e) {
Review Comment:
And now `VertxAsyncExec`. Switching to this component comes with a few caveats:
- It's shared with other parts of the application (NoSQL, OPA). Unless we create a separate bean, we'd have to share the underlying executor. But event delivery work may be quite intense depending on how much activity is received and how many listeners there are (and how slow they can be).
- Using `VertxAsyncExec.schedulePeriodic()` is tempting, but would move event delivery from a push model (deliver as soon as an event arrives) to a pure polling mechanism. I think we should stick with `VertxAsyncExce.schedule()`, but with a delay of zero.
Here is a version with `VertxAsyncExec`:
```java
@Override
public void execute(@NonNull Runnable task) {
if (!queue.offer(task)) {
throw new RejectedExecutionException("Event listener queue is full");
}
if (draining.compareAndSet(false, true)) {
asyncExec.schedule(this::drain, Duration.ZERO); // doesn't throw
}
}
private void drain() {
try {
Runnable task;
while ((task = queue.poll()) != null) {
task.run();
}
} finally {
draining.set(false);
if (!queue.isEmpty() && draining.compareAndSet(false, true)) {
asyncExec.schedule(this::drain, Duration.ZERO); // doesn't throw
}
}
}
```
--
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