jerryshao commented on code in PR #14023:
URL: https://github.com/apache/gravitino/pull/14023#discussion_r4324690128
##########
core/src/main/java/org/apache/gravitino/listener/AsyncQueueListener.java:
##########
@@ -97,13 +98,22 @@ 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;
+ }
try {
+ // Give the dispatcher thread time to drain the queued events. It exits
by itself once the
+ // queue is empty, and is interrupted if it does not finish in time.
asyncProcessor.join(dispatcherJoinSeconds * 1000L);
+ if (asyncProcessor.isAlive()) {
+ asyncProcessor.interrupt();
+ asyncProcessor.join(dispatcherJoinSeconds * 1000L);
+ }
} catch (InterruptedException e) {
LOG.warn("{} interrupt async processor failed.", asyncQueueListenerName,
e);
+ Thread.currentThread().interrupt();
}
eventListeners.forEach(listenerPlugin -> listenerPlugin.stop());
}
Review Comment:
Added a check after the interrupt join: if the dispatcher thread is still
alive, `stop()` logs a warning and then stops the inner listeners anyway, since
their resources must be released on shutdown (ce2d900fde).
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
core/src/main/java/org/apache/gravitino/listener/AsyncQueueListener.java:
##########
@@ -120,7 +130,14 @@ public List<EventListenerPlugin> getEventListeners() {
private void processEvents() {
while (!Thread.currentThread().isInterrupted()) {
try {
- BaseEvent baseEvent = queue.take();
+ BaseEvent baseEvent = queue.poll(POLL_INTERVAL_MS,
TimeUnit.MILLISECONDS);
+ if (baseEvent == null) {
+ if (stopped.get()) {
+ // The listener is stopped and all the queued events are processed.
+ break;
+ }
+ continue;
+ }
Review Comment:
Switched back to the blocking `take()`. `stop()` now queues a stop signal
behind the pending events, and the dispatcher thread exits when it takes it, so
an idle thread no longer wakes up (ce2d900fde).
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
core/src/test/java/org/apache/gravitino/audit/DummyAuditWriter.java:
##########
@@ -19,14 +19,15 @@
package org.apache.gravitino.audit;
-import java.util.LinkedList;
+import java.util.List;
import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
import lombok.Getter;
public class DummyAuditWriter implements AuditLogWriter {
private Formatter formatter;
- @Getter private final LinkedList<DummyAuditLog> auditLogs = new
LinkedList<>();
+ @Getter private final List<DummyAuditLog> auditLogs = new
CopyOnWriteArrayList<>();
Review Comment:
Changed to `Collections.synchronizedList(new ArrayList<>())`. The test needs
indexed access for the ordered `get(0)`/`get(1)` checks (ce2d900fde).
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
core/src/test/java/org/apache/gravitino/listener/TestEventListenerManager.java:
##########
@@ -270,6 +332,116 @@ void testTransformPreEvent() {
Assertions.assertEquals(2, ((CountingPreEvent)
transformedEvent.get()).count());
}
+ @Test
+ void testAddEventListenerHonorsMode() {
+ EventListenerManager eventListenerManager = new EventListenerManager();
+ eventListenerManager.init(new HashMap<>());
+
+ ThreadRecordingEventListener syncListener = new
ThreadRecordingEventListener(Mode.SYNC);
+ ThreadRecordingEventListener isolatedListener =
+ new ThreadRecordingEventListener(Mode.ASYNC_ISOLATED);
+ ThreadRecordingEventListener sharedListener1 =
+ new ThreadRecordingEventListener(Mode.ASYNC_SHARED);
+ ThreadRecordingEventListener sharedListener2 =
+ new ThreadRecordingEventListener(Mode.ASYNC_SHARED);
+ eventListenerManager.addEventListener("sync", syncListener);
+ eventListenerManager.addEventListener("isolated", isolatedListener);
+ eventListenerManager.addEventListener("shared1", sharedListener1);
+ eventListenerManager.addEventListener("shared2", sharedListener2);
+ eventListenerManager.start();
+ EventBus eventBus = eventListenerManager.createEventBus();
+ try {
+ List<EventListenerPlugin> listeners = eventBus.getEventListeners();
+ Assertions.assertEquals(3, listeners.size());
+ Assertions.assertEquals(
+ 1,
listeners.stream().filter(EventListenerPluginWrapper.class::isInstance).count());
+ Set<Integer> queueSizes =
+ listeners.stream()
+ .filter(AsyncQueueListener.class::isInstance)
+ .map(listener -> ((AsyncQueueListener)
listener).getEventListeners().size())
+ .collect(Collectors.toSet());
+ Assertions.assertEquals(ImmutableSet.of(1, 2), queueSizes);
+
+ eventBus.dispatchEvent(DUMMY_POST_EVENT_INSTANCE);
+
+ Assertions.assertEquals(
+ ImmutableList.of(Thread.currentThread().getName()),
syncListener.awaitPostEventThreads());
+ Assertions.assertEquals(
+ ImmutableList.of("async-queue-listener-isolated"),
+ isolatedListener.awaitPostEventThreads());
+ Assertions.assertEquals(
+ ImmutableList.of("async-queue-listener-default"),
+ sharedListener1.awaitPostEventThreads());
+ Assertions.assertEquals(
+ ImmutableList.of("async-queue-listener-default"),
+ sharedListener2.awaitPostEventThreads());
Review Comment:
Kept the exact names on purpose. The thread name is what tells the isolated
queue (`async-queue-listener-<name>`) apart from the shared `default` queue,
and `AsyncQueueListener` is in the same package, so a naming change would
update this test in the same PR. Happy to relax it if a maintainer prefers.
_🤖 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]
