jerryshao commented on code in PR #14023:
URL: https://github.com/apache/gravitino/pull/14023#discussion_r4332504988
##########
core/src/main/java/org/apache/gravitino/listener/AsyncQueueListener.java:
##########
@@ -97,13 +99,33 @@ public void start() {
@Override
public void stop() {
- Preconditions.checkState(!stopped.get(), asyncQueueListenerName + " had
already stopped");
- stopped.compareAndSet(false, true);
- asyncProcessor.interrupt();
+ // Stop can be called more than once, e.g. by the server shutdown hook
after an explicit stop.
+ if (!stopped.compareAndSet(false, true)) {
+ LOG.debug("{} had already stopped", asyncQueueListenerName);
+ return;
+ }
+ long joinMillis = TimeUnit.SECONDS.toMillis(dispatcherJoinSeconds);
+ long deadlineNanos = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(joinMillis);
try {
- asyncProcessor.join(dispatcherJoinSeconds * 1000L);
+ // Give the dispatcher thread time to drain the queued events, then
interrupt it if it does
+ // not finish in time.
+ if (asyncProcessor.isAlive()
+ && queue.offer(STOP_SIGNAL, remainingMillis(deadlineNanos),
TimeUnit.MILLISECONDS)) {
+ asyncProcessor.join(remainingMillis(deadlineNanos));
+ }
+ if (asyncProcessor.isAlive()) {
+ asyncProcessor.interrupt();
+ asyncProcessor.join(joinMillis);
Review Comment:
[Question] `stop()` can wait up to twice `dispatcherJoinSeconds`, which is
not the bound the PR description and the comment above state.
`deadlineNanos` (:108) bounds the offer at :113 and the first join at :114,
so the drain phase is capped at `dispatcherJoinSeconds`. The join on this line
then passes the full `joinMillis` again rather than
`remainingMillis(deadlineNanos)`, so a dispatcher that overruns its drain
window costs `dispatcherJoinSeconds` to drain plus another
`dispatcherJoinSeconds` after the interrupt. `EventListenerManager.stop()`
(EventListenerManager.java:96-99) walks the listeners serially, so with the
audit-log and built-in-template listeners both `ASYNC_ISOLATED` plus any
configured ones, shutdown can stall for roughly `2 x 3s x <number of async
queues>` on the defaults -- precisely in the case where a listener is stuck and
you most want shutdown bounded.
The comment at :110-111 reads "interrupt it if it does not finish in time"
and the PR body says async listeners finish their queued events "waiting at
most `dispatcherJoinSeconds`", neither of which matches a 2x worst case. Is the
second full wait deliberate -- giving the interrupt its own grace period on top
of the drain budget? If so, saying that in the comment (and in the PR's
user-facing-change note) would settle it. If not, bounding this join by the
same deadline, or splitting `dispatcherJoinSeconds` between the drain and the
post-interrupt wait, keeps the configured value meaningful.
Verified by: read AsyncQueueListener.java:101-131 and `remainingMillis` at
:172-175 on ce2d900; confirmed `dispatcherJoinSeconds > 0` is enforced in
EventListenerConfig.java:44-50 so `joinMillis` is a full second or more;
confirmed `EventListenerManager.stop()` iterates listeners one at a time.
##########
core/src/main/java/org/apache/gravitino/listener/EventBus.java:
##########
@@ -112,7 +100,12 @@ public Optional<BaseEvent> dispatchEvent(BaseEvent
baseEvent) {
}
public boolean isHighWatermark() {
- return
asyncQueueListeners.stream().anyMatch(AsyncQueueListener::isHighWatermark);
+ // Scan the listeners on each call, so async queues registered after this
bus was created are
+ // also checked.
+ return eventListeners.stream()
+ .filter(AsyncQueueListener.class::isInstance)
+ .map(AsyncQueueListener.class::cast)
+ .anyMatch(AsyncQueueListener::isHighWatermark);
Review Comment:
[Nit] `isHighWatermark()` now allocates a stream over every listener on each
call, and it sits on the lineage ingestion path.
The precomputed `asyncQueueListeners` list this replaces was built once in
the constructor. The new version builds a stream with a filter, a map and an
`anyMatch` over all listeners (sync wrappers included) for every call, and
`LineageService.java:71` calls it per lineage event, so the cost lands per
event rather than per bus.
The listener count is small, so this is unlikely to show up in a profile,
but the previous shape is recoverable without giving up the fix: have
`EventListenerManager` keep a second `CopyOnWriteArrayList<AsyncQueueListener>`
that `assembleEventListener` appends to, and hand that to the `EventBus`
alongside `eventListeners`. A later-registered queue is then still visible,
with no per-call scan.
Verified by: read EventBus.java:95-109 and the removed constructor body in
the diff on ce2d900; grepped `isHighWatermark` call sites outside tests and
found LineageService.java:71 via LineageSinkManager.java:52-53.
##########
core/src/main/java/org/apache/gravitino/listener/EventListenerManager.java:
##########
@@ -83,61 +85,70 @@ public void init(Map<String, String> properties) {
throw new IllegalStateException(
"Duplicate event listener name detected: " +
existingValue);
}));
- this.eventListeners = assembleEventListeners(userEventListenerPlugins);
+ userEventListenerPlugins.forEach(this::assembleEventListener);
Review Comment:
[Nit] `init()` is now additive rather than replacing, so a second `init()`
double-registers every configured listener.
Before this change, `init()` ended with `this.eventListeners =
assembleEventListeners(...)`, which replaced whatever was there. Now
`eventListeners` and `sharedQueueListeners` are final `CopyOnWriteArrayList`
fields (:62-63) and this line appends into them, so calling `init()` twice on
one manager leaves two wrappers per configured listener, two isolated queues
per `ASYNC_ISOLATED` listener, and a `default` shared queue holding each shared
listener twice -- every event then reaching each listener twice.
`GravitinoEnv.java:886` calls it once, so nothing hits this today, and I
found no second call site. Still, since `addEventListener` is a public entry
point into the same lists, a cheap guard would make the invariant explicit: a
`Preconditions.checkState` on an `initialized` flag at the top of `init()`, or
clearing both lists before assembling.
Verified by: read EventListenerManager.java:58-99 and :117-151 on ce2d900
against the removed `assembleEventListeners` in the diff; grepped for
`eventListenerManager.init(` and `new EventListenerManager()` outside tests and
found only the single GravitinoEnv call site.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
