jerryshao commented on code in PR #14023:
URL: https://github.com/apache/gravitino/pull/14023#discussion_r4332532574
##########
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:
Deliberate: the post-interrupt join keeps the wait `stop()` had before
draining was added, so the drain is an extra budget on top. The comment now
says a stuck dispatcher delays `stop()` by at most twice
`dispatcherJoinSeconds`, the PR description says the same, and
`testStopInterruptsDispatcherThatDoesNotFinishInTime` pins the bound
(40131ad4a7).
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
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:
Done as suggested: `EventListenerManager` keeps a
`CopyOnWriteArrayList<AsyncQueueListener>` that `assembleEventListener` appends
to and hands to the `EventBus`, so `isHighWatermark()` no longer scans every
listener (40131ad4a7).
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
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:
Added a `Preconditions.checkState` on an `initialized` flag at the top of
`init()`, with `testInitTwiceFails` (40131ad4a7).
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
--
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]
