Copilot commented on code in PR #14023:
URL: https://github.com/apache/gravitino/pull/14023#discussion_r4324533673
##########
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:
After the second `join`, the dispatcher thread may still be alive. In that
case, `eventListeners.forEach(...stop)` can race with the dispatcher still
invoking `onPreEvent/onPostEvent` on those listeners. Consider re-checking
`asyncProcessor.isAlive()` after the second join and handling it explicitly
(e.g., log a warning and delay/stabilize shutdown semantics, or loop with a
bounded total wait) so inner listeners are not stopped while they may still be
in use.
##########
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:
`CopyOnWriteArrayList` copies the backing array on every write; with async
audit writes and tests pushing large counts (e.g., thousands), this can add
unnecessary overhead. Since this is a test utility primarily appending logs, a
concurrent append-friendly structure (e.g., `ConcurrentLinkedQueue`) or a
synchronized `ArrayList` can provide thread-safety with lower write
amplification; pick based on whether indexed access is actually required.
##########
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:
Switching from `queue.take()` to a fixed-interval `poll()` introduces
periodic wakeups even when idle (up to 10x/sec per async listener), which can
add avoidable CPU overhead at scale. A common alternative is to keep `take()`
and use a shutdown signal (e.g., a sentinel/poison event) or interrupt-only
shutdown while still ensuring drain semantics.
##########
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:
These assertions depend on exact dispatcher thread names, making the test
brittle to internal naming changes (or future suffixing/uniqueness changes).
Consider asserting on a stable property instead (e.g., thread name starts with
`async-queue-listener-` and contains the queue name), so the test continues to
validate dispatch mode without coupling to exact naming.
--
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]
