mchades commented on code in PR #14023:
URL: https://github.com/apache/gravitino/pull/14023#discussion_r4344200531
##########
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:
[P2] Interrupt the dispatcher when the stopping thread is interrupted
If the caller is already interrupted, the timed `queue.offer(STOP_SIGNAL,
...)` throws before enqueueing the signal. This catch only re-interrupts the
caller, so the dispatcher stays blocked on `take()` while its listeners are
stopped; subsequent `stop()` calls cannot recover because `stopped` is already
true. Reproduced by interrupting the calling thread before `stop()` and
checking that the dispatcher remains alive. Please also interrupt the
dispatcher and perform bounded cleanup on this path.
##########
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:
[P2] Preserve built-in template registration when the queue is full
This also puts `BuiltInJobTemplateEventListener` on a queue whose `offer()`
drops events when full. Dropping a `CreateMetalakeEvent` leaves a successfully
created metalake without built-in templates until restart; there is no retry or
later reconciliation. Reproduced with `queueCapacity=2`: block the first
template load, then create two more metalakes. All creates succeed, but after
draining, the third has zero templates. Please provide reliable delivery or a
retry/reconciliation path for this listener.
--
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]
