jerryshao commented on code in PR #14023:
URL: https://github.com/apache/gravitino/pull/14023#discussion_r4344242869
##########
core/src/main/java/org/apache/gravitino/listener/EventListenerManager.java:
##########
@@ -83,61 +90,73 @@ public void init(Map<String, String> properties) {
throw new IllegalStateException(
"Duplicate event listener name detected: " +
existingValue);
}));
- this.eventListeners = assembleEventListeners(userEventListenerPlugins);
+ userEventListenerPlugins.forEach(this::assembleEventListener);
}
- public void start() {
- eventListeners.stream().forEach(listener -> listener.start());
+ public synchronized void start() {
+ eventListeners.forEach(EventListenerPlugin::start);
+ started = true;
}
- public void stop() {
- eventListeners.stream().forEach(listener -> listener.stop());
+ public synchronized void stop() {
+ eventListeners.forEach(EventListenerPlugin::stop);
+ started = false;
}
public EventBus createEventBus() {
- return new EventBus(eventListeners);
+ return new EventBus(eventListeners, asyncQueueListeners);
}
- public void addEventListener(String listenerName, EventListenerPlugin
listener) {
- eventListeners.add(new EventListenerPluginWrapper(listenerName, listener));
+ /**
+ * Registers an event listener that is created in code rather than loaded
from the configuration.
+ * The listener is dispatched according to its {@link
EventListenerPlugin#mode()}, the same as the
+ * configured listeners. If the manager has already started, the listener is
started here.
+ *
+ * @param listenerName The name of the listener.
+ * @param listener The listener to register.
+ */
+ public synchronized void addEventListener(String listenerName,
EventListenerPlugin listener) {
+ assembleEventListener(listenerName, listener);
}
- private List<EventListenerPlugin> assembleEventListeners(
- Map<String, EventListenerPlugin> userEventListeners) {
- List<EventListenerPlugin> sharedQueueListeners = new ArrayList<>();
+ private void assembleEventListener(String listenerName, EventListenerPlugin
listener) {
+ EventListenerPluginWrapper wrapper = new
EventListenerPluginWrapper(listenerName, listener);
+ switch (listener.mode()) {
+ case SYNC:
+ addAndStartIfNeeded(eventListeners, wrapper);
+ break;
+ case ASYNC_ISOLATED:
+ addAsyncQueueListener(
+ new AsyncQueueListener(
+ ImmutableList.of(wrapper), listenerName, queueCapacity,
dispatcherJoinSeconds));
Review Comment:
Fixed in a503de62c2. `BuiltInJobTemplateEventListener` no longer uses an
async listener queue. `onPostEvent` only hands a `CreateMetalakeEvent` to the
listener's own single-thread executor, whose queue is unbounded (metalake
creation is rare), so no created metalake is dropped; the listener declares
`SYNC` since that hand-off is cheap. `stop()` finishes the pending
registrations before the entity store closes, and anything still pending after
its timeout is reconciled by `start()` on the next startup.
`testCreatedMetalakesAreNotDroppedWhileRegistrationIsBusy` covers your
scenario: it blocks the first registration, creates two more metalakes, and
checks all three get templates.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
core/src/main/java/org/apache/gravitino/listener/AsyncQueueListener.java:
##########
@@ -97,13 +99,35 @@ 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 up to dispatcherJoinSeconds to drain the
queued events. If it
+ // does not finish in time, interrupt it and wait up to another
dispatcherJoinSeconds for it
+ // to exit, which is how long stop() waited for the thread before
draining was added. So a
+ // stuck dispatcher delays stop() by at most twice dispatcherJoinSeconds.
+ if (asyncProcessor.isAlive()
+ && queue.offer(STOP_SIGNAL, remainingMillis(deadlineNanos),
TimeUnit.MILLISECONDS)) {
+ asyncProcessor.join(remainingMillis(deadlineNanos));
+ }
+ if (asyncProcessor.isAlive()) {
+ asyncProcessor.interrupt();
+ asyncProcessor.join(joinMillis);
+ }
+ if (asyncProcessor.isAlive()) {
+ LOG.warn(
+ "{} event dispatcher thread is still running after it was
interrupted, stop the "
+ + "listeners anyway.",
+ asyncQueueListenerName);
+ }
} catch (InterruptedException e) {
LOG.warn("{} interrupt async processor failed.", asyncQueueListenerName,
e);
+ Thread.currentThread().interrupt();
Review Comment:
Fixed in a503de62c2. An interrupt during the drain now only skips the drain:
`stop()` still interrupts the dispatcher and waits for it with
`Uninterruptibles.joinUninterruptibly` (bounded by `dispatcherJoinSeconds`),
then stops the listeners, and the caller keeps its interrupt status.
`testStopWhenCallerIsInterrupted` interrupts the caller before `stop()` and
checks the dispatcher thread is gone.
_🤖 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]
