This is an automated email from the ASF dual-hosted git repository.
roryqi 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 4a53719d7b [#12722] improvement(iceberg): let inner dispatchers
contribute customInfo to Iceberg table events (#12723)
4a53719d7b is described below
commit 4a53719d7b8c0063297db180e90274d1abd85a83
Author: Nevin Zheng <[email protected]>
AuthorDate: Sat Aug 29 08:29:38 2026 -0700
[#12722] improvement(iceberg): let inner dispatchers contribute customInfo
to Iceberg table events (#12723)
## Summary
An inner Iceberg REST dispatcher layer can now attach `Map<String,
String>` audit facts to the terminal create/update/load event that
`IcebergTableEventDispatcher` already emits, so one operation still
produces one audit record and that record can say what the dispatch
chain knew about it.
This is the Iceberg REST counterpart of #12692. The change is
deliberately confined to carrying facts. Dispatch semantics are
untouched: events still go out through `eventBus.dispatchEvent`, and a
listener exception propagates exactly as it does on `main`, whether or
not extras were stashed. Listener-failure isolation is out of scope.
## Context
Iceberg REST table events already expose `customInfo()`, but today that
map is only the request HTTP headers (`IcebergEvent.customInfo()`
returns `icebergRequestContext.httpHeaders()`). An inner dispatcher that
applied a policy or enforced a validation had no way to get that fact
into the terminal Iceberg create/update/load event the outer layer
publishes.
The layer's only options today are to publish a separate event, which
breaks the one-operation-one-event shape audit consumers rely on, or to
drop the fact. #12692 already landed the same hand-off for Gravitino
`TableEventDispatcher` via `RequestContext.setAuditExtras` /
`takeAuditExtras`. Iceberg REST has the same layered shape; this reuses
that stash rather than adding a parallel mechanism.
## Changes
- **New**: `IcebergRequestContext` gains an immutable extras map
(`auditExtras` / `withAuditExtras`). Extras are not part of
`httpHeaders()`.
- **New**: `IcebergEvent` and `IcebergFailureEvent` `customInfo()` is
headers ∪ extras. When extras are empty, `customInfo()` still returns
today's header-only map.
- **New**: `IcebergTableEventDispatcher` takes
`RequestContext.takeAuditExtras()` on create/update/load success and
failure and attaches them to the emitted Iceberg context. The take is
destructive, and `RequestContext.clear()` still removes the stash at
request teardown.
- **Unchanged**: drop/list/rename and every other Iceberg operation.
`EventBus` is untouched. Error propagation and operation results are
untouched.
- **Deferred**: listener-failure isolation. Today Iceberg dispatch lets
a success-path listener exception propagate. Whether to isolate that is
a question about dispatcher semantics that applies equally to events
carrying no custom info.
## Impact and risks
No behavior change for existing deployments and no configuration change.
Audit output changes only when a layer actually stashes facts. In that
case `customInfo()` carries those facts beside headers; the record's
shape, count, and every other field are unchanged. Facts never appear on
`httpHeaders()`.
The threading contract is the same one #12692 already accepted: the
contributor and the dispatcher must run on the same thread. That holds
for the servlet request path.
## Reviewer focus
The attach path is the thing I most want a read on.
`IcebergTableEventDispatcher` takes the stash after the inner call
returns or throws, then copies extras onto a new `IcebergRequestContext`
so the event's `customInfo()` can merge them without mutating the
caller's headers.
Second, the empty-extras identity: `customInfo()` returns the same
headers map it does today when nothing was stashed. If you would rather
always copy, say so.
## Related work
Fixes #12722
Counterpart of #12692 for Iceberg REST table events.
Nevin
Co-authored-by: Cursor Agent <[email protected]>
Co-authored-by: Nevin Zheng <[email protected]>
---
.../dispatcher/IcebergTableEventDispatcher.java | 36 ++-
.../gravitino/listener/api/event/IcebergEvent.java | 2 +-
.../listener/api/event/IcebergFailureEvent.java | 2 +-
.../listener/api/event/IcebergRequestContext.java | 87 +++++-
.../TestIcebergTableEventDispatcher.java | 325 +++++++++++++++++++++
.../api/event/TestIcebergRequestContext.java | 73 +++++
6 files changed, 512 insertions(+), 13 deletions(-)
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableEventDispatcher.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableEventDispatcher.java
index 9a6cb21ee2..7482504625 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableEventDispatcher.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableEventDispatcher.java
@@ -20,6 +20,7 @@
package org.apache.gravitino.iceberg.service.dispatcher;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.iceberg.service.IcebergRESTUtils;
@@ -53,6 +54,7 @@ import
org.apache.gravitino.listener.api.event.IcebergTableExistsPreEvent;
import org.apache.gravitino.listener.api.event.IcebergUpdateTableEvent;
import org.apache.gravitino.listener.api.event.IcebergUpdateTableFailureEvent;
import org.apache.gravitino.listener.api.event.IcebergUpdateTablePreEvent;
+import org.apache.gravitino.utils.RequestContext;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.rest.requests.CreateTableRequest;
@@ -68,6 +70,10 @@ import
org.apache.iceberg.rest.responses.PlanTableScanResponse;
* {@code IcebergTableEventDispatcher} is a decorator for {@link
IcebergTableOperationExecutor} that
* not only delegates table operations to the underlying dispatcher but also
dispatches
* corresponding events to an {@link org.apache.gravitino.listener.EventBus}.
+ *
+ * <p>Create, update, and load attach optional extras stashed on {@link
RequestContext} so an inner
+ * dispatcher can contribute {@code customInfo} to the terminal Iceberg event
without publishing a
+ * sibling event that consumers would have to correlate.
*/
public class IcebergTableEventDispatcher implements
IcebergTableOperationDispatcher {
@@ -104,12 +110,15 @@ public class IcebergTableEventDispatcher implements
IcebergTableOperationDispatc
} catch (Exception e) {
eventBus.dispatchEvent(
new IcebergCreateTableFailureEvent(
- context, nameIdentifier,
transformedCreateEvent.createTableRequest(), e));
+ contextWithAuditExtras(context),
+ nameIdentifier,
+ transformedCreateEvent.createTableRequest(),
+ e));
throw e;
}
eventBus.dispatchEvent(
new IcebergCreateTableEvent(
- context,
+ contextWithAuditExtras(context),
nameIdentifier,
transformedCreateEvent.createTableRequest(),
loadTableResponse));
@@ -137,12 +146,15 @@ public class IcebergTableEventDispatcher implements
IcebergTableOperationDispatc
} catch (Exception e) {
eventBus.dispatchEvent(
new IcebergUpdateTableFailureEvent(
- context, gravitinoNameIdentifier,
transformedUpdateEvent.updateTableRequest(), e));
+ contextWithAuditExtras(context),
+ gravitinoNameIdentifier,
+ transformedUpdateEvent.updateTableRequest(),
+ e));
throw e;
}
eventBus.dispatchEvent(
new IcebergUpdateTableEvent(
- context,
+ contextWithAuditExtras(context),
gravitinoNameIdentifier,
transformedUpdateEvent.updateTableRequest(),
loadTableResponse));
@@ -179,11 +191,14 @@ public class IcebergTableEventDispatcher implements
IcebergTableOperationDispatc
try {
loadTableResponse = icebergTableOperationDispatcher.loadTable(context,
tableIdentifier);
} catch (Exception e) {
- eventBus.dispatchEvent(new IcebergLoadTableFailureEvent(context,
gravitinoNameIdentifier, e));
+ eventBus.dispatchEvent(
+ new IcebergLoadTableFailureEvent(
+ contextWithAuditExtras(context), gravitinoNameIdentifier, e));
throw e;
}
eventBus.dispatchEvent(
- new IcebergLoadTableEvent(context, gravitinoNameIdentifier,
loadTableResponse));
+ new IcebergLoadTableEvent(
+ contextWithAuditExtras(context), gravitinoNameIdentifier,
loadTableResponse));
return loadTableResponse;
}
@@ -310,4 +325,13 @@ public class IcebergTableEventDispatcher implements
IcebergTableOperationDispatc
IcebergRequestContext context, TableIdentifier tableIdentifier) {
return icebergTableOperationDispatcher.getTableMetadataLocation(context,
tableIdentifier);
}
+
+ /**
+ * Takes extras stashed on {@link RequestContext} and returns a context that
carries them. The
+ * take is destructive so a later operation on the same thread cannot see
this request's extras.
+ */
+ private static IcebergRequestContext
contextWithAuditExtras(IcebergRequestContext context) {
+ Map<String, String> extras = RequestContext.takeAuditExtras();
+ return extras.isEmpty() ? context : context.withAuditExtras(extras);
+ }
}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergEvent.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergEvent.java
index 08d7cb1743..044000e831 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergEvent.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergEvent.java
@@ -55,6 +55,6 @@ public abstract class IcebergEvent extends Event {
@Override
public Map<String, String> customInfo() {
- return icebergRequestContext.httpHeaders();
+ return icebergRequestContext.customInfo();
}
}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFailureEvent.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFailureEvent.java
index 3ffc40ff99..403299db12 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFailureEvent.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFailureEvent.java
@@ -55,6 +55,6 @@ public abstract class IcebergFailureEvent extends
FailureEvent {
@Override
public Map<String, String> customInfo() {
- return icebergRequestContext.httpHeaders();
+ return icebergRequestContext.customInfo();
}
}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergRequestContext.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergRequestContext.java
index e10eb41734..d03cbde592 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergRequestContext.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergRequestContext.java
@@ -19,13 +19,20 @@
package org.apache.gravitino.listener.api.event;
+import com.google.common.collect.ImmutableMap;
+import java.util.LinkedHashMap;
import java.util.Map;
import javax.servlet.http.HttpServletRequest;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.iceberg.service.IcebergRESTUtils;
import org.apache.gravitino.utils.PrincipalUtils;
-/** The general request context information for Iceberg REST operations. */
+/**
+ * The general request context information for Iceberg REST operations.
+ *
+ * <p>Optional {@link #auditExtras()} are facts an inner dispatcher wants on
the terminal Iceberg
+ * event. They are not HTTP headers and never appear in {@link #httpHeaders()}.
+ */
public class IcebergRequestContext {
/** Header that opts a drop purge request into asynchronous file cleanup. */
@@ -41,6 +48,7 @@ public class IcebergRequestContext {
private final String remoteHostName;
private final Map<String, String> httpHeaders;
private final boolean requestCredentialVending;
+ private final Map<String, String> auditExtras;
/**
* Constructs a new {@code IcebergRequestContext} instance.
@@ -61,12 +69,31 @@ public class IcebergRequestContext {
*/
public IcebergRequestContext(
HttpServletRequest httpRequest, String catalogName, boolean
requestCredentialVending) {
- this.httpServletRequest = httpRequest;
- this.remoteHostName = resolveClientAddress(httpRequest);
- this.httpHeaders = IcebergRESTUtils.getHttpHeaders(httpRequest);
+ this(
+ httpRequest,
+ catalogName,
+ PrincipalUtils.getCurrentUserName(),
+ resolveClientAddress(httpRequest),
+ IcebergRESTUtils.getHttpHeaders(httpRequest),
+ requestCredentialVending,
+ ImmutableMap.of());
+ }
+
+ private IcebergRequestContext(
+ HttpServletRequest httpServletRequest,
+ String catalogName,
+ String userName,
+ String remoteHostName,
+ Map<String, String> httpHeaders,
+ boolean requestCredentialVending,
+ Map<String, String> auditExtras) {
+ this.httpServletRequest = httpServletRequest;
this.catalogName = catalogName;
- this.userName = PrincipalUtils.getCurrentUserName();
+ this.userName = userName;
+ this.remoteHostName = remoteHostName;
+ this.httpHeaders = httpHeaders;
this.requestCredentialVending = requestCredentialVending;
+ this.auditExtras = auditExtras;
}
private static String resolveClientAddress(HttpServletRequest request) {
@@ -116,6 +143,49 @@ public class IcebergRequestContext {
return httpHeaders;
}
+ /**
+ * Returns the immutable audit extras attached to this context. Empty when
none were attached.
+ *
+ * @return audit extras, never {@code null}
+ */
+ public Map<String, String> auditExtras() {
+ return auditExtras;
+ }
+
+ /**
+ * Returns a copy of this context with the given audit extras. Extras are
not part of {@link
+ * #httpHeaders()}. A {@code null} or empty map yields a context with empty
extras.
+ *
+ * @param extras optional facts for {@code customInfo()}, or {@code null}
for none
+ * @return a new context; this instance is unchanged
+ */
+ public IcebergRequestContext withAuditExtras(Map<String, String> extras) {
+ return new IcebergRequestContext(
+ httpServletRequest,
+ catalogName,
+ userName,
+ remoteHostName,
+ httpHeaders,
+ requestCredentialVending,
+ copyAuditExtras(extras));
+ }
+
+ /**
+ * Returns HTTP headers merged with {@link #auditExtras()}. When extras are
empty, returns {@link
+ * #httpHeaders()} unchanged so existing callers keep today's header-only
map. Extra keys overlay
+ * headers on conflict.
+ *
+ * @return headers, or headers overlaid with extras
+ */
+ public Map<String, String> customInfo() {
+ if (auditExtras.isEmpty()) {
+ return httpHeaders;
+ }
+ Map<String, String> merged = new LinkedHashMap<>(httpHeaders);
+ merged.putAll(auditExtras);
+ return ImmutableMap.copyOf(merged);
+ }
+
/**
* Checks whether this request opted into asynchronous table purge.
*
@@ -154,4 +224,11 @@ public class IcebergRequestContext {
public HttpServletRequest getHttpServletRequest() {
return httpServletRequest;
}
+
+ private static Map<String, String> copyAuditExtras(Map<String, String>
extras) {
+ if (extras == null || extras.isEmpty()) {
+ return ImmutableMap.of();
+ }
+ return ImmutableMap.copyOf(extras);
+ }
}
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableEventDispatcher.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableEventDispatcher.java
new file mode 100644
index 0000000000..4bbc70bb18
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableEventDispatcher.java
@@ -0,0 +1,325 @@
+/*
+ * 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.iceberg.service.dispatcher;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.LinkedList;
+import java.util.Map;
+import javax.servlet.http.HttpServletRequest;
+import org.apache.gravitino.listener.EventBus;
+import org.apache.gravitino.listener.api.EventListenerPlugin;
+import org.apache.gravitino.listener.api.event.Event;
+import org.apache.gravitino.listener.api.event.IcebergCreateTableEvent;
+import org.apache.gravitino.listener.api.event.IcebergCreateTableFailureEvent;
+import org.apache.gravitino.listener.api.event.IcebergEvent;
+import org.apache.gravitino.listener.api.event.IcebergFailureEvent;
+import org.apache.gravitino.listener.api.event.IcebergLoadTableEvent;
+import org.apache.gravitino.listener.api.event.IcebergLoadTableFailureEvent;
+import org.apache.gravitino.listener.api.event.IcebergRequestContext;
+import org.apache.gravitino.listener.api.event.IcebergUpdateTableEvent;
+import org.apache.gravitino.listener.api.event.IcebergUpdateTableFailureEvent;
+import org.apache.gravitino.utils.RequestContext;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.iceberg.types.Types.LongType;
+import org.apache.iceberg.types.Types.NestedField;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Covers the audit-extras path on {@link IcebergTableEventDispatcher}: how
facts stashed on {@link
+ * RequestContext} reach the emitted Iceberg event, and that listener-failure
propagation is
+ * identical with or without extras. Event field mapping itself is covered by
{@code
+ * TestIcebergRequestContext}.
+ */
+public class TestIcebergTableEventDispatcher {
+
+ private static final String METALAKE = "metalake";
+ private static final String CATALOG = "catalog";
+ private static final Namespace NAMESPACE = Namespace.of("ns");
+ private static final TableIdentifier TABLE_ID =
TableIdentifier.of(NAMESPACE, "table");
+ private static final Schema TABLE_SCHEMA =
+ new Schema(NestedField.required(1, "id", LongType.get()));
+ private static final String REQUEST_HEADER = "X-Request-Id";
+ private static final String REQUEST_HEADER_VALUE = "req-1";
+ private static final String EXTRA_KEY = "audit.reason";
+
+ @AfterEach
+ void cleanup() {
+ RequestContext.clear();
+ Thread.interrupted();
+ }
+
+ /**
+ * This change is meant to be purely additive: extras enrich the event, and
nothing about error
+ * propagation moves. A synchronous listener failure therefore has to reach
the caller exactly as
+ * it does today, whether or not extras were stashed. Asserting both halves
in one test is what
+ * makes a future swallow branch visible, since adding one would only break
the extras case.
+ */
+ @Test
+ void testAttachingExtrasDoesNotChangeListenerFailurePropagation() {
+ EventListenerPlugin listener = mock(EventListenerPlugin.class);
+ when(listener.transformPreEvent(any())).thenAnswer(invocation ->
invocation.getArgument(0));
+ doThrow(new RuntimeException("listener
failed")).when(listener).onPostEvent(any(Event.class));
+
+ IcebergTableEventDispatcher dispatcher = dispatcher(listener,
succeedingInner());
+
+ RuntimeException withoutExtras =
+ Assertions.assertThrows(RuntimeException.class, () ->
createTable(dispatcher));
+ Assertions.assertEquals("listener failed", withoutExtras.getMessage());
+
+ RequestContext.setAuditExtras(ImmutableMap.of(EXTRA_KEY,
"policy-applied"));
+ RuntimeException withExtras =
+ Assertions.assertThrows(RuntimeException.class, () ->
createTable(dispatcher));
+ Assertions.assertEquals("listener failed", withExtras.getMessage());
+ }
+
+ /**
+ * The point of routing extras through the request context is to enrich the
event the dispatcher
+ * already emits rather than to publish a second one. Pins both halves:
headers ∪ extras reach
+ * {@code customInfo}, extras stay off {@code httpHeaders}, and exactly one
post event is
+ * produced.
+ */
+ @Test
+ void testCreateWithExtrasMergesHeadersAndDoesNotTouchHttpHeaders() {
+ RecordingListener listener = new RecordingListener();
+ IcebergTableEventDispatcher dispatcher = dispatcher(listener,
succeedingInner());
+ RequestContext.setAuditExtras(ImmutableMap.of(EXTRA_KEY,
"policy-applied"));
+
+ createTable(dispatcher);
+
+ Event event = listener.popPostEvent();
+ Assertions.assertEquals(IcebergCreateTableEvent.class, event.getClass());
+ assertHeadersMergedWithExtras(event, "policy-applied");
+ Assertions.assertTrue(listener.postEvents.isEmpty());
+ }
+
+ @Test
+ void testCreateWithoutExtrasKeepsHeaderOnlyCustomInfo() {
+ RecordingListener listener = new RecordingListener();
+ IcebergTableEventDispatcher dispatcher = dispatcher(listener,
succeedingInner());
+
+ IcebergRequestContext context = requestContext();
+ dispatcher.createTable(context, NAMESPACE, createTableRequest());
+
+ Event event = listener.popPostEvent();
+ Assertions.assertEquals(IcebergCreateTableEvent.class, event.getClass());
+ Assertions.assertSame(context.httpHeaders(), event.customInfo());
+ Assertions.assertFalse(event.customInfo().containsKey(EXTRA_KEY));
+ }
+
+ /**
+ * A contributor that rejected the operation is exactly the case where the
reason matters most, so
+ * extras have to survive the exception path and reach the failure event.
+ */
+ @Test
+ void testCreateFailureEventMergesHeadersAndExtras() {
+ RecordingListener listener = new RecordingListener();
+ IcebergTableOperationDispatcher inner =
mock(IcebergTableOperationDispatcher.class);
+ when(inner.createTable(any(), any(), any())).thenThrow(new
RuntimeException("create failed"));
+ IcebergTableEventDispatcher dispatcher = dispatcher(listener, inner);
+ RequestContext.setAuditExtras(ImmutableMap.of(EXTRA_KEY,
"validation-failed"));
+
+ RuntimeException thrown =
+ Assertions.assertThrows(RuntimeException.class, () ->
createTable(dispatcher));
+ Assertions.assertEquals("create failed", thrown.getMessage());
+
+ Event event = listener.popPostEvent();
+ Assertions.assertEquals(IcebergCreateTableFailureEvent.class,
event.getClass());
+ assertHeadersMergedWithExtras(event, "validation-failed");
+ }
+
+ /**
+ * Each Iceberg table operation reads the stash in its own hand-written
branch, so create passing
+ * does not imply update and load pass. Covers the two remaining operations
and, by stashing a
+ * second fact between them, that consecutive operations on one thread get
their own value.
+ */
+ @Test
+ void testUpdateAndLoadEventsAttachStashedExtras() {
+ RecordingListener listener = new RecordingListener();
+ IcebergTableEventDispatcher dispatcher = dispatcher(listener,
succeedingInner());
+
+ RequestContext.setAuditExtras(ImmutableMap.of(EXTRA_KEY,
"policy-applied"));
+ dispatcher.updateTable(requestContext(), TABLE_ID, updateTableRequest());
+ Event updateEvent = listener.popPostEvent();
+ Assertions.assertEquals(IcebergUpdateTableEvent.class,
updateEvent.getClass());
+ assertHeadersMergedWithExtras(updateEvent, "policy-applied");
+
+ RequestContext.setAuditExtras(ImmutableMap.of(EXTRA_KEY, "cache-miss"));
+ dispatcher.loadTable(requestContext(), TABLE_ID);
+ Event loadEvent = listener.popPostEvent();
+ Assertions.assertEquals(IcebergLoadTableEvent.class, loadEvent.getClass());
+ assertHeadersMergedWithExtras(loadEvent, "cache-miss");
+ }
+
+ @Test
+ void testUpdateAndLoadFailureEventsAttachStashedExtras() {
+ RecordingListener listener = new RecordingListener();
+ IcebergTableOperationDispatcher inner =
mock(IcebergTableOperationDispatcher.class);
+ when(inner.updateTable(any(), any(), any())).thenThrow(new
RuntimeException("update failed"));
+ when(inner.loadTable(any(), any())).thenThrow(new RuntimeException("load
failed"));
+ IcebergTableEventDispatcher dispatcher = dispatcher(listener, inner);
+
+ RequestContext.setAuditExtras(ImmutableMap.of(EXTRA_KEY,
"validation-failed"));
+ RuntimeException updateThrown =
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () -> dispatcher.updateTable(requestContext(), TABLE_ID,
updateTableRequest()));
+ Assertions.assertEquals("update failed", updateThrown.getMessage());
+ Event updateEvent = listener.popPostEvent();
+ Assertions.assertEquals(IcebergUpdateTableFailureEvent.class,
updateEvent.getClass());
+ assertHeadersMergedWithExtras(updateEvent, "validation-failed");
+
+ RequestContext.setAuditExtras(ImmutableMap.of(EXTRA_KEY, "not-found"));
+ RuntimeException loadThrown =
+ Assertions.assertThrows(
+ RuntimeException.class, () ->
dispatcher.loadTable(requestContext(), TABLE_ID));
+ Assertions.assertEquals("load failed", loadThrown.getMessage());
+ Event loadEvent = listener.popPostEvent();
+ Assertions.assertEquals(IcebergLoadTableFailureEvent.class,
loadEvent.getClass());
+ assertHeadersMergedWithExtras(loadEvent, "not-found");
+ }
+
+ /**
+ * Server threads are pooled, so a fact left behind by one request would be
mis-attributed to
+ * whichever request reuses the thread next. Pins that the dispatcher
consumes the stash rather
+ * than reading it, by running a second operation on the same thread and
requiring clean extras.
+ */
+ @Test
+ void testExtrasDoNotLeakIntoTheNextOperationOnTheSameThread() {
+ RecordingListener listener = new RecordingListener();
+ IcebergTableEventDispatcher dispatcher = dispatcher(listener,
succeedingInner());
+ RequestContext.setAuditExtras(ImmutableMap.of(EXTRA_KEY,
"policy-applied"));
+
+ createTable(dispatcher);
+ Assertions.assertEquals("policy-applied",
listener.popPostEvent().customInfo().get(EXTRA_KEY));
+
+ createTable(dispatcher);
+
+ Event second = listener.popPostEvent();
+ Assertions.assertEquals(IcebergCreateTableEvent.class, second.getClass());
+ Assertions.assertFalse(
+ second.customInfo().containsKey(EXTRA_KEY),
+ "Extras must not survive into a later operation");
+ Assertions.assertSame(
+ ((IcebergEvent) second).icebergRequestContext().httpHeaders(),
second.customInfo());
+ }
+
+ private static void assertHeadersMergedWithExtras(Event event, String
extraValue) {
+ Assertions.assertEquals(REQUEST_HEADER_VALUE,
event.customInfo().get(REQUEST_HEADER));
+ Assertions.assertEquals(extraValue, event.customInfo().get(EXTRA_KEY));
+ Map<String, String> headers;
+ if (event instanceof IcebergEvent) {
+ headers = ((IcebergEvent) event).icebergRequestContext().httpHeaders();
+ } else {
+ headers = ((IcebergFailureEvent)
event).icebergRequestContext().httpHeaders();
+ }
+ Assertions.assertFalse(headers.containsKey(EXTRA_KEY));
+ Assertions.assertEquals(REQUEST_HEADER_VALUE, headers.get(REQUEST_HEADER));
+ }
+
+ private static IcebergTableEventDispatcher dispatcher(
+ EventListenerPlugin listener, IcebergTableOperationDispatcher inner) {
+ return new IcebergTableEventDispatcher(
+ inner, new EventBus(Collections.singletonList(listener)), METALAKE);
+ }
+
+ private static IcebergTableOperationDispatcher succeedingInner() {
+ IcebergTableOperationDispatcher inner =
mock(IcebergTableOperationDispatcher.class);
+ LoadTableResponse response = loadTableResponse();
+ when(inner.createTable(any(), any(), any())).thenReturn(response);
+ when(inner.updateTable(any(), any(), any())).thenReturn(response);
+ when(inner.loadTable(any(), any())).thenReturn(response);
+ return inner;
+ }
+
+ private static void createTable(IcebergTableEventDispatcher dispatcher) {
+ dispatcher.createTable(requestContext(), NAMESPACE, createTableRequest());
+ }
+
+ private static IcebergRequestContext requestContext() {
+ return new IcebergRequestContext(
+ requestWithHeader(REQUEST_HEADER, REQUEST_HEADER_VALUE), CATALOG);
+ }
+
+ private static CreateTableRequest createTableRequest() {
+ return
CreateTableRequest.builder().withName(TABLE_ID.name()).withSchema(TABLE_SCHEMA).build();
+ }
+
+ private static UpdateTableRequest updateTableRequest() {
+ return new UpdateTableRequest(Collections.emptyList(),
Collections.emptyList());
+ }
+
+ private static LoadTableResponse loadTableResponse() {
+ TableMetadata metadata =
+ TableMetadata.newTableMetadata(
+ TABLE_SCHEMA,
+ PartitionSpec.unpartitioned(),
+ "file:///tmp/iceberg-audit-extras",
+ ImmutableMap.of());
+ return LoadTableResponse.builder().withTableMetadata(metadata).build();
+ }
+
+ private static HttpServletRequest requestWithHeader(String name, String
value) {
+ HttpServletRequest request = mock(HttpServletRequest.class);
+ Enumeration<String> headerNames =
Collections.enumeration(Collections.singleton(name));
+ when(request.getRemoteHost()).thenReturn("localhost");
+ when(request.getHeaderNames()).thenReturn(headerNames);
+ when(request.getHeader(name)).thenReturn(value);
+ return request;
+ }
+
+ private static final class RecordingListener implements EventListenerPlugin {
+ private final LinkedList<Event> postEvents = new LinkedList<>();
+
+ @Override
+ public void init(Map<String, String> properties) {}
+
+ @Override
+ public void start() {}
+
+ @Override
+ public void stop() {}
+
+ @Override
+ public void onPostEvent(Event event) {
+ postEvents.add(event);
+ }
+
+ Event popPostEvent() {
+ Assertions.assertFalse(postEvents.isEmpty(), "No post events to pop");
+ return postEvents.removeLast();
+ }
+ }
+}
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/listener/api/event/TestIcebergRequestContext.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/listener/api/event/TestIcebergRequestContext.java
index eeff0219d2..95035cf649 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/listener/api/event/TestIcebergRequestContext.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/listener/api/event/TestIcebergRequestContext.java
@@ -22,9 +22,11 @@ package org.apache.gravitino.listener.api.event;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import com.google.common.collect.ImmutableMap;
import java.util.Collections;
import java.util.Enumeration;
import javax.servlet.http.HttpServletRequest;
+import org.apache.gravitino.NameIdentifier;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -67,6 +69,77 @@ class TestIcebergRequestContext {
.asyncPurge());
}
+ /**
+ * Extras are a parallel map, not headers. A later audit formatter that
reads {@code
+ * httpHeaders()} must not see a fact that was never on the wire.
+ */
+ @Test
+ void testExtrasDoNotAppearOnHttpHeaders() {
+ IcebergRequestContext original =
+ new IcebergRequestContext(requestWithHeader("X-Request-Id", "req-1"),
"cat");
+ IcebergRequestContext enriched =
+ original.withAuditExtras(ImmutableMap.of("audit.reason",
"policy-applied"));
+
+ Assertions.assertNotSame(original, enriched);
+ Assertions.assertTrue(original.auditExtras().isEmpty());
+ Assertions.assertEquals("policy-applied",
enriched.auditExtras().get("audit.reason"));
+ Assertions.assertFalse(enriched.httpHeaders().containsKey("audit.reason"));
+ Assertions.assertEquals("req-1",
enriched.httpHeaders().get("X-Request-Id"));
+ }
+
+ /**
+ * Today's {@code customInfo()} is the headers map. Empty extras have to
keep that identity so
+ * existing callers do not observe a copy.
+ */
+ @Test
+ void testCustomInfoIsHeadersWhenExtrasEmpty() {
+ IcebergRequestContext context =
+ new IcebergRequestContext(requestWithHeader("X-Request-Id", "req-1"),
"cat");
+ Assertions.assertSame(context.httpHeaders(), context.customInfo());
+ Assertions.assertSame(
+ context.httpHeaders(),
context.withAuditExtras(ImmutableMap.of()).customInfo());
+ Assertions.assertSame(context.httpHeaders(),
context.withAuditExtras(null).customInfo());
+ }
+
+ @Test
+ void testCustomInfoMergesHeadersAndExtras() {
+ IcebergRequestContext context =
+ new IcebergRequestContext(requestWithHeader("X-Request-Id", "req-1"),
"cat")
+ .withAuditExtras(ImmutableMap.of("audit.reason",
"policy-applied"));
+ Assertions.assertEquals("req-1", context.customInfo().get("X-Request-Id"));
+ Assertions.assertEquals("policy-applied",
context.customInfo().get("audit.reason"));
+ Assertions.assertFalse(context.httpHeaders().containsKey("audit.reason"));
+ }
+
+ /**
+ * Failure events are the case where the reason matters most. Pins that
{@link
+ * IcebergFailureEvent#customInfo()} is headers ∪ extras and that extras
stay off {@code
+ * httpHeaders()}.
+ */
+ @Test
+ void testFailureEventCustomInfoMergesHeadersAndExtras() {
+ IcebergRequestContext context =
+ new IcebergRequestContext(requestWithHeader("X-Request-Id", "req-1"),
"cat")
+ .withAuditExtras(ImmutableMap.of("audit.reason",
"validation-failed"));
+ IcebergLoadTableFailureEvent event =
+ new IcebergLoadTableFailureEvent(
+ context, NameIdentifier.of("ml", "cat", "ns", "t"), new
RuntimeException("boom"));
+
+ Assertions.assertEquals("req-1", event.customInfo().get("X-Request-Id"));
+ Assertions.assertEquals("validation-failed",
event.customInfo().get("audit.reason"));
+
Assertions.assertFalse(event.icebergRequestContext().httpHeaders().containsKey("audit.reason"));
+ }
+
+ @Test
+ void testFailureEventCustomInfoIsHeaderOnlyWhenExtrasEmpty() {
+ IcebergRequestContext context =
+ new IcebergRequestContext(requestWithHeader("X-Request-Id", "req-1"),
"cat");
+ IcebergLoadTableFailureEvent event =
+ new IcebergLoadTableFailureEvent(
+ context, NameIdentifier.of("ml", "cat", "ns", "t"), new
RuntimeException("boom"));
+ Assertions.assertSame(context.httpHeaders(), event.customInfo());
+ }
+
private static HttpServletRequest requestWithoutHeader() {
HttpServletRequest request = mock(HttpServletRequest.class);
when(request.getRemoteHost()).thenReturn("localhost");
