This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 34db649643 [Cherry-pick to branch-1.3] [#14022] fix(core): Honor the
declared mode of event listeners registered in code (#14023) (#14193)
34db649643 is described below
commit 34db649643ea6b43895495ebc47eb28eab11ae5a
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sun Oct 25 09:35:32 2026 +0800
[Cherry-pick to branch-1.3] [#14022] fix(core): Honor the declared mode of
event listeners registered in code (#14023) (#14193)
**Cherry-pick Information:**
- Original commit: 1d73fb1b7945279f79aa5e9d6bf5e7db8150f665
- Target branch: `branch-1.3`
- Status: ✅ Clean cherry-pick (no conflicts)
Co-authored-by: Jerry Shao <[email protected]>
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 c0222c2089..4fee8c7d27 100644
--- a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
+++ b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
@@ -670,32 +670,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<>();
