jerryshao opened a new pull request, #14023:
URL: https://github.com/apache/gravitino/pull/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, and it lets the dispatcher thread
finish the queued events. It interrupts the thread only if it does not finish
within `dispatcherJoinSeconds`.
- `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` 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 declare `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, waiting at most
`dispatcherJoinSeconds`, instead of dropping them.
### How was this patch tested?
- New tests in `TestEventListenerManager`: mode handling for listeners added
in code, listeners added after start, and the high watermark of a queue added
after the bus was created.
- New tests in `TestAsyncQueueListener`: idempotent stop, draining queued
events on stop, and interrupting a dispatcher that does not finish in time.
- `TestBuiltInJobTemplateEventListener` checks the template creator.
- `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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
