This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 1d73fb1b79 [#14022] fix(core): Honor the declared mode of event
listeners registered in code (#14023)
1d73fb1b79 is described below
commit 1d73fb1b7945279f79aa5e9d6bf5e7db8150f665
Author: Jerry Shao <[email protected]>
AuthorDate: Fri Oct 23 12:03:30 2026 +0800
[#14022] fix(core): Honor the declared mode of event listeners registered
in code (#14023)
### What changes were proposed in this pull request?
- `EventListenerManager.addEventListener` dispatches a listener
according to its `mode()`, the same as listeners loaded from
`gravitino.eventListener.names`:
- `SYNC`: plain wrapper
- `ASYNC_ISOLATED`: its own `AsyncQueueListener`
- `ASYNC_SHARED`: the shared queue
- A listener added after `start()` is started right away. The listener
lists are copy-on-write, so they are safe to read while listeners are
added.
- `EventBus.isHighWatermark` scans the current listeners on each call,
so async queues added after the bus was created are included.
- `AsyncQueueListener.stop` is idempotent, still stops the dispatcher
thread when the calling thread is interrupted, and it lets the
dispatcher thread finish the queued events. It interrupts the thread
only if it does not finish within `dispatcherJoinSeconds`, and then
waits up to another `dispatcherJoinSeconds` for the thread to exit, as
it did before. The dispatcher still blocks on `take()`; `stop` queues a
stop signal behind the pending events.
- `EventListenerManager.init` can only be called once, since the
listener lists are only appended to.
- `GravitinoEnv.shutdown` stops the auxiliary services and then the
event listeners, before it closes the entity store and the catalogs.
Queued audit logs and template registrations are written before the
store closes.
- `BuiltInJobTemplateEventListener` hands created metalakes to its own
single-thread executor with an unbounded queue, so a created metalake is
never dropped the way a full async listener queue would drop it. It
declares `SYNC`, since `onPostEvent` only hands the event over, and
`stop()` finishes the pending registrations. The listener captures the
created metalake's id with the event, and the queued registration looks
the metalake up by that id under the root read lock, so it follows a
rename and skips a dropped or re-created metalake. It skips a disabled
metalake while holding its read lock, and the listener also reconciles
the templates on `EnableMetalakeEvent`, so a metalake disabled before
its registration ran gets its templates once it is enabled. The
registration runs as the user from the event, so the templates still
record the metalake creator, not `anonymous`.
### Why are the changes needed?
`addEventListener` ignored `mode()`. The built-in job template listener
and the audit log listener both declared `ASYNC_ISOLATED`, but both ran
on the request thread. Every create-metalake request loaded the built-in
templates from `auxlib` and wrote them to the entity store before it
returned. With audit log enabled, every request also wrote its audit
entry before it returned.
Fix: #14022
### Does this PR introduce _any_ user-facing change?
- Built-in job templates are registered shortly after `createMetalake`
returns, not before it returns.
- Audit log entries follow the async queue semantics: they can be
dropped when the queue is full, the same as configured async listeners.
- On shutdown, async listeners finish their queued events instead of
dropping them. A stuck dispatcher delays shutdown by at most twice
`dispatcherJoinSeconds` per async queue.
### How was this patch tested?
- New tests in `TestEventListenerManager`: mode handling for listeners
added in code, listeners added after start (including the first shared
listener), the high watermark of a queue added after the bus was
created, and a second `init`.
- New tests in `TestAsyncQueueListener`: idempotent stop, stopping when
the caller is interrupted, draining queued events on stop, and
interrupting a dispatcher that does not finish in time, within the
expected time bound.
- New `TestGravitinoEnvShutdown`: an async listener drains before the
entity store closes. It fails with the old shutdown order.
- `TestBuiltInJobTemplateEventListener` checks the template creator,
that metalakes created while a registration is busy all get their
templates, that a queued registration follows a rename and skips a
dropped and re-created metalake, that a metalake disabled before its
registration gets its templates when it is enabled, and that events
after `stop()` are skipped.
- `TestAuditManager` waits for the async writes and stops its
dispatchers after each test.
- `./gradlew :core:test :server:test :lineage:test -PskipITs`
- `JobIT`, `JobAuthorizationIT`, `MetalakeIT`,
`MetalakeAuthorizationIT`, `AuditIT`, `HttpAuditFilterIT`,
`HttpAuditAuthzDenialIT`
- Ran the changed test classes 20 times in a row with no failures.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../java/org/apache/gravitino/GravitinoEnv.java | 27 +--
.../job/BuiltInJobTemplateEventListener.java | 177 ++++++++++++++++--
.../gravitino/listener/AsyncQueueListener.java | 48 ++++-
.../org/apache/gravitino/listener/EventBus.java | 18 +-
.../gravitino/listener/EventListenerManager.java | 109 ++++++-----
.../apache/gravitino/TestGravitinoEnvShutdown.java | 104 +++++++++++
.../apache/gravitino/audit/DummyAuditWriter.java | 7 +-
.../apache/gravitino/audit/TestAuditManager.java | 28 +++
.../job/TestBuiltInJobTemplateEventListener.java | 184 +++++++++++++++++-
.../gravitino/listener/TestAsyncQueueListener.java | 120 ++++++++++++
.../listener/TestEventListenerManager.java | 206 +++++++++++++++++++++
11 files changed, 933 insertions(+), 95 deletions(-)
diff --git a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
index 482e42fa28..3d17fb2b3e 100644
--- a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
+++ b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
@@ -814,32 +814,35 @@ public class GravitinoEnv {
public void shutdown() {
LOG.info("Shutting down Gravitino Environment...");
- if (entityStore != null) {
+ // Stop the auxiliary services first, since they share the event bus. Then
stop the event
+ // listeners, so the async listeners finish the queued events, such as
audit logs and built-in
+ // job template registration, before the entity store and the catalogs are
closed.
+ if (auxServiceManager != null) {
try {
- entityStore.close();
+ auxServiceManager.serviceStop();
} catch (Exception e) {
- LOG.warn("Failed to close EntityStore.", e);
+ LOG.warn("Failed to stop AuxServiceManager", e);
}
}
- if (catalogManager != null) {
- catalogManager.close();
+ if (eventListenerManager != null) {
+ eventListenerManager.stop();
}
- if (auxServiceManager != null) {
+ if (entityStore != null) {
try {
- auxServiceManager.serviceStop();
+ entityStore.close();
} catch (Exception e) {
- LOG.warn("Failed to stop AuxServiceManager", e);
+ LOG.warn("Failed to close EntityStore.", e);
}
}
- if (metricsSystem != null) {
- metricsSystem.close();
+ if (catalogManager != null) {
+ catalogManager.close();
}
- if (eventListenerManager != null) {
- eventListenerManager.stop();
+ if (metricsSystem != null) {
+ metricsSystem.close();
}
if (metalakeManager != null) {
diff --git
a/core/src/main/java/org/apache/gravitino/job/BuiltInJobTemplateEventListener.java
b/core/src/main/java/org/apache/gravitino/job/BuiltInJobTemplateEventListener.java
index 48568e3687..5613f83f93 100644
---
a/core/src/main/java/org/apache/gravitino/job/BuiltInJobTemplateEventListener.java
+++
b/core/src/main/java/org/apache/gravitino/job/BuiltInJobTemplateEventListener.java
@@ -21,6 +21,7 @@ package org.apache.gravitino.job;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.io.File;
import java.io.IOException;
import java.net.MalformedURLException;
@@ -32,6 +33,10 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.ServiceLoader;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
@@ -39,14 +44,18 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityStore;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.UserPrincipal;
import org.apache.gravitino.exceptions.JobTemplateAlreadyExistsException;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.listener.api.EventListenerPlugin;
import org.apache.gravitino.listener.api.event.CreateMetalakeEvent;
+import org.apache.gravitino.listener.api.event.EnableMetalakeEvent;
import org.apache.gravitino.listener.api.event.Event;
import org.apache.gravitino.lock.LockType;
import org.apache.gravitino.lock.TreeLockUtils;
import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.BaseMetalake;
import org.apache.gravitino.meta.JobTemplateEntity;
import org.apache.gravitino.metalake.MetalakeManager;
import org.apache.gravitino.storage.IdGenerator;
@@ -59,8 +68,8 @@ import org.slf4j.LoggerFactory;
/**
* Event listener that automatically registers built-in job templates when
metalakes are created.
*
- * <p>This listener monitors metalake creation events and registers all
discovered built-in job
- * templates (via JobTemplateProvider SPI) into the newly created metalake. It
also handles
+ * <p>This listener monitors metalake creation and enablement events and
registers all discovered
+ * built-in job templates (via JobTemplateProvider SPI) into the metalake. It
also handles
* registration for existing metalakes on first startup.
*/
public class BuiltInJobTemplateEventListener implements EventListenerPlugin {
@@ -72,15 +81,38 @@ public class BuiltInJobTemplateEventListener implements
EventListenerPlugin {
private static final Pattern VERSION_PATTERN =
Pattern.compile(JobTemplateProvider.VERSION_VALUE_PATTERN);
+ private static final long REGISTRATION_STOP_TIMEOUT_SECONDS = 10;
+
private final JobManager jobManager;
private final EntityStore entityStore;
private final IdGenerator idGenerator;
+ // Registers the templates of created metalakes off the request thread. Its
queue is unbounded,
+ // so a created metalake never misses its templates; metalakes are created
rarely.
+ private final ExecutorService registrationExecutor;
public BuiltInJobTemplateEventListener(
JobManager jobManager, EntityStore entityStore, IdGenerator idGenerator)
{
+ this(
+ jobManager,
+ entityStore,
+ idGenerator,
+ Executors.newSingleThreadExecutor(
+ new ThreadFactoryBuilder()
+ .setDaemon(true)
+ .setNameFormat("builtin-job-template-registration")
+ .build()));
+ }
+
+ @VisibleForTesting
+ BuiltInJobTemplateEventListener(
+ JobManager jobManager,
+ EntityStore entityStore,
+ IdGenerator idGenerator,
+ ExecutorService registrationExecutor) {
this.jobManager = jobManager;
this.entityStore = entityStore;
this.idGenerator = idGenerator;
+ this.registrationExecutor = registrationExecutor;
}
@Override
@@ -123,34 +155,58 @@ public class BuiltInJobTemplateEventListener implements
EventListenerPlugin {
@Override
public void stop() throws RuntimeException {
- // No resources to clean up
+ // Finish the pending registrations. Any that do not finish in time are
registered by start()
+ // on the next startup, which reconciles all the metalakes.
+ registrationExecutor.shutdown();
+ try {
+ if (!registrationExecutor.awaitTermination(
+ REGISTRATION_STOP_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+ LOG.warn(
+ "Built-in job template registration did not finish within {}
seconds, stop it",
+ REGISTRATION_STOP_TIMEOUT_SECONDS);
+ registrationExecutor.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ registrationExecutor.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
}
@Override
public void onPostEvent(Event postEvent) throws RuntimeException {
- if (postEvent instanceof CreateMetalakeEvent) {
- CreateMetalakeEvent event = (CreateMetalakeEvent) postEvent;
- String metalakeName = event.identifier().name();
-
- try {
- Map<String, JobTemplate> builtInTemplates = loadBuiltInJobTemplates();
- if (builtInTemplates.isEmpty()) {
- LOG.debug("No built-in job templates to register for metalake: {}",
metalakeName);
- return;
- }
+ // Register the templates when a metalake is created, and again when it is
enabled: a
+ // registration that runs while the metalake is disabled is skipped.
Reconciling is idempotent.
+ if (!(postEvent instanceof CreateMetalakeEvent)
+ && !(postEvent instanceof EnableMetalakeEvent)) {
+ return;
+ }
- reconcileBuiltInJobTemplates(metalakeName, builtInTemplates);
- LOG.info("Registered built-in job templates for metalake: {}",
metalakeName);
- } catch (Exception e) {
- LOG.error("Failed to register built-in job templates for metalake:
{}", metalakeName, e);
- }
+ String metalakeName = postEvent.identifier().name();
+ String user = postEvent.user();
+ // Capture the id of the metalake now. By the time the registration runs,
the metalake may be
+ // renamed, or dropped and re-created under the same name, so it is looked
up by id.
+ Optional<Long> metalakeId = loadMetalakeId(metalakeName);
+ if (!metalakeId.isPresent()) {
+ return;
+ }
+
+ try {
+ registrationExecutor.execute(
+ () -> registerBuiltInJobTemplates(metalakeId.get(), metalakeName,
user));
+ } catch (RejectedExecutionException e) {
+ LOG.warn(
+ "Skip registering built-in job templates for metalake {} since the
listener is stopped, "
+ + "they will be registered on the next startup",
+ metalakeName);
}
}
@Override
public Mode mode() {
- // Use async isolated to avoid blocking metalake creation
- return Mode.ASYNC_ISOLATED;
+ // onPostEvent only hands created metalakes to the registration executor,
which is cheap, so it
+ // runs on the caller thread. Unlike an async listener queue, the executor
never drops a
+ // created metalake when it is busy.
+ return Mode.SYNC;
}
@VisibleForTesting
@@ -246,6 +302,87 @@ public class BuiltInJobTemplateEventListener implements
EventListenerPlugin {
}
}
+ private Optional<Long> loadMetalakeId(String metalakeName) {
+ NameIdentifier ident = NameIdentifierUtil.ofMetalake(metalakeName);
+ try {
+ BaseMetalake metalake =
+ TreeLockUtils.doWithTreeLock(
+ ident,
+ LockType.READ,
+ () -> entityStore.get(ident, Entity.EntityType.METALAKE,
BaseMetalake.class));
+ return Optional.of(metalake.id());
+ } catch (Exception e) {
+ LOG.warn(
+ "Skip registering built-in job templates for metalake {} since it
cannot be loaded, "
+ + "they will be registered on the next startup",
+ metalakeName,
+ e);
+ return Optional.empty();
+ }
+ }
+
+ private void registerBuiltInJobTemplates(long metalakeId, String eventName,
String user) {
+ try {
+ Map<String, JobTemplate> builtInTemplates = loadBuiltInJobTemplates();
+ if (builtInTemplates.isEmpty()) {
+ LOG.debug("No built-in job templates to register for metalake: {}",
eventName);
+ return;
+ }
+
+ // Renaming or dropping a metalake takes the root write lock, so holding
the root read lock
+ // keeps the metalake found by id from being renamed or dropped during
the registration.
+ TreeLockUtils.doWithRootTreeLock(
+ LockType.READ,
+ () -> {
+ Optional<String> currentName = currentMetalakeName(metalakeId);
+ if (!currentName.isPresent()) {
+ LOG.info(
+ "Skip registering built-in job templates for metalake {}
since it was dropped",
+ eventName);
+ return null;
+ }
+
+ // Disabling or enabling a metalake takes its write lock, so it
stays in the same state
+ // while its templates are registered.
+ NameIdentifier ident =
NameIdentifierUtil.ofMetalake(currentName.get());
+ return TreeLockUtils.doWithTreeLock(
+ ident,
+ LockType.READ,
+ () -> {
+ if (!MetalakeManager.metalakeInUse(entityStore, ident)) {
+ LOG.info(
+ "Skip registering built-in job templates for metalake
{} since it is "
+ + "disabled, they will be registered when it is
enabled",
+ currentName.get());
+ return null;
+ }
+
+ // The registration thread has no caller principal. Run as
the user from the
+ // event, so the templates record that user as the creator.
+ PrincipalUtils.doAs(
+ new UserPrincipal(user),
+ () -> {
+ reconcileBuiltInJobTemplates(currentName.get(),
builtInTemplates);
+ return null;
+ });
+ LOG.info("Registered built-in job templates for metalake:
{}", currentName.get());
+ return null;
+ });
+ });
+ } catch (Exception e) {
+ LOG.error("Failed to register built-in job templates for metalake: {}",
eventName, e);
+ }
+ }
+
+ private Optional<String> currentMetalakeName(long metalakeId) throws
IOException {
+ return entityStore
+ .list(Namespace.empty(), BaseMetalake.class,
Entity.EntityType.METALAKE)
+ .stream()
+ .filter(metalake -> metalake.id() == metalakeId)
+ .map(BaseMetalake::name)
+ .findFirst();
+ }
+
private boolean isValidBuiltInJobTemplate(JobTemplate jobTemplate) {
if (!isValidBuiltInName(jobTemplate.name())) {
return false;
diff --git
a/core/src/main/java/org/apache/gravitino/listener/AsyncQueueListener.java
b/core/src/main/java/org/apache/gravitino/listener/AsyncQueueListener.java
index fd5387b40b..d25e6a377d 100644
--- a/core/src/main/java/org/apache/gravitino/listener/AsyncQueueListener.java
+++ b/core/src/main/java/org/apache/gravitino/listener/AsyncQueueListener.java
@@ -20,12 +20,13 @@
package org.apache.gravitino.listener;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.Uninterruptibles;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.gravitino.listener.api.EventListenerPlugin;
@@ -45,6 +46,8 @@ public class AsyncQueueListener implements
EventListenerPlugin {
private static final Logger LOG =
LoggerFactory.getLogger(AsyncQueueListener.class);
private static final String NAME_PREFIX = "async-queue-listener-";
private static final float HIGH_WATERMARK_RATIO = 0.9f;
+ // Queued by stop() behind the pending events. The dispatcher thread exits
when it takes it.
+ private static final BaseEvent STOP_SIGNAL = new BaseEvent(null, null) {};
private final List<EventListenerPlugin> eventListeners;
private final BlockingQueue<BaseEvent> queue;
@@ -97,13 +100,36 @@ public class AsyncQueueListener implements
EventListenerPlugin {
@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);
+ // 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.
try {
- asyncProcessor.join(dispatcherJoinSeconds * 1000L);
+ if (asyncProcessor.isAlive()
+ && queue.offer(STOP_SIGNAL, remainingMillis(deadlineNanos),
TimeUnit.MILLISECONDS)) {
+ asyncProcessor.join(remainingMillis(deadlineNanos));
+ }
} catch (InterruptedException e) {
- LOG.warn("{} interrupt async processor failed.", asyncQueueListenerName,
e);
+ // Skip the drain, but still stop the dispatcher thread below. The
interrupt status is kept.
+ LOG.warn("{} is interrupted while draining the queued events.",
asyncQueueListenerName);
+ Thread.currentThread().interrupt();
+ }
+ if (asyncProcessor.isAlive()) {
+ asyncProcessor.interrupt();
+ Uninterruptibles.joinUninterruptibly(asyncProcessor, joinMillis,
TimeUnit.MILLISECONDS);
+ }
+ if (asyncProcessor.isAlive()) {
+ LOG.warn(
+ "{} event dispatcher thread is still running after it was
interrupted, stop the "
+ + "listeners anyway.",
+ asyncQueueListenerName);
}
eventListeners.forEach(listenerPlugin -> listenerPlugin.stop());
}
@@ -121,6 +147,9 @@ public class AsyncQueueListener implements
EventListenerPlugin {
while (!Thread.currentThread().isInterrupted()) {
try {
BaseEvent baseEvent = queue.take();
+ if (baseEvent == STOP_SIGNAL) {
+ break;
+ }
if (baseEvent instanceof PreEvent) {
this.eventListeners.forEach(listener ->
listener.onPreEvent((PreEvent) baseEvent));
} else if (baseEvent instanceof Event) {
@@ -138,12 +167,17 @@ public class AsyncQueueListener implements
EventListenerPlugin {
if (!queue.isEmpty()) {
LOG.warn(
- "{} drop {} events since dispatch thread is interrupted",
+ "{} drop {} events since dispatch thread is stopped",
asyncQueueListenerName,
queue.size());
}
}
+ // Never returns 0, since Thread.join(0) waits forever.
+ private static long remainingMillis(long deadlineNanos) {
+ return Math.max(1, TimeUnit.NANOSECONDS.toMillis(deadlineNanos -
System.nanoTime()));
+ }
+
private void logDropEventsIfNecessary() {
long dropEventsToLog = 0;
Instant previousRecordTime = null;
diff --git a/core/src/main/java/org/apache/gravitino/listener/EventBus.java
b/core/src/main/java/org/apache/gravitino/listener/EventBus.java
index 33de94ce42..387b31a381 100644
--- a/core/src/main/java/org/apache/gravitino/listener/EventBus.java
+++ b/core/src/main/java/org/apache/gravitino/listener/EventBus.java
@@ -65,12 +65,24 @@ public class EventBus {
* with this EventBus for event dispatch.
*/
public EventBus(List<EventListenerPlugin> eventListeners) {
- this.eventListeners = eventListeners;
- this.asyncQueueListeners =
+ this(
+ eventListeners,
eventListeners.stream()
.filter(AsyncQueueListener.class::isInstance)
.map(AsyncQueueListener.class::cast)
- .collect(Collectors.toList());
+ .collect(Collectors.toList()));
+ }
+
+ /**
+ * Constructs an EventBus with the event listeners and the async queues
among them. The lists are
+ * kept by reference, so listeners the caller adds to them later are
dispatched and checked too.
+ *
+ * @param eventListeners The listeners to dispatch events to.
+ * @param asyncQueueListeners The {@link AsyncQueueListener} instances in
{@code eventListeners}.
+ */
+ EventBus(List<EventListenerPlugin> eventListeners, List<AsyncQueueListener>
asyncQueueListeners) {
+ this.eventListeners = eventListeners;
+ this.asyncQueueListeners = asyncQueueListeners;
}
/**
diff --git
a/core/src/main/java/org/apache/gravitino/listener/EventListenerManager.java
b/core/src/main/java/org/apache/gravitino/listener/EventListenerManager.java
index 224d4e2d16..4cf157946a 100644
--- a/core/src/main/java/org/apache/gravitino/listener/EventListenerManager.java
+++ b/core/src/main/java/org/apache/gravitino/listener/EventListenerManager.java
@@ -23,10 +23,9 @@ import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableList;
-import java.util.ArrayList;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.listener.api.EventListenerPlugin;
@@ -59,9 +58,17 @@ public class EventListenerManager {
private int queueCapacity;
private int dispatcherJoinSeconds;
- private List<EventListenerPlugin> eventListeners;
+ // Copy-on-write lists, so listeners added after start() are safe to iterate
concurrently.
+ private final List<EventListenerPlugin> eventListeners = new
CopyOnWriteArrayList<>();
+ private final List<EventListenerPlugin> sharedQueueListeners = new
CopyOnWriteArrayList<>();
+ private final List<AsyncQueueListener> asyncQueueListeners = new
CopyOnWriteArrayList<>();
+ private boolean initialized = false;
+ private volatile boolean started = false;
- public void init(Map<String, String> properties) {
+ public synchronized void init(Map<String, String> properties) {
+ // The listener lists are only appended to, so a second init would
register them twice.
+ Preconditions.checkState(!initialized, "EventListenerManager has already
been initialized");
+ initialized = true;
EventListenerConfig config = new EventListenerConfig(properties);
this.queueCapacity = config.get(EventListenerConfig.QUEUE_CAPACITY);
this.dispatcherJoinSeconds =
config.get(EventListenerConfig.DISPATCHER_JOIN_SECONDS);
@@ -83,61 +90,73 @@ public class EventListenerManager {
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));
+ break;
+ case ASYNC_SHARED:
+ if (!sharedQueueListeners.isEmpty()) {
+ addAndStartIfNeeded(sharedQueueListeners, wrapper);
+ } else {
+ // Starting the shared queue starts the wrapper as well.
+ sharedQueueListeners.add(wrapper);
+ addAsyncQueueListener(
+ new AsyncQueueListener(
+ sharedQueueListeners, "default", queueCapacity,
dispatcherJoinSeconds));
+ }
+ break;
+ default:
+ throw new RuntimeException("Unexpected listener mode:" +
listener.mode());
+ }
+ }
- List<EventListenerPlugin> listeners =
- userEventListeners.entrySet().stream()
- .map(
- entrySet -> {
- String listenerName = entrySet.getKey();
- EventListenerPlugin listener = entrySet.getValue();
- switch (listener.mode()) {
- case SYNC:
- return new EventListenerPluginWrapper(listenerName,
listener);
- case ASYNC_ISOLATED:
- return new AsyncQueueListener(
- ImmutableList.of(new
EventListenerPluginWrapper(listenerName, listener)),
- listenerName,
- queueCapacity,
- dispatcherJoinSeconds);
- case ASYNC_SHARED:
- sharedQueueListeners.add(
- new EventListenerPluginWrapper(listenerName,
listener));
- return null;
- default:
- throw new RuntimeException("Unexpected listener mode:" +
listener.mode());
- }
- })
- .filter(Objects::nonNull)
- .collect(Collectors.toList());
+ private void addAsyncQueueListener(AsyncQueueListener asyncQueueListener) {
+ addAndStartIfNeeded(eventListeners, asyncQueueListener);
+ asyncQueueListeners.add(asyncQueueListener);
+ }
- if (!sharedQueueListeners.isEmpty()) {
- listeners.add(
- new AsyncQueueListener(
- sharedQueueListeners, "default", queueCapacity,
dispatcherJoinSeconds));
+ private void addAndStartIfNeeded(
+ List<EventListenerPlugin> listeners, EventListenerPlugin listener) {
+ if (started) {
+ listener.start();
}
- return listeners;
+ listeners.add(listener);
}
private EventListenerPlugin loadUserEventListenerPlugin(
diff --git
a/core/src/test/java/org/apache/gravitino/TestGravitinoEnvShutdown.java
b/core/src/test/java/org/apache/gravitino/TestGravitinoEnvShutdown.java
new file mode 100644
index 0000000000..bcaa838a43
--- /dev/null
+++ b/core/src/test/java/org/apache/gravitino/TestGravitinoEnvShutdown.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino;
+
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.auxiliary.AuxiliaryServiceManager;
+import org.apache.gravitino.listener.EventListenerManager;
+import org.apache.gravitino.listener.api.EventListenerPlugin;
+import org.apache.gravitino.listener.api.event.Event;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
+
+public class TestGravitinoEnvShutdown {
+
+ @Test
+ void testAsyncListenersDrainBeforeEntityStoreCloses() throws Exception {
+ AtomicBoolean entityStoreClosed = new AtomicBoolean(false);
+ EntityStore entityStore = mock(EntityStore.class);
+ doAnswer(
+ invocation -> {
+ entityStoreClosed.set(true);
+ return null;
+ })
+ .when(entityStore)
+ .close();
+ AuxiliaryServiceManager auxServiceManager =
mock(AuxiliaryServiceManager.class);
+
+ // A slow async listener that records whether the entity store was still
open while it ran.
+ AtomicInteger eventsSeenWithOpenStore = new AtomicInteger();
+ EventListenerManager eventListenerManager = new EventListenerManager();
+ eventListenerManager.init(new HashMap<>());
+ eventListenerManager.addEventListener(
+ "slow",
+ new EventListenerPlugin() {
+ @Override
+ public void init(Map<String, String> properties) {}
+
+ @Override
+ public void start() {}
+
+ @Override
+ public void stop() {}
+
+ @Override
+ public void onPostEvent(Event event) {
+ try {
+ Thread.sleep(200);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ if (!entityStoreClosed.get()) {
+ eventsSeenWithOpenStore.incrementAndGet();
+ }
+ }
+
+ @Override
+ public Mode mode() {
+ return Mode.ASYNC_ISOLATED;
+ }
+ });
+ eventListenerManager.start();
+ eventListenerManager.createEventBus().dispatchEvent(mock(Event.class));
+
+ TestGravitinoEnv env = new TestGravitinoEnv();
+ FieldUtils.writeField(env, "entityStore", entityStore, true);
+ FieldUtils.writeField(env, "auxServiceManager", auxServiceManager, true);
+ FieldUtils.writeField(env, "eventListenerManager", eventListenerManager,
true);
+
+ env.shutdown();
+
+ Assertions.assertEquals(1, eventsSeenWithOpenStore.get());
+ InOrder order = inOrder(auxServiceManager, entityStore);
+ order.verify(auxServiceManager).serviceStop();
+ order.verify(entityStore).close();
+ }
+
+ private static final class TestGravitinoEnv extends GravitinoEnv {}
+}
diff --git
a/core/src/test/java/org/apache/gravitino/audit/DummyAuditWriter.java
b/core/src/test/java/org/apache/gravitino/audit/DummyAuditWriter.java
index a6ca2d6276..7608efd1eb 100644
--- a/core/src/test/java/org/apache/gravitino/audit/DummyAuditWriter.java
+++ b/core/src/test/java/org/apache/gravitino/audit/DummyAuditWriter.java
@@ -19,14 +19,17 @@
package org.apache.gravitino.audit;
-import java.util.LinkedList;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
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 =
Collections.synchronizedList(new ArrayList<>());
@Override
public Formatter getFormatter() {
diff --git
a/core/src/test/java/org/apache/gravitino/audit/TestAuditManager.java
b/core/src/test/java/org/apache/gravitino/audit/TestAuditManager.java
index 21ec1ca1c6..6bdfc8456c 100644
--- a/core/src/test/java/org/apache/gravitino/audit/TestAuditManager.java
+++ b/core/src/test/java/org/apache/gravitino/audit/TestAuditManager.java
@@ -22,7 +22,10 @@ package org.apache.gravitino.audit;
import static org.apache.gravitino.audit.AuditLog.Operation;
import static org.apache.gravitino.audit.AuditLog.Status;
+import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
import org.apache.gravitino.Config;
import org.apache.gravitino.Configs;
import org.apache.gravitino.NameIdentifier;
@@ -31,6 +34,8 @@ import org.apache.gravitino.listener.EventBus;
import org.apache.gravitino.listener.EventListenerManager;
import org.apache.gravitino.listener.api.event.Event;
import org.apache.gravitino.listener.api.event.FailureEvent;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
@@ -40,6 +45,18 @@ public class TestAuditManager {
private static final int EVENT_NUM = 2000;
+ private final List<EventListenerManager> eventListenerManagers = new
ArrayList<>();
+
+ /**
+ * Stops the async audit log dispatchers, so they do not keep writing to the
shared audit logger
+ * after the test ends.
+ */
+ @AfterEach
+ public void tearDown() {
+ eventListenerManagers.forEach(EventListenerManager::stop);
+ eventListenerManagers.clear();
+ }
+
/** Test audit log with custom audit writer and formatter. */
@Test
public void testAuditLog() {
@@ -62,6 +79,8 @@ public class TestAuditManager {
DummyAuditFormatter.class,
(auditLogManager.getAuditLogWriter()).getFormatter());
DummyAuditWriter dummyAuditWriter = (DummyAuditWriter)
auditLogManager.getAuditLogWriter();
+ // The audit log listener is async, so wait for the dispatcher thread to
write the log.
+ awaitAuditLogs(dummyAuditWriter, 1);
DummyAuditFormatter formatter = (DummyAuditFormatter)
dummyAuditWriter.getFormatter();
DummyAuditLog formattedAuditLog = formatter.format(dummyEvent);
@@ -77,6 +96,7 @@ public class TestAuditManager {
// dispatch fail event
DummyFailEvent dummyFailEvent = mockDummyFailEvent();
eventBus.dispatchEvent(dummyFailEvent);
+ awaitAuditLogs(dummyAuditWriter, 2);
DummyAuditLog formattedFailAuditLog = formatter.format(dummyFailEvent);
Assertions.assertEquals(formattedFailAuditLog,
dummyAuditWriter.getAuditLogs().get(1));
Assertions.assertEquals(Operation.UNKNOWN_OPERATION,
formattedFailAuditLog.operation());
@@ -116,6 +136,13 @@ public class TestAuditManager {
// No assertion needed — we just verify no exception is thrown for a large
batch.
}
+ private void awaitAuditLogs(DummyAuditWriter writer, int expectedSize) {
+ Awaitility.await()
+ .atMost(20, TimeUnit.SECONDS)
+ .pollInterval(10, TimeUnit.MILLISECONDS)
+ .until(() -> writer.getAuditLogs().size() >= expectedSize);
+ }
+
private AuditLogManager mockAuditLogManager(
Config config, EventListenerManager eventListenerManager) {
AuditLogManager auditLogManager = new AuditLogManager();
@@ -127,6 +154,7 @@ public class TestAuditManager {
EventListenerManager eventListenerManager = new EventListenerManager();
eventListenerManager.init(new HashMap<>());
eventListenerManager.start();
+ eventListenerManagers.add(eventListenerManager);
return eventListenerManager;
}
diff --git
a/core/src/test/java/org/apache/gravitino/job/TestBuiltInJobTemplateEventListener.java
b/core/src/test/java/org/apache/gravitino/job/TestBuiltInJobTemplateEventListener.java
index c20cfcb8ea..1f2b97bbb8 100644
---
a/core/src/test/java/org/apache/gravitino/job/TestBuiltInJobTemplateEventListener.java
+++
b/core/src/test/java/org/apache/gravitino/job/TestBuiltInJobTemplateEventListener.java
@@ -34,6 +34,7 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.MoreExecutors;
import java.io.IOException;
import java.time.Instant;
import java.util.Arrays;
@@ -41,6 +42,11 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.ServiceLoader;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.Config;
@@ -50,6 +56,7 @@ import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.exceptions.JobTemplateAlreadyExistsException;
import org.apache.gravitino.listener.api.event.CreateMetalakeEvent;
+import org.apache.gravitino.listener.api.event.EnableMetalakeEvent;
import org.apache.gravitino.listener.api.event.Event;
import org.apache.gravitino.listener.api.info.MetalakeInfo;
import org.apache.gravitino.lock.LockManager;
@@ -60,11 +67,13 @@ import org.apache.gravitino.meta.SchemaVersion;
import org.apache.gravitino.storage.IdGenerator;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.memory.TestMemoryEntityStore;
+import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.NamespaceUtil;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
@@ -91,7 +100,10 @@ public class TestBuiltInJobTemplateEventListener {
idGenerator = new RandomIdGenerator();
jobManager = mock(JobManager.class);
- listener = new BuiltInJobTemplateEventListener(jobManager, entityStore,
idGenerator);
+ // Register on the test thread, since the ServiceLoader static mocks are
thread local.
+ listener =
+ new BuiltInJobTemplateEventListener(
+ jobManager, entityStore, idGenerator,
MoreExecutors.newDirectExecutorService());
}
@AfterEach
@@ -102,8 +114,9 @@ public class TestBuiltInJobTemplateEventListener {
}
@Test
- public void testOnPostCreateMetalakeEvent() {
+ public void testOnPostCreateMetalakeEvent() throws IOException {
String metalakeName = "test_metalake";
+ createMetalake(metalakeName, true);
NameIdentifier identifier = NameIdentifier.of(metalakeName);
MetalakeInfo metalakeInfo = new MetalakeInfo(metalakeName, "comment",
null, null);
@@ -143,11 +156,115 @@ public class TestBuiltInJobTemplateEventListener {
listener.onPostEvent(event);
- // Verify registerJobTemplate was called
- verify(jobManager, times(1)).registerJobTemplate(eq(metalakeName),
any());
+ // Verify registerJobTemplate was called, with the metalake creator as
the template creator
+ ArgumentCaptor<JobTemplateEntity> captor =
ArgumentCaptor.forClass(JobTemplateEntity.class);
+ verify(jobManager, times(1)).registerJobTemplate(eq(metalakeName),
captor.capture());
+ Assertions.assertEquals("user", captor.getValue().auditInfo().creator());
}
}
+ @Test
+ public void testCreatedMetalakesAreNotDroppedWhileRegistrationIsBusy()
throws Exception {
+ for (String metalake : Arrays.asList("metalake1", "metalake2",
"metalake3")) {
+ createMetalake(metalake, true);
+ }
+ CountDownLatch firstLoadStarted = new CountDownLatch(1);
+ CountDownLatch releaseFirstLoad = new CountDownLatch(1);
+ BuiltInJobTemplateEventListener busyListener =
+ newBlockingListener(firstLoadStarted, releaseFirstLoad);
+
when(jobManager.listJobTemplates(any())).thenReturn(Collections.emptyList());
+
+ // Block the first registration, then create more metalakes while it is
busy.
+ busyListener.onPostEvent(createMetalakeEvent("metalake1"));
+ Assertions.assertTrue(firstLoadStarted.await(30, TimeUnit.SECONDS));
+ busyListener.onPostEvent(createMetalakeEvent("metalake2"));
+ busyListener.onPostEvent(createMetalakeEvent("metalake3"));
+ releaseFirstLoad.countDown();
+
+ // Stop finishes the pending registrations.
+ busyListener.stop();
+
+ for (String metalake : Arrays.asList("metalake1", "metalake2",
"metalake3")) {
+ verify(jobManager, times(1)).registerJobTemplate(eq(metalake), any());
+ }
+ }
+
+ @Test
+ public void testRegistrationFollowsMetalakeRenamedWhileQueued() throws
Exception {
+ long metalakeId = createMetalake("metalake_a", true).id();
+ CountDownLatch loadStarted = new CountDownLatch(1);
+ CountDownLatch releaseLoad = new CountDownLatch(1);
+ BuiltInJobTemplateEventListener busyListener =
newBlockingListener(loadStarted, releaseLoad);
+
when(jobManager.listJobTemplates(any())).thenReturn(Collections.emptyList());
+
+ busyListener.onPostEvent(createMetalakeEvent("metalake_a"));
+ Assertions.assertTrue(loadStarted.await(30, TimeUnit.SECONDS));
+ // Rename metalake_a to metalake_b before the registration resumes.
+ entityStore.delete(NameIdentifier.of("metalake_a"),
Entity.EntityType.METALAKE);
+ putMetalake("metalake_b", metalakeId, true);
+ releaseLoad.countDown();
+ busyListener.stop();
+
+ verify(jobManager, times(1)).registerJobTemplate(eq("metalake_b"), any());
+ verify(jobManager, times(0)).registerJobTemplate(eq("metalake_a"), any());
+ }
+
+ @Test
+ public void testRegistrationSkipsMetalakeDroppedAndRecreatedWhileQueued()
throws Exception {
+ createMetalake("metalake_a", true);
+ CountDownLatch loadStarted = new CountDownLatch(1);
+ CountDownLatch releaseLoad = new CountDownLatch(1);
+ BuiltInJobTemplateEventListener busyListener =
newBlockingListener(loadStarted, releaseLoad);
+
+ busyListener.onPostEvent(createMetalakeEvent("metalake_a"));
+ Assertions.assertTrue(loadStarted.await(30, TimeUnit.SECONDS));
+ // Drop metalake_a and create another metalake with the same name before
the registration
+ // resumes. The old event must not populate the new metalake.
+ entityStore.delete(NameIdentifier.of("metalake_a"),
Entity.EntityType.METALAKE);
+ createMetalake("metalake_a", true);
+ releaseLoad.countDown();
+ busyListener.stop();
+
+ verify(jobManager, times(0)).listJobTemplates(any());
+ verify(jobManager, times(0)).registerJobTemplate(any(), any());
+ }
+
+ @Test
+ public void testRegistrationSkipsDisabledMetalakeAndRetriesOnEnable() throws
Exception {
+ long metalakeId = createMetalake("metalake_a", true).id();
+ CountDownLatch loadStarted = new CountDownLatch(1);
+ CountDownLatch releaseLoad = new CountDownLatch(1);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ BuiltInJobTemplateEventListener busyListener =
+ newBlockingListener(loadStarted, releaseLoad, executor);
+
when(jobManager.listJobTemplates(any())).thenReturn(Collections.emptyList());
+
+ busyListener.onPostEvent(createMetalakeEvent("metalake_a"));
+ Assertions.assertTrue(loadStarted.await(30, TimeUnit.SECONDS));
+ // Disable metalake_a before the registration resumes, then wait for the
registration to end.
+ replaceMetalake("metalake_a", metalakeId, false);
+ releaseLoad.countDown();
+ executor.submit(() -> {}).get(30, TimeUnit.SECONDS);
+ verify(jobManager, times(0)).registerJobTemplate(any(), any());
+
+ // Enabling metalake_a registers its templates.
+ replaceMetalake("metalake_a", metalakeId, true);
+ busyListener.onPostEvent(
+ new EnableMetalakeEvent("user",
NameIdentifierUtil.ofMetalake("metalake_a")));
+ busyListener.stop();
+
+ verify(jobManager, times(1)).registerJobTemplate(eq("metalake_a"), any());
+ }
+
+ @Test
+ public void testOnPostEventAfterStop() {
+ listener.stop();
+
+ // A metalake created after stop is skipped; start() registers its
templates on next startup.
+ Assertions.assertDoesNotThrow(() ->
listener.onPostEvent(createMetalakeEvent("metalake1")));
+ verify(jobManager, times(0)).listJobTemplates(any());
+ }
+
@Test
public void testOnPostEventWithNonMetalakeEvent() {
Event otherEvent = mock(Event.class);
@@ -566,13 +683,22 @@ public class TestBuiltInJobTemplateEventListener {
Assertions.assertEquals("builtin-existing", updated.name());
}
- private void createMetalake(String name, boolean inUse) throws IOException {
+ private BaseMetalake createMetalake(String name, boolean inUse) throws
IOException {
+ return putMetalake(name, idGenerator.nextId(), inUse);
+ }
+
+ private void replaceMetalake(String name, long id, boolean inUse) throws
IOException {
+ entityStore.delete(NameIdentifier.of(name), Entity.EntityType.METALAKE);
+ putMetalake(name, id, inUse);
+ }
+
+ private BaseMetalake putMetalake(String name, long id, boolean inUse) throws
IOException {
Map<String, String> props = new HashMap<>();
props.put(PROPERTY_IN_USE, String.valueOf(inUse));
BaseMetalake metalake =
BaseMetalake.builder()
- .withId(idGenerator.nextId())
+ .withId(id)
.withName(name)
.withComment("test metalake")
.withProperties(props)
@@ -582,6 +708,7 @@ public class TestBuiltInJobTemplateEventListener {
.build();
entityStore.put(metalake, false);
+ return metalake;
}
private JobTemplateEntity createJobTemplateEntity(String name, String
version) {
@@ -607,4 +734,49 @@ public class TestBuiltInJobTemplateEventListener {
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
.build();
}
+
+ /**
+ * Creates a listener with a real registration thread, whose first template
load blocks until
+ * {@code releaseFirstLoad} is counted down.
+ */
+ private BuiltInJobTemplateEventListener newBlockingListener(
+ CountDownLatch firstLoadStarted, CountDownLatch releaseFirstLoad) {
+ return newBlockingListener(
+ firstLoadStarted, releaseFirstLoad,
Executors.newSingleThreadExecutor());
+ }
+
+ private BuiltInJobTemplateEventListener newBlockingListener(
+ CountDownLatch firstLoadStarted,
+ CountDownLatch releaseFirstLoad,
+ ExecutorService registrationExecutor) {
+ ShellJobTemplate template =
+ ShellJobTemplate.builder()
+ .withName("builtin-test")
+ .withExecutable("/bin/echo")
+ .withCustomFields(Collections.singletonMap("version", "v1"))
+ .build();
+ AtomicInteger loads = new AtomicInteger();
+ return new BuiltInJobTemplateEventListener(
+ jobManager, entityStore, idGenerator, registrationExecutor) {
+ @Override
+ Map<String, JobTemplate> loadBuiltInJobTemplates() {
+ if (loads.incrementAndGet() == 1) {
+ firstLoadStarted.countDown();
+ try {
+ releaseFirstLoad.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ return Collections.singletonMap(template.name(), template);
+ }
+ };
+ }
+
+ private CreateMetalakeEvent createMetalakeEvent(String metalakeName) {
+ return new CreateMetalakeEvent(
+ "user",
+ NameIdentifier.of(metalakeName),
+ new MetalakeInfo(metalakeName, "comment", null, null));
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/listener/TestAsyncQueueListener.java
b/core/src/test/java/org/apache/gravitino/listener/TestAsyncQueueListener.java
index a5d578a7c0..b947b9244d 100644
---
a/core/src/test/java/org/apache/gravitino/listener/TestAsyncQueueListener.java
+++
b/core/src/test/java/org/apache/gravitino/listener/TestAsyncQueueListener.java
@@ -20,16 +20,21 @@
package org.apache.gravitino.listener;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.gravitino.listener.api.EventListenerPlugin;
import org.apache.gravitino.listener.api.event.Event;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
@@ -100,6 +105,121 @@ class TestAsyncQueueListener {
}
}
+ @Test
+ void testStopIsIdempotent() {
+ // Do not block the WARN log the dispatcher thread writes when it is
interrupted.
+ blockingAppender.releaseFirstLog.countDown();
+ EventListenerPlugin innerListener = mock(EventListenerPlugin.class);
+ AsyncQueueListener listener = new
AsyncQueueListener(List.of(innerListener), "test", 1, 1);
+ listener.start();
+
+ listener.stop();
+ listener.stop();
+
+ verify(innerListener, times(1)).stop();
+ }
+
+ @Test
+ void testStopWhenCallerIsInterrupted() {
+ blockingAppender.releaseFirstLog.countDown();
+ EventListenerPlugin innerListener = mock(EventListenerPlugin.class);
+ AsyncQueueListener listener =
+ new AsyncQueueListener(List.of(innerListener), "interrupted-caller",
10, 1);
+ listener.start();
+
+ Thread.currentThread().interrupt();
+ listener.stop();
+
+ // The caller keeps its interrupt status, and the dispatcher thread is
still stopped.
+ assertTrue(Thread.interrupted());
+ assertFalse(
+ Thread.getAllStackTraces().keySet().stream()
+ .anyMatch(
+ thread ->
+
thread.getName().equals("async-queue-listener-interrupted-caller")
+ && thread.isAlive()));
+ verify(innerListener, times(1)).stop();
+ }
+
+ @Test
+ void testStopDrainsQueuedEvents() {
+ blockingAppender.releaseFirstLog.countDown();
+ AtomicInteger processedEvents = new AtomicInteger();
+ EventListenerPlugin slowListener =
+ new EventListenerPlugin() {
+ @Override
+ public void init(Map<String, String> properties) {}
+
+ @Override
+ public void start() {}
+
+ @Override
+ public void stop() {}
+
+ @Override
+ public void onPostEvent(Event event) {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ processedEvents.incrementAndGet();
+ }
+ };
+ AsyncQueueListener listener = new
AsyncQueueListener(List.of(slowListener), "test", 10, 10);
+ listener.start();
+
+ Event event = mock(Event.class);
+ for (int i = 0; i < 5; i++) {
+ listener.onPostEvent(event);
+ }
+ listener.stop();
+
+ assertEquals(5, processedEvents.get());
+ }
+
+ @Test
+ void testStopInterruptsDispatcherThatDoesNotFinishInTime() throws Exception {
+ blockingAppender.releaseFirstLog.countDown();
+ CountDownLatch neverReleased = new CountDownLatch(1);
+ CountDownLatch entered = new CountDownLatch(1);
+ EventListenerPlugin blockedListener =
+ new EventListenerPlugin() {
+ @Override
+ public void init(Map<String, String> properties) {}
+
+ @Override
+ public void start() {}
+
+ @Override
+ public void stop() {}
+
+ @Override
+ public void onPostEvent(Event event) {
+ entered.countDown();
+ try {
+ neverReleased.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ };
+ AsyncQueueListener listener = new
AsyncQueueListener(List.of(blockedListener), "test", 10, 1);
+ listener.start();
+ listener.onPostEvent(mock(Event.class));
+ assertTrue(entered.await(30, TimeUnit.SECONDS));
+
+ long startNanos = System.nanoTime();
+ listener.stop();
+
+ // Stop waits at most twice dispatcherJoinSeconds: once to drain, once
after the interrupt.
+ // The dispatcher exits right after the interrupt, so stop() takes a
little over 1 second.
+ long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNanos);
+ assertTrue(elapsedMillis >= 900, "stop() returned before the drain
timeout: " + elapsedMillis);
+ assertTrue(elapsedMillis < 5000, "stop() took too long: " + elapsedMillis);
+ }
+
private static class BlockingAppender extends AbstractAppender {
private final AtomicInteger logCount = new AtomicInteger();
private final CountDownLatch firstLogEntered = new CountDownLatch(1);
diff --git
a/core/src/test/java/org/apache/gravitino/listener/TestEventListenerManager.java
b/core/src/test/java/org/apache/gravitino/listener/TestEventListenerManager.java
index 30faada6be..280860a9f1 100644
---
a/core/src/test/java/org/apache/gravitino/listener/TestEventListenerManager.java
+++
b/core/src/test/java/org/apache/gravitino/listener/TestEventListenerManager.java
@@ -19,22 +19,30 @@
package org.apache.gravitino.listener;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.exceptions.ForbiddenException;
import
org.apache.gravitino.listener.DummyEventListener.DummyAsyncEventListener;
import
org.apache.gravitino.listener.DummyEventListener.DummyAsyncIsolatedEventListener;
import org.apache.gravitino.listener.api.EventListenerPlugin;
+import org.apache.gravitino.listener.api.EventListenerPlugin.Mode;
import org.apache.gravitino.listener.api.event.BaseEvent;
import org.apache.gravitino.listener.api.event.Event;
import org.apache.gravitino.listener.api.event.OperationStatus;
import org.apache.gravitino.listener.api.event.PreEvent;
+import org.awaitility.Awaitility;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -59,6 +67,60 @@ public class TestEventListenerManager {
}
}
+ /** Records the threads that deliver post events, optionally blocking until
released. */
+ static class ThreadRecordingEventListener implements EventListenerPlugin {
+ private final Mode mode;
+ private final CountDownLatch release;
+ private final List<String> postEventThreads = new CopyOnWriteArrayList<>();
+ private final AtomicBoolean started = new AtomicBoolean(false);
+ private final AtomicInteger receivedEvents = new AtomicInteger(0);
+
+ ThreadRecordingEventListener(Mode mode) {
+ this(mode, new CountDownLatch(0));
+ }
+
+ ThreadRecordingEventListener(Mode mode, CountDownLatch release) {
+ this.mode = mode;
+ this.release = release;
+ }
+
+ @Override
+ public void init(Map<String, String> properties) {}
+
+ @Override
+ public void start() {
+ started.set(true);
+ }
+
+ @Override
+ public void stop() {}
+
+ @Override
+ public void onPostEvent(Event event) {
+ receivedEvents.incrementAndGet();
+ try {
+ release.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ postEventThreads.add(Thread.currentThread().getName());
+ }
+
+ @Override
+ public Mode mode() {
+ return mode;
+ }
+
+ List<String> awaitPostEventThreads() {
+ Awaitility.await()
+ .atMost(20, TimeUnit.SECONDS)
+ .pollInterval(10, TimeUnit.MILLISECONDS)
+ .until(() -> !postEventThreads.isEmpty());
+ return postEventThreads;
+ }
+ }
+
private static final DummyPostEvent DUMMY_POST_EVENT_INSTANCE =
new DummyPostEvent("user", NameIdentifier.of("a", "b"));
@@ -270,6 +332,150 @@ public class TestEventListenerManager {
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());
+ } finally {
+ eventListenerManager.stop();
+ }
+ }
+
+ @Test
+ void testAddEventListenerAfterStart() {
+ EventListenerManager eventListenerManager = new EventListenerManager();
+ eventListenerManager.init(createAsyncEventListenerConfig("async1",
"async2"));
+ eventListenerManager.start();
+ EventBus eventBus = eventListenerManager.createEventBus();
+
+ ThreadRecordingEventListener isolatedListener =
+ new ThreadRecordingEventListener(Mode.ASYNC_ISOLATED);
+ ThreadRecordingEventListener sharedListener =
+ new ThreadRecordingEventListener(Mode.ASYNC_SHARED);
+ try {
+ eventListenerManager.addEventListener("isolated", isolatedListener);
+ eventListenerManager.addEventListener("shared", sharedListener);
+ Assertions.assertTrue(isolatedListener.started.get());
+ Assertions.assertTrue(sharedListener.started.get());
+
+ // The shared listener joins the queue created for the configured shared
listeners.
+ Assertions.assertEquals(2, eventBus.getEventListeners().size());
+
+ eventBus.dispatchEvent(DUMMY_POST_EVENT_INSTANCE);
+ Assertions.assertEquals(
+ ImmutableList.of("async-queue-listener-isolated"),
+ isolatedListener.awaitPostEventThreads());
+ Assertions.assertEquals(
+ ImmutableList.of("async-queue-listener-default"),
sharedListener.awaitPostEventThreads());
+ } finally {
+ eventListenerManager.stop();
+ }
+ }
+
+ @Test
+ void testFirstSharedListenerAddedAfterStart() {
+ EventListenerManager eventListenerManager = new EventListenerManager();
+ eventListenerManager.init(new HashMap<>());
+ eventListenerManager.start();
+ EventBus eventBus = eventListenerManager.createEventBus();
+
+ ThreadRecordingEventListener sharedListener =
+ new ThreadRecordingEventListener(Mode.ASYNC_SHARED);
+ try {
+ // No shared queue exists yet, so registering the listener creates and
starts one.
+ eventListenerManager.addEventListener("shared", sharedListener);
+ Assertions.assertTrue(sharedListener.started.get());
+ Assertions.assertEquals(1, eventBus.getEventListeners().size());
+ Assertions.assertTrue(eventBus.getEventListeners().get(0) instanceof
AsyncQueueListener);
+
+ eventBus.dispatchEvent(DUMMY_POST_EVENT_INSTANCE);
+ Assertions.assertEquals(
+ ImmutableList.of("async-queue-listener-default"),
sharedListener.awaitPostEventThreads());
+ } finally {
+ eventListenerManager.stop();
+ }
+ }
+
+ @Test
+ void testInitTwiceFails() {
+ EventListenerManager eventListenerManager = new EventListenerManager();
+ eventListenerManager.init(createSyncEventListenerConfig("sync1", "sync2"));
+
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () -> eventListenerManager.init(createSyncEventListenerConfig("sync1",
"sync2")));
+ }
+
+ @Test
+ void testHighWatermarkIncludesListenerAddedAfterEventBusCreated() {
+ Map<String, String> properties = new HashMap<>();
+
properties.put(EventListenerManager.GRAVITINO_EVENT_LISTENER_QUEUE_CAPACITY,
"10");
+ EventListenerManager eventListenerManager = new EventListenerManager();
+ eventListenerManager.init(properties);
+ eventListenerManager.start();
+ EventBus eventBus = eventListenerManager.createEventBus();
+
+ CountDownLatch release = new CountDownLatch(1);
+ ThreadRecordingEventListener blockedListener =
+ new ThreadRecordingEventListener(Mode.ASYNC_ISOLATED, release);
+ eventListenerManager.addEventListener("blocked", blockedListener);
+ Assertions.assertFalse(eventBus.isHighWatermark());
+
+ try {
+ // Block the dispatcher thread on the first event, then fill the queue.
+ eventBus.dispatchEvent(DUMMY_POST_EVENT_INSTANCE);
+ Awaitility.await()
+ .atMost(20, TimeUnit.SECONDS)
+ .until(() -> blockedListener.receivedEvents.get() == 1);
+ for (int i = 0; i < 10; i++) {
+ eventBus.dispatchEvent(DUMMY_POST_EVENT_INSTANCE);
+ }
+ Assertions.assertTrue(eventBus.isHighWatermark());
+ } finally {
+ release.countDown();
+ eventListenerManager.stop();
+ }
+ }
+
private Map<String, String> createIsolatedAsyncEventListenerConfig(String
async1, String async2) {
Map<String, String> config = new HashMap<>();
